Add support for fluent limit(int) and scroll(OffsetScrollPosition) to Query by Example queries.

Closes #1609

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/repository/support/FetchableFluentQueryByExample.java

@@ -15,25 +15,32 @@
  */
 package org.springframework.data.jdbc.repository.support;
 
+import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.function.Function;
 import java.util.function.UnaryOperator;
-import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 
 import org.springframework.data.domain.Example;
+import org.springframework.data.domain.OffsetScrollPosition;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.ScrollPosition;
 import org.springframework.data.domain.Sort;
+import org.springframework.data.domain.Window;
 import org.springframework.data.jdbc.core.JdbcAggregateOperations;
 import org.springframework.data.relational.core.query.Query;
 import org.springframework.data.relational.repository.query.RelationalExampleMapper;
+import org.springframework.util.Assert;
 
 /**
  * {@link org.springframework.data.repository.query.FluentQuery.FetchableFluentQuery} using {@link Example}.
  *
  * @author Diego Krupitza
+ * @author Mark Paluch
  * @since 3.0
  */
 class FetchableFluentQueryByExample<S, R> extends FluentQuerySupport<S, R> {
@@ -43,13 +50,13 @@ class FetchableFluentQueryByExample<S, R> extends FluentQuerySupport<S, R> {
 
 	FetchableFluentQueryByExample(Example<S> example, Class<R> resultType, RelationalExampleMapper exampleMapper,
 			JdbcAggregateOperations entityOperations) {
-		this(example, Sort.unsorted(), resultType, Collections.emptyList(), exampleMapper, entityOperations);
+		this(example, Sort.unsorted(), 0, resultType, Collections.emptyList(), exampleMapper, entityOperations);
 	}
 
-	FetchableFluentQueryByExample(Example<S> example, Sort sort, Class<R> resultType, List<String> fieldsToInclude,
-			RelationalExampleMapper exampleMapper, JdbcAggregateOperations entityOperations) {
+	FetchableFluentQueryByExample(Example<S> example, Sort sort, int limit, Class<R> resultType,
+			List<String> fieldsToInclude, RelationalExampleMapper exampleMapper, JdbcAggregateOperations entityOperations) {
 
-		super(example, sort, resultType, fieldsToInclude);
+		super(example, sort, limit, resultType, fieldsToInclude);
 
 		this.exampleMapper = exampleMapper;
 		this.entityOperations = entityOperations;
@@ -71,10 +78,40 @@ class FetchableFluentQueryByExample<S, R> extends FluentQuerySupport<S, R> {
 
 	@Override
 	public List<R> all() {
+		return findAll(createQuery().sort(getSort()));
+	}
 
-		return StreamSupport
-				.stream(this.entityOperations.findAll(createQuery().sort(getSort()), getExampleType()).spliterator(), false)
-				.map(item -> this.getConversionFunction().apply(item)).collect(Collectors.toList());
+	private List<R> findAll(Query query) {
+
+		Function<Object, R> conversionFunction = this.getConversionFunction();
+		Iterable<S> raw = this.entityOperations.findAll(query, getExampleType());
+
+		List<R> result = new ArrayList<>(raw instanceof Collections ? ((Collection<?>) raw).size() : 16);
+
+		for (S s : raw) {
+			result.add(conversionFunction.apply(s));
+		}
+
+		return result;
+	}
+
+	@Override
+	public Window<R> scroll(ScrollPosition scrollPosition) {
+
+		Assert.notNull(scrollPosition, "ScrollPosition must not be null");
+
+		if (scrollPosition instanceof OffsetScrollPosition osp) {
+
+			Query query = createQuery().sort(getSort()).offset(osp.getOffset());
+
+			if (getLimit() > 0) {
+				query = query.limit(getLimit());
+			}
+
+			return ScrollDelegate.scroll(query, this::findAll, osp);
+		}
+
+		return super.scroll(scrollPosition);
 	}
 
 	@Override
@@ -114,16 +151,18 @@ class FetchableFluentQueryByExample<S, R> extends FluentQuerySupport<S, R> {
 			query = query.columns(getFieldsToInclude().toArray(new String[0]));
 		}
 
+		query = query.limit(getLimit());
+
 		query = queryCustomizer.apply(query);
 
 		return query;
 	}
 
 	@Override
-	protected <R> FluentQuerySupport<S, R> create(Example<S> example, Sort sort, Class<R> resultType,
+	protected <R> FluentQuerySupport<S, R> create(Example<S> example, Sort sort, int limit, Class<R> resultType,
 			List<String> fieldsToInclude) {
 
-		return new FetchableFluentQueryByExample<>(example, sort, resultType, fieldsToInclude, this.exampleMapper,
+		return new FetchableFluentQueryByExample<>(example, sort, limit, resultType, fieldsToInclude, this.exampleMapper,
 				this.entityOperations);
 	}
 }

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/repository/support/FluentQuerySupport.java

@@ -15,6 +15,11 @@
  */
 package org.springframework.data.jdbc.repository.support;
 
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.function.Function;
+
 import org.springframework.core.convert.support.DefaultConversionService;
 import org.springframework.data.domain.Example;
 import org.springframework.data.domain.Sort;
@@ -22,30 +27,28 @@ import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
 import org.springframework.data.repository.query.FluentQuery;
 import org.springframework.util.Assert;
 
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-import java.util.function.Function;
-
 /**
  * Support class for {@link FluentQuery.FetchableFluentQuery} implementations.
  *
  * @author Diego Krupitza
+ * @author Mark Paluch
  * @since 3.0
  */
 abstract class FluentQuerySupport<S, R> implements FluentQuery.FetchableFluentQuery<R> {
 
 	private final Example<S> example;
 	private final Sort sort;
+	private final int limit;
 	private final Class<R> resultType;
 	private final List<String> fieldsToInclude;
 
 	private final SpelAwareProxyProjectionFactory projectionFactory = new SpelAwareProxyProjectionFactory();
 
-	FluentQuerySupport(Example<S> example, Sort sort, Class<R> resultType, List<String> fieldsToInclude) {
+	FluentQuerySupport(Example<S> example, Sort sort, int limit, Class<R> resultType, List<String> fieldsToInclude) {
 
 		this.example = example;
 		this.sort = sort;
+		this.limit = limit;
 		this.resultType = resultType;
 		this.fieldsToInclude = fieldsToInclude;
 	}
@@ -55,7 +58,15 @@ abstract class FluentQuerySupport<S, R> implements FluentQuery.FetchableFluentQu
 
 		Assert.notNull(sort, "Sort must not be null!");
 
-		return create(example, sort, resultType, fieldsToInclude);
+		return create(example, sort, limit, resultType, fieldsToInclude);
+	}
+
+	@Override
+	public FetchableFluentQuery<R> limit(int limit) {
+
+		Assert.isTrue(limit >= 0, "Limit must not be negative");
+
+		return create(example, sort, limit, resultType, fieldsToInclude);
 	}
 
 	@Override
@@ -63,7 +74,7 @@ abstract class FluentQuerySupport<S, R> implements FluentQuery.FetchableFluentQu
 
 		Assert.notNull(projection, "Projection target type must not be null!");
 
-		return create(example, sort, projection, fieldsToInclude);
+		return create(example, sort, limit, projection, fieldsToInclude);
 	}
 
 	@Override
@@ -71,10 +82,10 @@ abstract class FluentQuerySupport<S, R> implements FluentQuery.FetchableFluentQu
 
 		Assert.notNull(properties, "Projection properties must not be null!");
 
-		return create(example, sort, resultType, new ArrayList<>(properties));
+		return create(example, sort, limit, resultType, new ArrayList<>(properties));
 	}
 
-	protected abstract <R> FluentQuerySupport<S, R> create(Example<S> example, Sort sort, Class<R> resultType,
+	protected abstract <R> FluentQuerySupport<S, R> create(Example<S> example, Sort sort, int limit, Class<R> resultType,
 			List<String> fieldsToInclude);
 
 	Class<S> getExampleType() {
@@ -89,6 +100,10 @@ abstract class FluentQuerySupport<S, R> implements FluentQuery.FetchableFluentQu
 		return sort;
 	}
 
+	int getLimit() {
+		return limit;
+	}
+
 	Class<R> getResultType() {
 		return resultType;
 	}

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/repository/support/ScrollDelegate.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright 2023 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.jdbc.repository.support;
+
+import java.util.List;
+import java.util.function.Function;
+import java.util.function.IntFunction;
+
+import org.springframework.data.domain.OffsetScrollPosition;
+import org.springframework.data.domain.ScrollPosition;
+import org.springframework.data.domain.Window;
+import org.springframework.data.relational.core.query.Query;
+import org.springframework.util.Assert;
+
+/**
+ * Delegate to run {@link ScrollPosition scroll queries} and create result {@link Window}.
+ *
+ * @author Mark Paluch
+ * @since 3.1.4
+ */
+public class ScrollDelegate {
+
+	/**
+	 * Run the {@link Query} and return a scroll {@link Window}.
+	 *
+	 * @param query must not be {@literal null}.
+	 * @param scrollPosition must not be {@literal null}.
+	 * @return the scroll {@link Window}.
+	 */
+	@SuppressWarnings("unchecked")
+	public static <T> Window<T> scroll(Query query, Function<Query, List<T>> queryFunction,
+			ScrollPosition scrollPosition) {
+
+		Assert.notNull(scrollPosition, "ScrollPosition must not be null");
+
+		int limit = query.getLimit();
+		if (limit > 0 && limit != Integer.MAX_VALUE) {
+			query = query.limit(limit + 1);
+		}
+
+		List<T> result = queryFunction.apply(query);
+
+		if (scrollPosition instanceof OffsetScrollPosition offset) {
+			return createWindow(result, limit, OffsetScrollPosition.positionFunction(offset.getOffset()));
+		}
+
+		throw new UnsupportedOperationException("ScrollPosition " + scrollPosition + " not supported");
+	}
+
+	private static <T> Window<T> createWindow(List<T> result, int limit,
+			IntFunction<? extends ScrollPosition> positionFunction) {
+		return Window.from(getFirst(limit, result), positionFunction, hasMoreElements(result, limit));
+	}
+
+	private static boolean hasMoreElements(List<?> result, int limit) {
+		return !result.isEmpty() && result.size() > limit;
+	}
+
+	/**
+	 * Return the first {@code count} items from the list.
+	 *
+	 * @param count the number of first elements to be included in the returned list.
+	 * @param list must not be {@literal null}
+	 * @return the returned sublist if the {@code list} is greater {@code count}.
+	 * @param <T> the element type of the lists.
+	 */
+	public static <T> List<T> getFirst(int count, List<T> list) {
+
+		if (count > 0 && list.size() > count) {
+			return list.subList(0, count);
+		}
+
+		return list;
+	}
+
+}

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/repository/JdbcRepositoryIntegrationTests.java

@@ -53,11 +53,14 @@ import org.springframework.core.io.ClassPathResource;
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.domain.Example;
+import org.springframework.data.domain.ExampleMatcher;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.ScrollPosition;
 import org.springframework.data.domain.Slice;
 import org.springframework.data.domain.Sort;
+import org.springframework.data.domain.Window;
 import org.springframework.data.jdbc.core.mapping.AggregateReference;
 import org.springframework.data.jdbc.repository.query.Modifying;
 import org.springframework.data.jdbc.repository.query.Query;
@@ -82,6 +85,8 @@ import org.springframework.data.repository.query.FluentQuery;
 import org.springframework.data.repository.query.Param;
 import org.springframework.data.repository.query.QueryByExampleExecutor;
 import org.springframework.data.spel.spi.EvaluationContextExtension;
+import org.springframework.data.support.WindowIterator;
+import org.springframework.data.util.Streamable;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.jdbc.core.RowMapper;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
@@ -1073,8 +1078,6 @@ public class JdbcRepositoryIntegrationTests {
 		String searchName = "Diego";
 		Instant now = Instant.now().truncatedTo(ChronoUnit.MILLIS);
 
-		final DummyEntity one = repository.save(createDummyEntity());
-
 		DummyEntity two = createDummyEntity();
 
 		two.setName(searchName);
@@ -1101,6 +1104,42 @@ public class JdbcRepositoryIntegrationTests {
 		assertThat(matches).containsExactly(two, third);
 	}
 
+	@Test // GH-1609
+	void findByScrollPosition() {
+
+		DummyEntity one = new DummyEntity("one");
+		one.setFlag(true);
+
+		DummyEntity two = new DummyEntity("two");
+		two.setFlag(true);
+
+		DummyEntity three = new DummyEntity("three");
+		three.setFlag(true);
+
+		DummyEntity four = new DummyEntity("four");
+		four.setFlag(false);
+
+		repository.saveAll(Arrays.asList(one, two, three, four));
+
+		Example<DummyEntity> example = Example.of(one, ExampleMatcher.matching().withIgnorePaths("name", "idProp"));
+
+		Window<DummyEntity> first = repository.findBy(example, q -> q.limit(2).sortBy(Sort.by("name")))
+				.scroll(ScrollPosition.offset());
+		assertThat(first.map(DummyEntity::getName)).containsExactly("one", "three");
+
+		Window<DummyEntity> second = repository.findBy(example, q -> q.limit(2).sortBy(Sort.by("name")))
+				.scroll(ScrollPosition.offset(2));
+		assertThat(second.map(DummyEntity::getName)).containsExactly("two");
+
+		WindowIterator<DummyEntity> iterator = WindowIterator.of(
+				scrollPosition -> repository.findBy(example, q -> q.limit(2).sortBy(Sort.by("name")).scroll(scrollPosition)))
+				.startingAt(ScrollPosition.offset());
+
+		List<String> result = Streamable.of(() -> iterator).stream().map(DummyEntity::getName).toList();
+
+		assertThat(result).hasSize(3).containsExactly("one", "three", "two");
+	}
+
 	@Test // GH-1192
 	void fetchByExampleFluentCountSimple() {
 
@@ -1777,10 +1816,14 @@ public class JdbcRepositoryIntegrationTests {
 
 		@Override
 		public boolean equals(Object o) {
-			if (this == o) return true;
-			if (o == null || getClass() != o.getClass()) return false;
+			if (this == o)
+				return true;
+			if (o == null || getClass() != o.getClass())
+				return false;
 			DummyEntity that = (DummyEntity) o;
-			return flag == that.flag && Objects.equals(name, that.name) && Objects.equals(pointInTime, that.pointInTime) && Objects.equals(offsetDateTime, that.offsetDateTime) && Objects.equals(idProp, that.idProp) && Objects.equals(ref, that.ref) && direction == that.direction;
+			return flag == that.flag && Objects.equals(name, that.name) && Objects.equals(pointInTime, that.pointInTime)
+					&& Objects.equals(offsetDateTime, that.offsetDateTime) && Objects.equals(idProp, that.idProp)
+					&& Objects.equals(ref, that.ref) && direction == that.direction;
 		}
 
 		@Override

spring-data-r2dbc/src/main/java/org/springframework/data/r2dbc/repository/support/ReactiveFluentQuerySupport.java

@@ -33,12 +33,14 @@ abstract class ReactiveFluentQuerySupport<P, T> implements FluentQuery.ReactiveF
 
 	private final P predicate;
 	private final Sort sort;
+	private final int limit;
 	private final Class<T> resultType;
 	private final List<String> fieldsToInclude;
 
-	ReactiveFluentQuerySupport(P predicate, Sort sort, Class<T> resultType, List<String> fieldsToInclude) {
+	ReactiveFluentQuerySupport(P predicate, Sort sort, int limit, Class<T> resultType, List<String> fieldsToInclude) {
 		this.predicate = predicate;
 		this.sort = sort;
+		this.limit = limit;
 		this.resultType = resultType;
 		this.fieldsToInclude = fieldsToInclude;
 	}
@@ -48,7 +50,15 @@ abstract class ReactiveFluentQuerySupport<P, T> implements FluentQuery.ReactiveF
 
 		Assert.notNull(sort, "Sort must not be null");
 
-		return create(predicate, sort, resultType, fieldsToInclude);
+		return create(predicate, sort, limit, resultType, fieldsToInclude);
+	}
+
+	@Override
+	public ReactiveFluentQuery<T> limit(int limit) {
+
+		Assert.isTrue(limit >= 0, "Limit must not be negative");
+
+		return create(predicate, sort, limit, resultType, fieldsToInclude);
 	}
 
 	@Override
@@ -56,7 +66,7 @@ abstract class ReactiveFluentQuerySupport<P, T> implements FluentQuery.ReactiveF
 
 		Assert.notNull(projection, "Projection target type must not be null");
 
-		return create(predicate, sort, projection, fieldsToInclude);
+		return create(predicate, sort, limit, projection, fieldsToInclude);
 	}
 
 	@Override
@@ -64,10 +74,10 @@ abstract class ReactiveFluentQuerySupport<P, T> implements FluentQuery.ReactiveF
 
 		Assert.notNull(properties, "Projection properties must not be null");
 
-		return create(predicate, sort, resultType, new ArrayList<>(properties));
+		return create(predicate, sort, limit, resultType, new ArrayList<>(properties));
 	}
 
-	protected abstract <R> ReactiveFluentQuerySupport<P, R> create(P predicate, Sort sort, Class<R> resultType,
+	protected abstract <R> ReactiveFluentQuerySupport<P, R> create(P predicate, Sort sort, int limit, Class<R> resultType,
 			List<String> fieldsToInclude);
 
 	P getPredicate() {
@@ -78,6 +88,10 @@ abstract class ReactiveFluentQuerySupport<P, T> implements FluentQuery.ReactiveF
 		return sort;
 	}
 
+	int getLimit() {
+		return limit;
+	}
+
 	Class<T> getResultType() {
 		return resultType;
 	}

spring-data-r2dbc/src/main/java/org/springframework/data/r2dbc/repository/support/ScrollDelegate.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2023 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.r2dbc.repository.support;
+
+import java.util.List;
+import java.util.function.IntFunction;
+
+import org.springframework.data.domain.ScrollPosition;
+import org.springframework.data.domain.Window;
+
+/**
+ * Delegate to handle {@link ScrollPosition scroll queries} and create result {@link Window}.
+ *
+ * @author Mark Paluch
+ * @since 3.1.4
+ */
+public class ScrollDelegate {
+
+	static <T> Window<T> createWindow(List<T> result, int limit, IntFunction<? extends ScrollPosition> positionFunction) {
+		return Window.from(getFirst(limit, result), positionFunction, hasMoreElements(result, limit));
+	}
+
+	private static boolean hasMoreElements(List<?> result, int limit) {
+		return !result.isEmpty() && result.size() > limit;
+	}
+
+	/**
+	 * Return the first {@code count} items from the list.
+	 *
+	 * @param count the number of first elements to be included in the returned list.
+	 * @param list must not be {@literal null}
+	 * @return the returned sublist if the {@code list} is greater {@code count}.
+	 * @param <T> the element type of the lists.
+	 */
+	public static <T> List<T> getFirst(int count, List<T> list) {
+
+		if (count > 0 && list.size() > count) {
+			return list.subList(0, count);
+		}
+
+		return list;
+	}
+
+}

spring-data-r2dbc/src/main/java/org/springframework/data/r2dbc/repository/support/SimpleR2dbcRepository.java

@@ -25,9 +25,12 @@ import java.util.function.UnaryOperator;
 
 import org.reactivestreams.Publisher;
 import org.springframework.data.domain.Example;
+import org.springframework.data.domain.OffsetScrollPosition;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.ScrollPosition;
 import org.springframework.data.domain.Sort;
+import org.springframework.data.domain.Window;
 import org.springframework.data.r2dbc.convert.R2dbcConverter;
 import org.springframework.data.r2dbc.core.R2dbcEntityOperations;
 import org.springframework.data.r2dbc.core.R2dbcEntityTemplate;
@@ -367,17 +370,18 @@ public class SimpleR2dbcRepository<T, ID> implements R2dbcRepository<T, ID> {
 	class ReactiveFluentQueryByExample<S, T> extends ReactiveFluentQuerySupport<Example<S>, T> {
 
 		ReactiveFluentQueryByExample(Example<S> example, Class<T> resultType) {
-			this(example, Sort.unsorted(), resultType, Collections.emptyList());
+			this(example, Sort.unsorted(), 0, resultType, Collections.emptyList());
 		}
 
-		ReactiveFluentQueryByExample(Example<S> example, Sort sort, Class<T> resultType, List<String> fieldsToInclude) {
-			super(example, sort, resultType, fieldsToInclude);
+		ReactiveFluentQueryByExample(Example<S> example, Sort sort, int limit, Class<T> resultType,
+				List<String> fieldsToInclude) {
+			super(example, sort, limit, resultType, fieldsToInclude);
 		}
 
 		@Override
-		protected <R> ReactiveFluentQueryByExample<S, R> create(Example<S> predicate, Sort sort, Class<R> resultType,
-				List<String> fieldsToInclude) {
-			return new ReactiveFluentQueryByExample<>(predicate, sort, resultType, fieldsToInclude);
+		protected <R> ReactiveFluentQueryByExample<S, R> create(Example<S> predicate, Sort sort, int limit,
+				Class<R> resultType, List<String> fieldsToInclude) {
+			return new ReactiveFluentQueryByExample<>(predicate, sort, limit, resultType, fieldsToInclude);
 		}
 
 		@Override
@@ -395,6 +399,34 @@ public class SimpleR2dbcRepository<T, ID> implements R2dbcRepository<T, ID> {
 			return createQuery().all();
 		}
 
+		@Override
+		public Mono<Window<T>> scroll(ScrollPosition scrollPosition) {
+
+			Assert.notNull(scrollPosition, "ScrollPosition must not be null");
+
+			if (scrollPosition instanceof OffsetScrollPosition osp) {
+
+				int limit = getLimit();
+				return createQuery(q -> {
+
+					Query queryToUse = q.offset(osp.getOffset());
+
+					if (limit > 0) {
+						queryToUse = queryToUse.limit(limit + 1);
+					}
+
+					return queryToUse;
+				}).all() //
+						.collectList() //
+						.map(content -> {
+							return ScrollDelegate.createWindow(content, limit,
+									OffsetScrollPosition.positionFunction(osp.getOffset()));
+						});
+			}
+
+			return super.scroll(scrollPosition);
+		}
+
 		@Override
 		public Mono<Page<T>> page(Pageable pageable) {
 

spring-data-r2dbc/src/test/java/org/springframework/data/r2dbc/repository/support/AbstractSimpleR2dbcRepositoryIntegrationTests.java

@@ -15,6 +15,23 @@
  */
 package org.springframework.data.r2dbc.repository.support;
 
+import static org.assertj.core.api.Assertions.*;
+import static org.springframework.data.domain.ExampleMatcher.*;
+import static org.springframework.data.domain.ExampleMatcher.GenericPropertyMatchers.*;
+import static org.springframework.data.domain.ExampleMatcher.StringMatcher.*;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import javax.sql.DataSource;
+
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -25,6 +42,7 @@ import org.springframework.data.annotation.Id;
 import org.springframework.data.annotation.Version;
 import org.springframework.data.domain.Example;
 import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.ScrollPosition;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.r2dbc.convert.MappingR2dbcConverter;
 import org.springframework.data.r2dbc.core.ReactiveDataAccessStrategy;
@@ -37,21 +55,6 @@ import org.springframework.data.relational.repository.support.MappingRelationalE
 import org.springframework.data.repository.query.FluentQuery;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.r2dbc.core.DatabaseClient;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-import reactor.test.StepVerifier;
-
-import javax.sql.DataSource;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-
-import static org.assertj.core.api.Assertions.*;
-import static org.springframework.data.domain.ExampleMatcher.GenericPropertyMatchers.*;
-import static org.springframework.data.domain.ExampleMatcher.StringMatcher.*;
-import static org.springframework.data.domain.ExampleMatcher.*;
 
 /**
  * Abstract integration tests for {@link SimpleR2dbcRepository} to be ran against various databases.
@@ -832,6 +835,36 @@ public abstract class AbstractSimpleR2dbcRepositoryIntegrationTests extends R2db
 				.verifyComplete();
 	}
 
+	@Test // GH-1609
+	void findByScrollPosition() {
+
+		jdbc.execute("INSERT INTO legoset (name, manual) VALUES('FORSCHUNGSSCHIFF', 13)");
+		jdbc.execute("INSERT INTO legoset (name, manual) VALUES('SCHAUFELRADBAGGER', 13)");
+		jdbc.execute("INSERT INTO legoset (name, manual) VALUES('VOLTRON', 13)");
+		jdbc.execute("INSERT INTO legoset (name, manual) VALUES('RALLYEAUTO', 14)");
+
+		LegoSet probe = new LegoSet();
+		probe.setManual(13);
+
+		repository
+				.findBy(Example.of(probe, matching().withIgnorePaths("id")),
+						q -> q.sortBy(Sort.by("name")).limit(2).scroll(ScrollPosition.offset())) //
+				.as(StepVerifier::create) //
+				.consumeNextWith(window -> {
+
+					assertThat(window.map(it -> it.name)).containsOnly("FORSCHUNGSSCHIFF", "SCHAUFELRADBAGGER");
+				}).verifyComplete();
+
+		repository
+				.findBy(Example.of(probe, matching().withIgnorePaths("id")),
+						q -> q.sortBy(Sort.by("name")).limit(2).scroll(ScrollPosition.offset(2))) //
+				.as(StepVerifier::create) //
+				.consumeNextWith(window -> {
+
+					assertThat(window.map(it -> it.name)).containsOnly("VOLTRON");
+				}).verifyComplete();
+	}
+
 	@Test // GH-663
 	void findByShouldApplySortAll() {
 
@@ -981,8 +1014,7 @@ public abstract class AbstractSimpleR2dbcRepositoryIntegrationTests extends R2db
 	@Table("legoset")
 	static class LegoSet {
 
-		@Id
-		int id;
+		@Id int id;
 		String name;
 		Integer manual;
 
@@ -992,8 +1024,7 @@ public abstract class AbstractSimpleR2dbcRepositoryIntegrationTests extends R2db
 			this.manual = manual;
 		}
 
-		public LegoSet() {
-		}
+		public LegoSet() {}
 
 		public int getId() {
 			return this.id;
@@ -1027,8 +1058,7 @@ public abstract class AbstractSimpleR2dbcRepositoryIntegrationTests extends R2db
 	@Table("legoset")
 	static class LegoSetWithNonScalarId {
 
-		@Id
-		Integer id;
+		@Id Integer id;
 		String name;
 		Integer manual;
 		String extra;
@@ -1040,8 +1070,7 @@ public abstract class AbstractSimpleR2dbcRepositoryIntegrationTests extends R2db
 			this.extra = extra;
 		}
 
-		public LegoSetWithNonScalarId() {
-		}
+		public LegoSetWithNonScalarId() {}
 
 		public Integer getId() {
 			return this.id;
@@ -1077,10 +1106,13 @@ public abstract class AbstractSimpleR2dbcRepositoryIntegrationTests extends R2db
 
 		@Override
 		public boolean equals(Object o) {
-			if (this == o) return true;
-			if (o == null || getClass() != o.getClass()) return false;
+			if (this == o)
+				return true;
+			if (o == null || getClass() != o.getClass())
+				return false;
 			LegoSetWithNonScalarId that = (LegoSetWithNonScalarId) o;
-			return Objects.equals(id, that.id) && Objects.equals(name, that.name) && Objects.equals(manual, that.manual) && Objects.equals(extra, that.extra);
+			return Objects.equals(id, that.id) && Objects.equals(name, that.name) && Objects.equals(manual, that.manual)
+					&& Objects.equals(extra, that.extra);
 		}
 
 		@Override
@@ -1092,16 +1124,14 @@ public abstract class AbstractSimpleR2dbcRepositoryIntegrationTests extends R2db
 	@Table("legoset")
 	static class LegoSetVersionable extends LegoSet {
 
-		@Version
-		Integer version;
+		@Version Integer version;
 
 		LegoSetVersionable(int id, String name, Integer manual, Integer version) {
 			super(id, name, manual);
 			this.version = version;
 		}
 
-		public LegoSetVersionable() {
-		}
+		public LegoSetVersionable() {}
 
 		public Integer getVersion() {
 			return this.version;
@@ -1115,16 +1145,14 @@ public abstract class AbstractSimpleR2dbcRepositoryIntegrationTests extends R2db
 	@Table("legoset")
 	static class LegoSetPrimitiveVersionable extends LegoSet {
 
-		@Version
-		int version;
+		@Version int version;
 
 		LegoSetPrimitiveVersionable(int id, String name, Integer manual, int version) {
 			super(id, name, manual);
 			this.version = version;
 		}
 
-		public LegoSetPrimitiveVersionable() {
-		}
+		public LegoSetPrimitiveVersionable() {}
 
 		public int getVersion() {
 			return this.version;