Add support for search_after.

src/main/java/org/springframework/data/elasticsearch/core/RequestFactory.java

@@ -1127,7 +1127,6 @@ private SearchRequest prepareSearchRequest(Query query, @Nullable Class<?> clazz
 
 		if (query instanceof NativeSearchQuery) {
 			prepareNativeSearch((NativeSearchQuery) query, sourceBuilder);
-
 		}
 
 		if (query.getTrackTotalHits() != null) {
@@ -1147,6 +1146,10 @@ private SearchRequest prepareSearchRequest(Query query, @Nullable Class<?> clazz
 
 		sourceBuilder.explain(query.getExplain());
 
+		if (query.getSearchAfter() != null) {
+			sourceBuilder.searchAfter(query.getSearchAfter().toArray());
+		}
+
 		request.source(sourceBuilder);
 		return request;
 	}
@@ -1229,6 +1232,10 @@ private SearchRequestBuilder prepareSearchRequestBuilder(Query query, Client cli
 
 		searchRequestBuilder.setExplain(query.getExplain());
 
+		if (query.getSearchAfter() != null) {
+			searchRequestBuilder.searchAfter(query.getSearchAfter().toArray());
+		}
+
 		return searchRequestBuilder;
 	}
 

src/main/java/org/springframework/data/elasticsearch/core/query/AbstractQuery.java

@@ -62,6 +62,7 @@ abstract class AbstractQuery implements Query {
 	@Nullable private Duration scrollTime;
 	@Nullable private TimeValue timeout;
 	private boolean explain = false;
+	@Nullable private List<Object> searchAfter;
 
 	@Override
 	@Nullable
@@ -283,4 +284,15 @@ public boolean getExplain() {
 	public void setExplain(boolean explain) {
 		this.explain = explain;
 	}
+
+	@Override
+	public void setSearchAfter(@Nullable List<Object> searchAfter) {
+		this.searchAfter = searchAfter;
+	}
+
+	@Nullable
+	@Override
+	public List<Object> getSearchAfter() {
+		return searchAfter;
+	}
 }

src/main/java/org/springframework/data/elasticsearch/core/query/Query.java

@@ -27,6 +27,7 @@
 import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
+import org.springframework.data.elasticsearch.core.SearchHit;
 import org.springframework.lang.Nullable;
 
 /**
@@ -287,10 +288,26 @@ default boolean hasScrollTime() {
 	TimeValue getTimeout();
 
 	/**
-	 * @return {@literal true} when the query has the eplain parameter set, defaults to {@literal false}
+	 * @return {@literal true} when the query has the explain parameter set, defaults to {@literal false}
 	 * @since 4.2
 	 */
 	default boolean getExplain() {
 		return false;
 	}
+
+	/**
+	 * Sets the setSearchAfter objects for this query.
+	 * 
+	 * @param searchAfter the setSearchAfter objects. These are obtained with {@link SearchHit#getSortValues()} from a
+	 *          search result.
+	 * @since 4.2
+	 */
+	void setSearchAfter(@Nullable List<Object> searchAfter);
+
+	/**
+	 * @return the search_after objects.
+	 * @since 4.2
+	 */
+	@Nullable
+	List<Object> getSearchAfter();
 }

src/test/java/org/springframework/data/elasticsearch/core/paginating/ReactiveSearchAfterIntegrationTests.java

@@ -0,0 +1,100 @@
+/*
+ * Copyright 2021 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.elasticsearch.core.paginating;
+
+import static org.assertj.core.api.Assertions.*;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import reactor.core.publisher.Mono;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.elasticsearch.annotations.Document;
+import org.springframework.data.elasticsearch.annotations.Field;
+import org.springframework.data.elasticsearch.annotations.FieldType;
+import org.springframework.data.elasticsearch.core.ReactiveElasticsearchOperations;
+import org.springframework.data.elasticsearch.core.SearchHit;
+import org.springframework.data.elasticsearch.core.query.Query;
+import org.springframework.data.elasticsearch.junit.jupiter.ReactiveElasticsearchRestTemplateConfiguration;
+import org.springframework.data.elasticsearch.junit.jupiter.SpringIntegrationTest;
+import org.springframework.test.context.ContextConfiguration;
+
+/**
+ * @author Peter-Josef Meisch
+ */
+@SpringIntegrationTest
+@ContextConfiguration(classes = { ReactiveElasticsearchRestTemplateConfiguration.class })
+public class ReactiveSearchAfterIntegrationTests {
+
+	@Autowired private ReactiveElasticsearchOperations operations;
+
+	@Test // #1143
+	@DisplayName("should read pages with search_after")
+	void shouldReadPagesWithSearchAfter() {
+
+		List<Entity> entities = IntStream.rangeClosed(1, 10)
+				.mapToObj(i -> Entity.builder().id((long) i).message("message " + i).build()).collect(Collectors.toList());
+		operations.saveAll(Mono.just(entities), Entity.class).blockLast();
+
+		Query query = Query.findAll();
+		query.setPageable(PageRequest.of(0, 3));
+		query.addSort(Sort.by(Sort.Direction.ASC, "id"));
+
+		List<Object> searchAfter = null;
+		List<Entity> foundEntities = new ArrayList<>();
+
+		int loop = 0;
+		do {
+			query.setSearchAfter(searchAfter);
+			List<SearchHit<Entity>> searchHits = operations.search(query, Entity.class).collectList().block();
+
+			if (searchHits.size() == 0) {
+				break;
+			}
+			foundEntities.addAll(searchHits.stream().map(searchHit -> searchHit.getContent()).collect(Collectors.toList()));
+			searchAfter = searchHits.get((int) (searchHits.size() - 1)).getSortValues();
+
+			if (++loop > 10) {
+				fail("loop not terminating");
+			}
+		} while (true);
+
+		assertThat(foundEntities).containsExactlyElementsOf(entities);
+	}
+
+	@Data
+	@AllArgsConstructor
+	@NoArgsConstructor
+	@Builder
+	@Document(indexName = "test-search-after")
+	private static class Entity {
+		@Id private Long id;
+		@Field(type = FieldType.Text) private String message;
+	}
+
+}

src/test/java/org/springframework/data/elasticsearch/core/paginating/SearchAfterIntegrationTests.java

@@ -0,0 +1,98 @@
+/*
+ * Copyright 2021 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.elasticsearch.core.paginating;
+
+import static org.assertj.core.api.Assertions.*;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.elasticsearch.annotations.Document;
+import org.springframework.data.elasticsearch.annotations.Field;
+import org.springframework.data.elasticsearch.annotations.FieldType;
+import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
+import org.springframework.data.elasticsearch.core.SearchHits;
+import org.springframework.data.elasticsearch.core.query.Query;
+import org.springframework.data.elasticsearch.junit.jupiter.ElasticsearchRestTemplateConfiguration;
+import org.springframework.data.elasticsearch.junit.jupiter.SpringIntegrationTest;
+import org.springframework.test.context.ContextConfiguration;
+
+/**
+ * @author Peter-Josef Meisch
+ */
+@SpringIntegrationTest
+@ContextConfiguration(classes = { ElasticsearchRestTemplateConfiguration.class })
+public class SearchAfterIntegrationTests {
+
+	@Autowired private ElasticsearchOperations operations;
+
+	@Test // #1143
+	@DisplayName("should read pages with search_after")
+	void shouldReadPagesWithSearchAfter() {
+
+		List<Entity> entities = IntStream.rangeClosed(1, 10)
+				.mapToObj(i -> Entity.builder().id((long) i).message("message " + i).build()).collect(Collectors.toList());
+		operations.save(entities);
+
+		Query query = Query.findAll();
+		query.setPageable(PageRequest.of(0, 3));
+		query.addSort(Sort.by(Sort.Direction.ASC, "id"));
+
+		List<Object> searchAfter = null;
+		List<Entity> foundEntities = new ArrayList<>();
+
+		int loop = 0;
+		do {
+			query.setSearchAfter(searchAfter);
+			SearchHits<Entity> searchHits = operations.search(query, Entity.class);
+
+			if (searchHits.getSearchHits().size() == 0) {
+				break;
+			}
+			foundEntities.addAll(searchHits.stream().map(searchHit -> searchHit.getContent()).collect(Collectors.toList()));
+			searchAfter = searchHits.getSearchHit((int) (searchHits.getSearchHits().size() - 1)).getSortValues();
+
+			if (++loop > 10) {
+				fail("loop not terminating");
+			}
+		} while (true);
+
+		assertThat(foundEntities).containsExactlyElementsOf(entities);
+	}
+
+	@Data
+	@AllArgsConstructor
+	@NoArgsConstructor
+	@Builder
+	@Document(indexName = "test-search-after")
+	private static class Entity {
+		@Id private Long id;
+		@Field(type = FieldType.Text) private String message;
+	}
+}

src/test/java/org/springframework/data/elasticsearch/core/paginating/SearchAfterTransportIntegrationTests.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2021 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.elasticsearch.core.paginating;
+
+import org.springframework.data.elasticsearch.junit.jupiter.ElasticsearchTemplateConfiguration;
+import org.springframework.test.context.ContextConfiguration;
+
+/**
+ * @author Peter-Josef Meisch
+ */
+@ContextConfiguration(classes = { ElasticsearchTemplateConfiguration.class })
+public class SearchAfterTransportIntegrationTests extends SearchAfterIntegrationTests {}

src/test/java/org/springframework/data/elasticsearch/core/paginating/package-info.java

@@ -0,0 +1,6 @@
+/**
+ * Test for paginating support with search_after and point_in_time API
+ */
+@org.springframework.lang.NonNullApi
+@org.springframework.lang.NonNullFields
+package org.springframework.data.elasticsearch.core.paginating;