Skip to content

Commit dbf2a16

Browse files
mp911demp911de
committed
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.
1 parent e9eabf9 commit dbf2a16

36 files changed

+1196
-122
lines changed

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

Lines changed: 35 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717

1818
import java.util.Collection;
1919
import java.util.Iterator;
20+
import java.util.LinkedHashMap;
2021
import java.util.Map;
2122
import java.util.Optional;
2223

@@ -44,6 +45,7 @@
4445
import org.springframework.data.mongodb.core.query.Query;
4546
import org.springframework.data.mongodb.core.timeseries.Granularity;
4647
import org.springframework.data.mongodb.core.validation.Validator;
48+
import org.springframework.data.mongodb.util.BsonUtils;
4749
import org.springframework.data.projection.EntityProjection;
4850
import org.springframework.data.projection.EntityProjectionIntrospector;
4951
import org.springframework.data.projection.ProjectionFactory;
@@ -454,6 +456,9 @@ default boolean isVersionedEntity() {
454456
* @since 2.1.2
455457
*/
456458
boolean isNew();
459+
460+
Map<String, Object> extractKeys(Document sortObject);
461+
457462
}
458463

459464
/**
@@ -475,7 +480,7 @@ interface AdaptibleEntity<T> extends Entity<T> {
475480
T populateIdIfNecessary(@Nullable Object id);
476481

477482
/**
478-
* Initializes the version property of the of the current entity if available.
483+
* Initializes the version property of the current entity if available.
479484
*
480485
* @return the entity with the version property updated if available.
481486
*/
@@ -567,6 +572,19 @@ public T getBean() {
567572
public boolean isNew() {
568573
return map.get(ID_FIELD) != null;
569574
}
575+
576+
@Override
577+
public Map<String, Object> extractKeys(Document sortObject) {
578+
579+
Map<String, Object> keyset = new LinkedHashMap<>();
580+
keyset.put(ID_FIELD, getId());
581+
582+
for (String key : sortObject.keySet()) {
583+
keyset.put(key, BsonUtils.resolveValue(map, key));
584+
}
585+
586+
return keyset;
587+
}
570588
}
571589

