Add support for scrolling using offset- and keyset-based strategies.

We now support scrolling through large query results using ScrollPosition and Window's of data.

See: #4308
Original Pull Request: #4317

Author: mp911de

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/EntityOperations.java

@@ -17,6 +17,7 @@
 
 import java.util.Collection;
 import java.util.Iterator;
+import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Optional;
 
@@ -44,6 +45,7 @@
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.timeseries.Granularity;
 import org.springframework.data.mongodb.core.validation.Validator;
+import org.springframework.data.mongodb.util.BsonUtils;
 import org.springframework.data.projection.EntityProjection;
 import org.springframework.data.projection.EntityProjectionIntrospector;
 import org.springframework.data.projection.ProjectionFactory;
@@ -454,6 +456,9 @@ default boolean isVersionedEntity() {
 		 * @since 2.1.2
 		 */
 		boolean isNew();
+
+		Map<String, Object> extractKeys(Document sortObject);
+
 	}
 
 	/**
@@ -475,7 +480,7 @@ interface AdaptibleEntity<T> extends Entity<T> {
 		T populateIdIfNecessary(@Nullable Object id);
 
 		/**
-		 * Initializes the version property of the of the current entity if available.
+		 * Initializes the version property of the current entity if available.
 		 *
 		 * @return the entity with the version property updated if available.
 		 */
