Add support for reverse scrolling.

Closes #4325
Related to: #4308
Original Pull Request: #4317

Author: mp911de

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

@@ -23,6 +23,7 @@
 import org.bson.BsonNull;
 import org.bson.Document;
 import org.springframework.data.domain.KeysetScrollPosition;
+import org.springframework.data.domain.KeysetScrollPosition.Direction;
 import org.springframework.data.domain.ScrollPosition;
 import org.springframework.data.domain.Window;
 import org.springframework.data.mongodb.core.EntityOperations.Entity;
@@ -46,6 +47,10 @@ class ScrollUtils {
 	 */
 	static KeySetScrollQuery createKeysetPaginationQuery(Query query, String idPropertyName) {
 
+		KeysetScrollPosition keyset = query.getKeyset();
+		Map<String, Object> keysetValues = keyset.getKeys();
+		Document queryObject = query.getQueryObject();
+
 		Document sortObject = query.isSorted() ? query.getSortObject() : new Document();
 		sortObject.put(idPropertyName, 1);
 
@@ -57,13 +62,7 @@ static KeySetScrollQuery createKeysetPaginationQuery(Query query, String idPrope
 			}
 		}
 
-		Document queryObject = query.getQueryObject();
-
 		List<Document> or = (List<Document>) queryObject.getOrDefault("$or", new ArrayList<>());
-
-		// TODO: reverse scrolling
-		Map<String, Object> keysetValues = query.getKeyset().getKeys();
-		Document keysetSort = new Document();
 		List<String> sortKeys = new ArrayList<>(sortObject.keySet());
 
 		if (!keysetValues.isEmpty() && !keysetValues.keySet().containsAll(sortKeys)) {
@@ -86,10 +85,11 @@ static KeySetScrollQuery createKeysetPaginationQuery(Query query, String idPrope
 					Object o = keysetValues.get(sortSegment);
 
 					if (j >= i) { // tail segment
-						if(o instanceof BsonNull) {
-							throw new IllegalStateException("Cannot resume from KeysetScrollPosition. Offending key: '%s' is 'null'".formatted(sortSegment));
+						if (o instanceof BsonNull) {
+							throw new IllegalStateException(
+									"Cannot resume from KeysetScrollPosition. Offending key: '%s' is 'null'".formatted(sortSegment));
 						}
-						sortConstraint.put(sortSegment, new Document(sortOrder == 1 ? "$gt" : "$lt", o));
+						sortConstraint.put(sortSegment, new Document(getComparator(sortOrder, keyset.getDirection()), o));
 						break;
 					}
 
@@ -102,16 +102,25 @@ static KeySetScrollQuery createKeysetPaginationQuery(Query query, String idPrope
 			}
 		}
 
-		if (!keysetSort.isEmpty()) {
-			or.add(keysetSort);
-		}
 		if (!or.isEmpty()) {
 			queryObject.put("$or", or);
 		}
 
 		return new KeySetScrollQuery(queryObject, fieldsObject, sortObject);
 	}
 
+	private static String getComparator(int sortOrder, Direction direction) {
+
+		// use gte/lte to include the object at the cursor/keyset so that
+		// we can include it in the result to check whether there is a next object.
+		// It needs to be filtered out later on.
+		if (direction == Direction.Backward) {
+			return sortOrder == 0 ? "$gte" : "$lte";
+		}
+
+		return sortOrder == 1 ? "$gt" : "$lt";
+	}
+
 	static <T> Window<T> createWindow(Document sortObject, int limit, List<T> result, EntityOperations operations) {
 
 		IntFunction<KeysetScrollPosition> positionFunction = value -> {

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoTemplateScrollTests.java

@@ -15,8 +15,8 @@
  */
 package org.springframework.data.mongodb.core;
 
+import static org.assertj.core.api.Assertions.*;
 import static org.springframework.data.mongodb.core.query.Criteria.*;
-import static org.springframework.data.mongodb.test.util.Assertions.*;
 
 import lombok.Data;
 import lombok.NoArgsConstructor;
@@ -37,6 +37,7 @@
 import org.springframework.data.annotation.PersistenceCreator;
 import org.springframework.data.auditing.IsNewAwareAuditingHandler;
 import org.springframework.data.domain.KeysetScrollPosition;
+import org.springframework.data.domain.KeysetScrollPosition.Direction;
 import org.springframework.data.domain.OffsetScrollPosition;
 import org.springframework.data.domain.ScrollPosition;
 import org.springframework.data.domain.Sort;
@@ -118,7 +119,7 @@ void shouldUseKeysetScrollingWithNestedSort() {
 		assertThat(scroll).hasSize(2);
 		assertThat(scroll).containsOnly(john20, john40);
 
-		scroll = template.scroll(q.with(scroll.positionAt(scroll.size()-1)), WithNestedDocument.class);
+		scroll = template.scroll(q.with(scroll.positionAt(scroll.size() - 1)), WithNestedDocument.class);
 
 		assertThat(scroll.hasNext()).isFalse();
 		assertThat(scroll.isLast()).isTrue();
@@ -143,6 +144,19 @@ void shouldErrorOnNullValueForQuery() {
 				new Document("name", "foo"));
 
 		template.insertAll(Arrays.asList(john20, john40, john41, john42, john43, john44));
+	}
+
+	@Test // GH-4308
+	void shouldAllowReverseSort() {
+
+		WithNestedDocument john20 = new WithNestedDocument(null, "John", 120, new WithNestedDocument("John", 20),
+				new Document("name", "bar"));
+		WithNestedDocument john40 = new WithNestedDocument(null, "John", 140, new WithNestedDocument("John", 40),
+				new Document("name", "baz"));
+		WithNestedDocument john41 = new WithNestedDocument(null, "John", 141, new WithNestedDocument("John", 41),
+				new Document("name", "foo"));
+
+		template.insertAll(Arrays.asList(john20, john40, john41));
 
 		Query q = new Query(where("name").regex("J.*")).with(Sort.by("nested.name", "nested.age", "document.name"))
 				.limit(2);
@@ -155,11 +169,22 @@ void shouldErrorOnNullValueForQuery() {
 		assertThat(scroll).hasSize(2);
 		assertThat(scroll).containsOnly(john20, john40);
 
-		ScrollPosition startAfter = scroll.positionAt(scroll.size()-1);
+		scroll = template.scroll(q.with(scroll.positionAt(scroll.size() - 1)), WithNestedDocument.class);
 
-		assertThatExceptionOfType(IllegalStateException.class)
-				.isThrownBy(() -> template.scroll(q.with(startAfter), WithNestedDocument.class))
-				.withMessageContaining("document.name");
+		assertThat(scroll.hasNext()).isFalse();
+		assertThat(scroll.isLast()).isTrue();
+		assertThat(scroll).hasSize(1);
+		assertThat(scroll).containsOnly(john41);
+
+		KeysetScrollPosition scrollPosition = (KeysetScrollPosition) scroll.positionAt(0);
+		KeysetScrollPosition reversePosition = KeysetScrollPosition.of(scrollPosition.getKeys(), Direction.Backward);
+
+		scroll = template.scroll(q.with(reversePosition), WithNestedDocument.class);
+
+		assertThat(scroll.hasNext()).isTrue();
+		assertThat(scroll.isLast()).isFalse();
+		assertThat(scroll).hasSize(2);
+		assertThat(scroll).containsOnly(john20, john40);
 	}
 
 	@ParameterizedTest // GH-4308
@@ -185,15 +210,15 @@ public <T> void shouldApplyCursoringCorrectly(ScrollPosition scrollPosition, Cla
 		assertThat(scroll).hasSize(2);
 		assertThat(scroll).containsOnly(assertionConverter.apply(jane_20), assertionConverter.apply(jane_40));
 
-		scroll = template.scroll(q.with(scroll.positionAt(scroll.size()-1)).limit(3), resultType, "person");
+		scroll = template.scroll(q.with(scroll.positionAt(scroll.size() - 1)).limit(3), resultType, "person");
 
 		assertThat(scroll.hasNext()).isTrue();
 		assertThat(scroll.isLast()).isFalse();
 		assertThat(scroll).hasSize(3);
 		assertThat(scroll).contains(assertionConverter.apply(jane_42), assertionConverter.apply(john20));
 		assertThat(scroll).containsAnyOf(assertionConverter.apply(john40_1), assertionConverter.apply(john40_2));
 
-		scroll = template.scroll(q.with(scroll.positionAt(scroll.size()-1)).limit(1), resultType, "person");
+		scroll = template.scroll(q.with(scroll.positionAt(scroll.size() - 1)).limit(1), resultType, "person");
 
 		assertThat(scroll.hasNext()).isFalse();
 		assertThat(scroll.isLast()).isTrue();