572590
private static class SimpleMappedEntity<T extends Map<String, Object>> extends UnmappedEntity<T> {
@@ -701,6 +719,22 @@ public T getBean() {
701719
public boolean isNew() {
702720
return entity.isNew(propertyAccessor.getBean());
703721
}
722+
723+
@Override
724+
public Map<String, Object> extractKeys(Document sortObject) {
725+
726+
Map<String, Object> keyset = new LinkedHashMap<>();
727+
keyset.put(entity.getRequiredIdProperty().getName(), getId());
728+
729+
for (String key : sortObject.keySet()) {
730+
731+
// TODO: make this work for nested properties
732+
MongoPersistentProperty persistentProperty = entity.getRequiredPersistentProperty(key);
733+
keyset.put(key, propertyAccessor.getProperty(persistentProperty));
734+
}
735+
736+
return keyset;
737+
}
704738
}
705739

706740
private static class AdaptibleMappedEntity<T> extends MappedEntity<T> implements AdaptibleEntity<T> {

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

Lines changed: 20 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,8 @@
2020
import java.util.stream.Stream;
2121

2222
import org.springframework.dao.DataAccessException;
23+
import org.springframework.data.domain.Scroll;
24+
import org.springframework.data.domain.ScrollPosition;
2325
import org.springframework.data.geo.GeoResults;
2426
import org.springframework.data.mongodb.core.query.CriteriaDefinition;
2527
import org.springframework.data.mongodb.core.query.NearQuery;
@@ -124,12 +126,24 @@ default Optional<T> first() {
124126
Stream<T> stream();
125127

126128
/**
127-
* Get the number of matching elements.
128-
* <br />
129-
* This method uses an {@link com.mongodb.client.MongoCollection#countDocuments(org.bson.conversions.Bson, com.mongodb.client.model.CountOptions) aggregation
130-
* execution} even for empty {@link Query queries} which may have an impact on performance, but guarantees shard,
131-
* session and transaction compliance. In case an inaccurate count satisfies the applications needs use
132-
* {@link MongoOperations#estimatedCount(String)} for empty queries instead.
129+
* Return a scroll of elements either starting or resuming at
130+
* {@link org.springframework.data.domain.ScrollPosition}.
131+
*
132+
* @param scrollPosition the scroll position.
133+
* @return a scroll of the resulting elements.
134+
* @since 4.1
135+
* @see org.springframework.data.domain.OffsetScrollPosition
136+
* @see org.springframework.data.domain.KeysetScrollPosition
137+
*/
138+
Scroll<T> scroll(ScrollPosition scrollPosition);
139+
140+
/**
141+
* Get the number of matching elements. <br />
142+
* This method uses an
143+
* {@link com.mongodb.client.MongoCollection#countDocuments(org.bson.conversions.Bson, com.mongodb.client.model.CountOptions)
144+
* aggregation execution} even for empty {@link Query queries} which may have an impact on performance, but
145+
* guarantees shard, session and transaction compliance. In case an inaccurate count satisfies the applications
146+
* needs use {@link MongoOperations#estimatedCount(String)} for empty queries instead.
133147
*
134148
* @return total number of matching elements.
135149
*/

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

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,8 @@
2121

2222
import org.bson.Document;
2323
import org.springframework.dao.IncorrectResultSizeDataAccessException;
24+
import org.springframework.data.domain.Scroll;
25+
import org.springframework.data.domain.ScrollPosition;
2426
import org.springframework.data.mongodb.core.query.NearQuery;
2527
import org.springframework.data.mongodb.core.query.Query;
2628
import org.springframework.data.mongodb.core.query.SerializationUtils;
@@ -71,8 +73,8 @@ static class ExecutableFindSupport<T>
7173
private final @Nullable String collection;
7274
private final Query query;
7375

74-
ExecutableFindSupport(MongoTemplate template, Class<?> domainType, Class<T> returnType,
75-
@Nullable String collection, Query query) {
76+
ExecutableFindSupport(MongoTemplate template, Class<?> domainType, Class<T> returnType, @Nullable String collection,
77+
Query query) {
7678
this.template = template;
7779
this.domainType = domainType;
7880
this.returnType = returnType;
@@ -138,6 +140,11 @@ public Stream<T> stream() {
138140
return doStream();
139141
}
140142

143+
@Override
144+
public Scroll<T> scroll(ScrollPosition scrollPosition) {
145+
return template.doScroll(query.with(scrollPosition), domainType, returnType, getCollectionName());
146+
}
147+
141148
@Override
142149
public TerminatingFindNear<T> near(NearQuery nearQuery) {
143150
return () -> template.geoNear(nearQuery, domainType, getCollectionName(), returnType);
@@ -168,8 +175,7 @@ private List<T> doFind(@Nullable CursorPreparer preparer) {
168175
Document fieldsObject = query.getFieldsObject();
169176

170177
return template.doFind(template.createDelegate(query), getCollectionName(), queryObject, fieldsObject, domainType,
171-
returnType,
172-
getCursorPreparer(query, preparer));
178+
returnType, getCursorPreparer(query, preparer));
173179
}
174180

175181
private List<T> doFindDistinct(String field) {

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

Lines changed: 46 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,8 @@
2323
import java.util.stream.Stream;
2424

2525
import org.bson.Document;
26+
import org.springframework.data.domain.KeysetScrollPosition;
27+
import org.springframework.data.domain.Scroll;
2628
import org.springframework.data.geo.GeoResults;
2729
import org.springframework.data.mongodb.core.BulkOperations.BulkMode;
2830
import org.springframework.data.mongodb.core.aggregation.Aggregation;
@@ -319,7 +321,8 @@ default MongoCollection<Document> createView(String name, Class<?> source, Aggre
319321
* @param options additional settings to apply when creating the view. Can be {@literal null}.
320322
* @since 4.0
321323
*/
322-
MongoCollection<Document> createView(String name, Class<?> source, AggregationPipeline pipeline, @Nullable ViewOptions options);
324+
MongoCollection<Document> createView(String name, Class<?> source, AggregationPipeline pipeline,
325+
@Nullable ViewOptions options);
323326

324327
/**
325328
* 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
331334
* @param options additional settings to apply when creating the view. Can be {@literal null}.
332335
* @since 4.0
333336
*/
334-
MongoCollection<Document> createView(String name, String source, AggregationPipeline pipeline, @Nullable ViewOptions options);
337+
MongoCollection<Document> createView(String name, String source, AggregationPipeline pipeline,
338+
@Nullable ViewOptions options);
335339

336340
/**
337341
* A set of collection names.
@@ -802,6 +806,45 @@ <T> MapReduceResults<T> mapReduce(Query query, String inputCollectionName, Strin
802806
*/
803807
<T> List<T> find(Query query, Class<T> entityClass, String collectionName);
804808

809+
/**
810+
* Query for a scroll window of objects of type T from the specified collection. <br />
811+
* Make sure to either set {@link Query#skip(long)} or {@link Query#with(KeysetScrollPosition)} along with
812+
* {@link Query#limit(int)} to limit large query results for efficient scrolling. <br />
813+
* Result objects are converted from the MongoDB native representation using an instance of {@see MongoConverter}.
814+
* Unless configured otherwise, an instance of {@link MappingMongoConverter} will be used. <br />
815+
* If your collection does not contain a homogeneous collection of types, this operation will not be an efficient way
816+
* to map objects since the test for class type is done in the client and not on the server.
817+
*
818+
* @param query the query class that specifies the criteria used to find a record and also an optional fields
819+
* specification. Must not be {@literal null}.
820+
* @param entityType the parametrized type of the returned list.
821+
* @return the converted scroll.
822+
* @since 4.1
823+
* @see Query#with(org.springframework.data.domain.OffsetScrollPosition)
824+
* @see Query#with(org.springframework.data.domain.KeysetScrollPosition)
825+
*/
826+
<T> Scroll<T> scroll(Query query, Class<T> entityType);
827+
828+
/**
829+
* Query for a scroll of objects of type T from the specified collection. <br />
830+
* Make sure to either set {@link Query#skip(long)} or {@link Query#with(KeysetScrollPosition)} along with
831+
* {@link Query#limit(int)} to limit large query results for efficient scrolling. <br />
832+
* Result objects are converted from the MongoDB native representation using an instance of {@see MongoConverter}.
833+
* Unless configured otherwise, an instance of {@link MappingMongoConverter} will be used. <br />
834+
* If your collection does not contain a homogeneous collection of types, this operation will not be an efficient way
835+
* to map objects since the test for class type is done in the client and not on the server.
836+
*
837+
* @param query the query class that specifies the criteria used to find a record and also an optional fields
838+
* specification. Must not be {@literal null}.
839+
* @param entityType the parametrized type of the returned list.
840+
* @param collectionName name of the collection to retrieve the objects from.
841+
* @return the converted scroll.
842+
* @since 4.1
843+
* @see Query#with(org.springframework.data.domain.OffsetScrollPosition)
844+
* @see Query#with(org.springframework.data.domain.KeysetScrollPosition)
845+
*/
846+
<T> Scroll<T> scroll(Query query, Class<T> entityType, String collectionName);
847+
805848
/**
806849
* Returns a document with the given id mapped onto the given class. The collection the query is ran against will be
807850
* derived from the given target class as well.
@@ -1175,7 +1218,7 @@ <S, T> T findAndReplace(Query query, S replacement, FindAndReplaceOptions option
11751218
* @param entityClass class that determines the collection to use. Must not be {@literal null}.
11761219
* @return the count of matching documents.
11771220
* @throws org.springframework.data.mapping.MappingException if the collection name cannot be
1178-
* {@link #getCollectionName(Class) derived} from the given type.
1221+
* {@link #getCollectionName(Class) derived} from the given type.
11791222
* @see #exactCount(Query, Class)
11801223
* @see #estimatedCount(Class)
11811224
*/

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

Lines changed: 80 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,8 @@
4444
import org.springframework.dao.OptimisticLockingFailureException;
4545
import org.springframework.dao.support.PersistenceExceptionTranslator;
4646
import org.springframework.data.convert.EntityReader;
47+
import org.springframework.data.domain.OffsetScrollPosition;
48+
import org.springframework.data.domain.Scroll;
4749
import org.springframework.data.geo.Distance;
4850
import org.springframework.data.geo.GeoResult;
4951
import org.springframework.data.geo.GeoResults;
@@ -64,6 +66,7 @@
6466
import org.springframework.data.mongodb.core.QueryOperations.DistinctQueryContext;
6567
import org.springframework.data.mongodb.core.QueryOperations.QueryContext;
6668
import org.springframework.data.mongodb.core.QueryOperations.UpdateContext;
69+
import org.springframework.data.mongodb.core.ScrollUtils.KeySetCursorQuery;
6770
import org.springframework.data.mongodb.core.aggregation.Aggregation;
6871
import org.springframework.data.mongodb.core.aggregation.AggregationOperationContext;
6972
import org.springframework.data.mongodb.core.aggregation.AggregationOptions;
@@ -847,6 +850,48 @@ public <T> List<T> find(Query query, Class<T> entityClass, String collectionName
847850
new QueryCursorPreparer(query, entityClass));
848851
}
849852

853+
@Override
854+
public <T> Scroll<T> scroll(Query query, Class<T> entityType) {
855+
856+
Assert.notNull(entityType, "Entity type must not be null");
857+
858+
return scroll(query, entityType, getCollectionName(entityType));
859+
}
860+
861+
@Override
862+
public <T> Scroll<T> scroll(Query query, Class<T> entityType, String collectionName) {
863+
return doScroll(query, entityType, entityType, collectionName);
864+
}
865+
866+
<T> Scroll<T> doScroll(Query query, Class<?> sourceClass, Class<T> targetClass, String collectionName) {
867+
868+
Assert.notNull(query, "Query must not be null");
869+
Assert.notNull(collectionName, "CollectionName must not be null");
870+
Assert.notNull(sourceClass, "Entity type must not be null");
871+
Assert.notNull(targetClass, "Target type must not be null");
872+
873+
ReadDocumentCallback<T> callback = new ReadDocumentCallback<>(mongoConverter, targetClass, collectionName);
874+
int limit = query.isLimited() ? query.getLimit() + 1 : Integer.MAX_VALUE;
875+
876+
if (query.hasKeyset()) {
877+
878+
KeySetCursorQuery keysetPaginationQuery = ScrollUtils.createKeysetPaginationQuery(query,
879+
operations.getIdPropertyName(sourceClass));
880+
881+
List<T> result = doFind(collectionName, createDelegate(query), keysetPaginationQuery.query(),
882+
keysetPaginationQuery.fields(), sourceClass,
883+
new QueryCursorPreparer(query, keysetPaginationQuery.sort(), limit, 0, sourceClass), callback);
884+
885+
return ScrollUtils.createWindow(query.getSortObject(), query.getLimit(), result, operations);
886+
}
887+
888+
List<T> result = doFind(collectionName, createDelegate(query), query.getQueryObject(), query.getFieldsObject(),
889+
sourceClass, new QueryCursorPreparer(query, query.getSortObject(), limit, query.getSkip(), sourceClass),
890+
callback);
891+
892+
return ScrollUtils.createWindow(result, query.getLimit(), OffsetScrollPosition.positionFunction(query.getSkip()));
893+
}
894+
850895
@Nullable
851896
@Override
852897
public <T> T findById(Object id, Class<T> entityClass) {
@@ -953,7 +998,7 @@ public <T> GeoResults<T> geoNear(NearQuery near, Class<?> domainType, String col
953998
optionsBuilder.readPreference(near.getReadPreference());
954999
}
9551000

956-
if(near.hasReadConcern()) {
1001+
if (near.hasReadConcern()) {
9571002
optionsBuilder.readConcern(near.getReadConcern());
9581003
}
9591004

@@ -2837,13 +2882,24 @@ private static MongoConverter getDefaultMongoConverter(MongoDatabaseFactory fact
28372882
return converter;
28382883
}
28392884

2840-
private Document getMappedSortObject(Query query, Class<?> type) {
2885+
@Nullable
2886+
private Document getMappedSortObject(@Nullable Query query, Class<?> type) {
28412887

2842-
if (query == null || ObjectUtils.isEmpty(query.getSortObject())) {
2888+
if (query == null) {
28432889
return null;
28442890
}
28452891

2846-
return queryMapper.getMappedSort(query.getSortObject(), mappingContext.getPersistentEntity(type));
2892+
return getMappedSortObject(query.getSortObject(), type);
2893+
}
2894+
2895+
@Nullable
2896+
private Document getMappedSortObject(Document sortObject, Class<?> type) {
2897+
2898+
if (ObjectUtils.isEmpty(sortObject)) {
2899+
return null;
2900+
}
2901+
2902+
return queryMapper.getMappedSort(sortObject, mappingContext.getPersistentEntity(type));
28472903
}
28482904

28492905
/**
@@ -3206,11 +3262,23 @@ public T doWith(Document document) {
32063262
class QueryCursorPreparer implements CursorPreparer {
32073263

32083264
private final Query query;
3265+
3266+
private final Document sortObject;
3267+
3268+
private final int limit;
3269+
3270+
private final long skip;
32093271
private final @Nullable Class<?> type;
32103272

32113273
QueryCursorPreparer(Query query, @Nullable Class<?> type) {
3274+
this(query, query.getSortObject(), query.getLimit(), query.getSkip(), type);
3275+
}
32123276

3277+
QueryCursorPreparer(Query query, Document sortObject, int limit, long skip, @Nullable Class<?> type) {
32133278
this.query = query;
3279+
this.sortObject = sortObject;
3280+
this.limit = limit;
3281+
this.skip = skip;
32143282
this.type = type;
32153283
}
32163284

@@ -3225,20 +3293,20 @@ public FindIterable<Document> prepare(FindIterable<Document> iterable) {
32253293

32263294
Meta meta = query.getMeta();
32273295
HintFunction hintFunction = HintFunction.from(query.getHint());
3228-
if (query.getSkip() <= 0 && query.getLimit() <= 0 && ObjectUtils.isEmpty(query.getSortObject())
3229-
&& hintFunction.isEmpty() && !meta.hasValues() && query.getCollation().isEmpty()) {
3296+
if (skip <= 0 && limit <= 0 && ObjectUtils.isEmpty(sortObject) && hintFunction.isEmpty() && !meta.hasValues()
3297+
&& query.getCollation().isEmpty()) {
32303298
return cursorToUse;
32313299
}
32323300

32333301
try {
3234-
if (query.getSkip() > 0) {
3235-
cursorToUse = cursorToUse.skip((int) query.getSkip());
3302+
if (skip > 0) {
3303+
cursorToUse = cursorToUse.skip((int) skip);
32363304
}
3237-
if (query.getLimit() > 0) {
3238-
cursorToUse = cursorToUse.limit(query.getLimit());
3305+
if (limit > 0) {
3306+
cursorToUse = cursorToUse.limit(limit);
32393307
}
3240-
if (!ObjectUtils.isEmpty(query.getSortObject())) {
3241-
Document sort = type != null ? getMappedSortObject(query, type) : query.getSortObject();
3308+
if (!ObjectUtils.isEmpty(sortObject)) {
3309+
Document sort = type != null ? getMappedSortObject(sortObject, type) : sortObject;
32423310
cursorToUse = cursorToUse.sort(sort);
32433311
}
32443312

0 commit comments

Comments
 (0)