@@ -567,6 +572,19 @@ public T getBean() {
 		public boolean isNew() {
 			return map.get(ID_FIELD) != null;
 		}
+
+		@Override
+		public Map<String, Object> extractKeys(Document sortObject) {
+
+			Map<String, Object> keyset = new LinkedHashMap<>();
+			keyset.put(ID_FIELD, getId());
+
+			for (String key : sortObject.keySet()) {
+				keyset.put(key, BsonUtils.resolveValue(map, key));
+			}
+
+			return keyset;
+		}
 	}
 
 	private static class SimpleMappedEntity<T extends Map<String, Object>> extends UnmappedEntity<T> {
@@ -701,6 +719,22 @@ public T getBean() {
 		public boolean isNew() {
 			return entity.isNew(propertyAccessor.getBean());
 		}
+
+		@Override
+		public Map<String, Object> extractKeys(Document sortObject) {
+
+			Map<String, Object> keyset = new LinkedHashMap<>();
+			keyset.put(entity.getRequiredIdProperty().getName(), getId());
+
+			for (String key : sortObject.keySet()) {
+
+				// TODO: make this work for nested properties
+				MongoPersistentProperty persistentProperty = entity.getRequiredPersistentProperty(key);
+				keyset.put(key, propertyAccessor.getProperty(persistentProperty));
+			}
+
+			return keyset;
+		}
 	}
 
 	private static class AdaptibleMappedEntity<T> extends MappedEntity<T> implements AdaptibleEntity<T> {

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableFindOperation.java

@@ -20,6 +20,8 @@
 import java.util.stream.Stream;
 
 import org.springframework.dao.DataAccessException;
+import org.springframework.data.domain.Scroll;
+import org.springframework.data.domain.ScrollPosition;
 import org.springframework.data.geo.GeoResults;
 import org.springframework.data.mongodb.core.query.CriteriaDefinition;
 import org.springframework.data.mongodb.core.query.NearQuery;
@@ -124,12 +126,24 @@ default Optional<T> first() {
 		Stream<T> stream();
 
 		/**
-		 * Get the number of matching elements.
-		 * <br />
-		 * This method uses an {@link com.mongodb.client.MongoCollection#countDocuments(org.bson.conversions.Bson, com.mongodb.client.model.CountOptions) aggregation
-		 * execution} even for empty {@link Query queries} which may have an impact on performance, but guarantees shard,
-		 * session and transaction compliance. In case an inaccurate count satisfies the applications needs use
-		 * {@link MongoOperations#estimatedCount(String)} for empty queries instead.
+		 * Return a scroll of elements either starting or resuming at
+		 * {@link org.springframework.data.domain.ScrollPosition}.
+		 *
+		 * @param scrollPosition the scroll position.
+		 * @return a scroll of the resulting elements.
+		 * @since 4.1
+		 * @see org.springframework.data.domain.OffsetScrollPosition
+		 * @see org.springframework.data.domain.KeysetScrollPosition
+		 */
+		Scroll<T> scroll(ScrollPosition scrollPosition);
+
+		/**
+		 * Get the number of matching elements. <br />
+		 * This method uses an
+		 * {@link com.mongodb.client.MongoCollection#countDocuments(org.bson.conversions.Bson, com.mongodb.client.model.CountOptions)
+		 * aggregation execution} even for empty {@link Query queries} which may have an impact on performance, but
+		 * guarantees shard, session and transaction compliance. In case an inaccurate count satisfies the applications
+		 * needs use {@link MongoOperations#estimatedCount(String)} for empty queries instead.
 		 *
 		 * @return total number of matching elements.
 		 */

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableFindOperationSupport.java

@@ -21,6 +21,8 @@
 
 import org.bson.Document;
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
+import org.springframework.data.domain.Scroll;
+import org.springframework.data.domain.ScrollPosition;
 import org.springframework.data.mongodb.core.query.NearQuery;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.query.SerializationUtils;
@@ -71,8 +73,8 @@ static class ExecutableFindSupport<T>
 		private final @Nullable String collection;
 		private final Query query;
 
-		ExecutableFindSupport(MongoTemplate template, Class<?> domainType, Class<T> returnType,
-				@Nullable String collection, Query query) {
+		ExecutableFindSupport(MongoTemplate template, Class<?> domainType, Class<T> returnType, @Nullable String collection,
+				Query query) {
 			this.template = template;
 			this.domainType = domainType;
 			this.returnType = returnType;
@@ -138,6 +140,11 @@ public Stream<T> stream() {
 			return doStream();
 		}
 
+		@Override
+		public Scroll<T> scroll(ScrollPosition scrollPosition) {
+			return template.doScroll(query.with(scrollPosition), domainType, returnType, getCollectionName());
+		}
+
 		@Override
 		public TerminatingFindNear<T> near(NearQuery nearQuery) {
 			return () -> template.geoNear(nearQuery, domainType, getCollectionName(), returnType);
@@ -168,8 +175,7 @@ private List<T> doFind(@Nullable CursorPreparer preparer) {
 			Document fieldsObject = query.getFieldsObject();
 
 			return template.doFind(template.createDelegate(query), getCollectionName(), queryObject, fieldsObject, domainType,
-					returnType,
-					getCursorPreparer(query, preparer));
+					returnType, getCursorPreparer(query, preparer));
 		}
 
 		private List<T> doFindDistinct(String field) {

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java

@@ -23,6 +23,8 @@
 import java.util.stream.Stream;
 
 import org.bson.Document;
+import org.springframework.data.domain.KeysetScrollPosition;
+import org.springframework.data.domain.Scroll;
 import org.springframework.data.geo.GeoResults;
 import org.springframework.data.mongodb.core.BulkOperations.BulkMode;
 import org.springframework.data.mongodb.core.aggregation.Aggregation;
@@ -319,7 +321,8 @@ default MongoCollection<Document> createView(String name, Class<?> source, Aggre
 	 * @param options additional settings to apply when creating the view. Can be {@literal null}.
 	 * @since 4.0
 	 */
-	MongoCollection<Document> createView(String name, Class<?> source, AggregationPipeline pipeline, @Nullable ViewOptions options);
+	MongoCollection<Document> createView(String name, Class<?> source, AggregationPipeline pipeline,
+			@Nullable ViewOptions options);
 
 	/**
 	 * Create a view with the provided name. The view content is defined by the {@link AggregationPipeline pipeline} on
@@ -331,7 +334,8 @@ default MongoCollection<Document> createView(String name, Class<?> source, Aggre
 	 * @param options additional settings to apply when creating the view. Can be {@literal null}.
 	 * @since 4.0
 	 */
-	MongoCollection<Document> createView(String name, String source, AggregationPipeline pipeline, @Nullable ViewOptions options);
+	MongoCollection<Document> createView(String name, String source, AggregationPipeline pipeline,
+			@Nullable ViewOptions options);
 
 	/**
 	 * A set of collection names.
@@ -802,6 +806,45 @@ <T> MapReduceResults<T> mapReduce(Query query, String inputCollectionName, Strin
 	 */
 	<T> List<T> find(Query query, Class<T> entityClass, String collectionName);
 
+	/**
+	 * Query for a scroll window of objects of type T from the specified collection. <br />
+	 * Make sure to either set {@link Query#skip(long)} or {@link Query#with(KeysetScrollPosition)} along with
+	 * {@link Query#limit(int)} to limit large query results for efficient scrolling. <br />
+	 * Result objects are converted from the MongoDB native representation using an instance of {@see MongoConverter}.
+	 * Unless configured otherwise, an instance of {@link MappingMongoConverter} will be used. <br />
+	 * If your collection does not contain a homogeneous collection of types, this operation will not be an efficient way
+	 * to map objects since the test for class type is done in the client and not on the server.
+	 *
+	 * @param query the query class that specifies the criteria used to find a record and also an optional fields
+	 *          specification. Must not be {@literal null}.
+	 * @param entityType the parametrized type of the returned list.
+	 * @return the converted scroll.
+	 * @since 4.1
+	 * @see Query#with(org.springframework.data.domain.OffsetScrollPosition)
+	 * @see Query#with(org.springframework.data.domain.KeysetScrollPosition)
+	 */
+	<T> Scroll<T> scroll(Query query, Class<T> entityType);
+
+	/**
+	 * Query for a scroll of objects of type T from the specified collection. <br />
+	 * Make sure to either set {@link Query#skip(long)} or {@link Query#with(KeysetScrollPosition)} along with
+	 * {@link Query#limit(int)} to limit large query results for efficient scrolling. <br />
+	 * Result objects are converted from the MongoDB native representation using an instance of {@see MongoConverter}.
+	 * Unless configured otherwise, an instance of {@link MappingMongoConverter} will be used. <br />
+	 * If your collection does not contain a homogeneous collection of types, this operation will not be an efficient way
+	 * to map objects since the test for class type is done in the client and not on the server.
+	 *
+	 * @param query the query class that specifies the criteria used to find a record and also an optional fields
+	 *          specification. Must not be {@literal null}.
+	 * @param entityType the parametrized type of the returned list.
+	 * @param collectionName name of the collection to retrieve the objects from.
+	 * @return the converted scroll.
+	 * @since 4.1
+	 * @see Query#with(org.springframework.data.domain.OffsetScrollPosition)
+	 * @see Query#with(org.springframework.data.domain.KeysetScrollPosition)
+	 */
+	<T> Scroll<T> scroll(Query query, Class<T> entityType, String collectionName);
+
 	/**
 	 * Returns a document with the given id mapped onto the given class. The collection the query is ran against will be
 	 * derived from the given target class as well.
@@ -1175,7 +1218,7 @@ <S, T> T findAndReplace(Query query, S replacement, FindAndReplaceOptions option
 	 * @param entityClass class that determines the collection to use. Must not be {@literal null}.
 	 * @return the count of matching documents.
 	 * @throws org.springframework.data.mapping.MappingException if the collection name cannot be
-	 * 	  {@link #getCollectionName(Class) derived} from the given type.
+	 *           {@link #getCollectionName(Class) derived} from the given type.
 	 * @see #exactCount(Query, Class)
 	 * @see #estimatedCount(Class)
 	 */

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoTemplate.java

@@ -44,6 +44,8 @@
 import org.springframework.dao.OptimisticLockingFailureException;
 import org.springframework.dao.support.PersistenceExceptionTranslator;
 import org.springframework.data.convert.EntityReader;
+import org.springframework.data.domain.OffsetScrollPosition;
+import org.springframework.data.domain.Scroll;
 import org.springframework.data.geo.Distance;
 import org.springframework.data.geo.GeoResult;
 import org.springframework.data.geo.GeoResults;
@@ -64,6 +66,7 @@
 import org.springframework.data.mongodb.core.QueryOperations.DistinctQueryContext;
 import org.springframework.data.mongodb.core.QueryOperations.QueryContext;
 import org.springframework.data.mongodb.core.QueryOperations.UpdateContext;
+import org.springframework.data.mongodb.core.ScrollUtils.KeySetCursorQuery;
 import org.springframework.data.mongodb.core.aggregation.Aggregation;
 import org.springframework.data.mongodb.core.aggregation.AggregationOperationContext;
 import org.springframework.data.mongodb.core.aggregation.AggregationOptions;
@@ -847,6 +850,48 @@ public <T> List<T> find(Query query, Class<T> entityClass, String collectionName
 				new QueryCursorPreparer(query, entityClass));
 	}
 
+	@Override
+	public <T> Scroll<T> scroll(Query query, Class<T> entityType) {
+
+		Assert.notNull(entityType, "Entity type must not be null");
+
+		return scroll(query, entityType, getCollectionName(entityType));
+	}
+
+	@Override
+	public <T> Scroll<T> scroll(Query query, Class<T> entityType, String collectionName) {
+		return doScroll(query, entityType, entityType, collectionName);
+	}
+
+	<T> Scroll<T> doScroll(Query query, Class<?> sourceClass, Class<T> targetClass, String collectionName) {
+
+		Assert.notNull(query, "Query must not be null");
+		Assert.notNull(collectionName, "CollectionName must not be null");
+		Assert.notNull(sourceClass, "Entity type must not be null");
+		Assert.notNull(targetClass, "Target type must not be null");
+
+		ReadDocumentCallback<T> callback = new ReadDocumentCallback<>(mongoConverter, targetClass, collectionName);
+		int limit = query.isLimited() ? query.getLimit() + 1 : Integer.MAX_VALUE;
+
+		if (query.hasKeyset()) {
+
+			KeySetCursorQuery keysetPaginationQuery = ScrollUtils.createKeysetPaginationQuery(query,
+					operations.getIdPropertyName(sourceClass));
+
+			List<T> result = doFind(collectionName, createDelegate(query), keysetPaginationQuery.query(),
+					keysetPaginationQuery.fields(), sourceClass,
+					new QueryCursorPreparer(query, keysetPaginationQuery.sort(), limit, 0, sourceClass), callback);
+
+			return ScrollUtils.createWindow(query.getSortObject(), query.getLimit(), result, operations);
+		}
+
+		List<T> result = doFind(collectionName, createDelegate(query), query.getQueryObject(), query.getFieldsObject(),
+				sourceClass, new QueryCursorPreparer(query, query.getSortObject(), limit, query.getSkip(), sourceClass),
+				callback);
+
+		return ScrollUtils.createWindow(result, query.getLimit(), OffsetScrollPosition.positionFunction(query.getSkip()));
+	}
+
 	@Nullable
 	@Override
 	public <T> T findById(Object id, Class<T> entityClass) {
@@ -953,7 +998,7 @@ public <T> GeoResults<T> geoNear(NearQuery near, Class<?> domainType, String col
 			optionsBuilder.readPreference(near.getReadPreference());
 		}
 
-		if(near.hasReadConcern()) {
+		if (near.hasReadConcern()) {
 			optionsBuilder.readConcern(near.getReadConcern());
 		}
 
@@ -2837,13 +2882,24 @@ private static MongoConverter getDefaultMongoConverter(MongoDatabaseFactory fact
 		return converter;
 	}
 
-	private Document getMappedSortObject(Query query, Class<?> type) {
+	@Nullable
+	private Document getMappedSortObject(@Nullable Query query, Class<?> type) {
 
-		if (query == null || ObjectUtils.isEmpty(query.getSortObject())) {
+		if (query == null) {
 			return null;
 		}
 
-		return queryMapper.getMappedSort(query.getSortObject(), mappingContext.getPersistentEntity(type));
+		return getMappedSortObject(query.getSortObject(), type);
+	}
+
+	@Nullable
+	private Document getMappedSortObject(Document sortObject, Class<?> type) {
+
+		if (ObjectUtils.isEmpty(sortObject)) {
+			return null;
+		}
+
+		return queryMapper.getMappedSort(sortObject, mappingContext.getPersistentEntity(type));
 	}
 
 	/**
@@ -3199,11 +3255,23 @@ public T doWith(Document document) {
 	class QueryCursorPreparer implements CursorPreparer {
 
 		private final Query query;
+
+		private final Document sortObject;
+
+		private final int limit;
+
+		private final long skip;
 		private final @Nullable Class<?> type;
 
 		QueryCursorPreparer(Query query, @Nullable Class<?> type) {
+			this(query, query.getSortObject(), query.getLimit(), query.getSkip(), type);
+		}
 
+		QueryCursorPreparer(Query query, Document sortObject, int limit, long skip, @Nullable Class<?> type) {
 			this.query = query;
+			this.sortObject = sortObject;
+			this.limit = limit;
+			this.skip = skip;
 			this.type = type;
 		}
 
@@ -3218,20 +3286,20 @@ public FindIterable<Document> prepare(FindIterable<Document> iterable) {
 
 			Meta meta = query.getMeta();
 			HintFunction hintFunction = HintFunction.from(query.getHint());
-			if (query.getSkip() <= 0 && query.getLimit() <= 0 && ObjectUtils.isEmpty(query.getSortObject())
-					&& hintFunction.isEmpty() && !meta.hasValues() && query.getCollation().isEmpty()) {
+			if (skip <= 0 && limit <= 0 && ObjectUtils.isEmpty(sortObject) && hintFunction.isEmpty() && !meta.hasValues()
+					&& query.getCollation().isEmpty()) {
 				return cursorToUse;
 			}
 
 			try {
-				if (query.getSkip() > 0) {
-					cursorToUse = cursorToUse.skip((int) query.getSkip());
+				if (skip > 0) {
+					cursorToUse = cursorToUse.skip((int) skip);
 				}
-				if (query.getLimit() > 0) {
-					cursorToUse = cursorToUse.limit(query.getLimit());
+				if (limit > 0) {
+					cursorToUse = cursorToUse.limit(limit);
 				}
-				if (!ObjectUtils.isEmpty(query.getSortObject())) {
-					Document sort = type != null ? getMappedSortObject(query, type) : query.getSortObject();
+				if (!ObjectUtils.isEmpty(sortObject)) {
+					Document sort = type != null ? getMappedSortObject(sortObject, type) : sortObject;
 					cursorToUse = cursorToUse.sort(sort);
 				}