Support for Keyset scrolling.

Closes #2878
Original pull request #2885
See https://github.com/spring-projects/spring-data-commons/issues/2151

spring-data-jpa/src/main/java/org/springframework/data/jpa/repository/query/CollectionUtils.java

@@ -0,0 +1,61 @@
+/*
+ * 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.jpa.repository.query;
+
+import java.util.List;
+
+/**
+ * Utility methods to obtain sublists.
+ *
+ * @author Mark Paluch
+ */
+class CollectionUtils {
+
+	/**
+	 * 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>
+	 */
+	public static <T> List<T> getFirst(int count, List<T> list) {
+
+		if (count > 0 && list.size() > count) {
+			return list.subList(0, count);
+		}
+
+		return list;
+	}
+
+	/**
+	 * Return the last {@code count} items from the list.
+	 *
+	 * @param count the number of last 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>
+	 */
+	public static <T> List<T> getLast(int count, List<T> list) {
+
+		if (count > 0 && list.size() > count) {
+			return list.subList(list.size() - (count), list.size());
+		}
+
+		return list;
+	}
+
+}

spring-data-jpa/src/main/java/org/springframework/data/jpa/repository/query/JpaKeysetScrollQueryCreator.java

@@ -0,0 +1,83 @@
+/*
+ * 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.jpa.repository.query;
+
+import jakarta.persistence.criteria.CriteriaBuilder;
+import jakarta.persistence.criteria.CriteriaQuery;
+import jakarta.persistence.criteria.Predicate;
+import jakarta.persistence.criteria.Root;
+
+import java.util.Collection;
+import java.util.LinkedHashSet;
+import java.util.Set;
+
+import org.springframework.data.domain.KeysetScrollPosition;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.jpa.repository.support.JpaEntityInformation;
+import org.springframework.data.repository.query.ReturnedType;
+import org.springframework.data.repository.query.parser.PartTree;
+import org.springframework.lang.Nullable;
+
+/**
+ * Extension to {@link JpaQueryCreator} to create queries considering {@link KeysetScrollPosition keyset scrolling}.
+ *
+ * @author Mark Paluch
+ * @since 3.1
+ */
+class JpaKeysetScrollQueryCreator extends JpaQueryCreator {
+
+	private final JpaEntityInformation<?, ?> entityInformation;
+	private final KeysetScrollPosition scrollPosition;
+
+	public JpaKeysetScrollQueryCreator(PartTree tree, ReturnedType type, CriteriaBuilder builder,
+			ParameterMetadataProvider provider, JpaEntityInformation<?, ?> entityInformation,
+			KeysetScrollPosition scrollPosition) {
+		super(tree, type, builder, provider);
+		this.entityInformation = entityInformation;
+		this.scrollPosition = scrollPosition;
+	}
+
+	@Override
+	protected CriteriaQuery<?> complete(@Nullable Predicate predicate, Sort sort, CriteriaQuery<?> query,
+			CriteriaBuilder builder, Root<?> root) {
+
+		KeysetScrollSpecification<Object> keysetSpec = new KeysetScrollSpecification<>(scrollPosition, sort,
+				entityInformation);
+		Predicate keysetPredicate = keysetSpec.createPredicate(root, builder);
+
+		CriteriaQuery<?> queryToUse = super.complete(predicate, keysetSpec.sort(), query, builder, root);
+
+		if (keysetPredicate != null) {
+			if (queryToUse.getRestriction() != null) {
+				return queryToUse.where(builder.and(queryToUse.getRestriction(), keysetPredicate));
+			}
+			return queryToUse.where(keysetPredicate);
+		}
+
+		return queryToUse;
+	}
+
+	@Override
+	Collection<String> getRequiredSelection(Sort sort, ReturnedType returnedType) {
+
+		Sort sortToUse = KeysetScrollSpecification.createSort(scrollPosition, sort, entityInformation);
+
+		Set<String> selection = new LinkedHashSet<>(returnedType.getInputProperties());
+		sortToUse.forEach(it -> selection.add(it.getProperty()));
+
+		return selection;
+	}
+}

spring-data-jpa/src/main/java/org/springframework/data/jpa/repository/query/JpaQueryCreator.java

@@ -118,7 +118,6 @@ public class JpaQueryCreator extends AbstractQueryCreator<CriteriaQuery<? extend
 
 	@Override
 	protected Predicate create(Part part, Iterator<Object> iterator) {
-
 		return toPredicate(part, root);
 	}
 
@@ -158,9 +157,10 @@ public class JpaQueryCreator extends AbstractQueryCreator<CriteriaQuery<? extend
 
 		if (returnedType.needsCustomConstruction()) {
 
+			Collection<String> requiredSelection = getRequiredSelection(sort, returnedType);
 			List<Selection<?>> selections = new ArrayList<>();
 
-			for (String property : returnedType.getInputProperties()) {
+			for (String property : requiredSelection) {
 
 				PropertyPath path = PropertyPath.from(property, returnedType.getDomainType());
 				selections.add(toExpressionRecursively(root, path, true).alias(property));
@@ -195,6 +195,10 @@ public class JpaQueryCreator extends AbstractQueryCreator<CriteriaQuery<? extend
 		return predicate == null ? select : select.where(predicate);
 	}
 
+	Collection<String> getRequiredSelection(Sort sort, ReturnedType returnedType) {
+		return returnedType.getInputProperties();
+	}
+
 	/**
 	 * Creates a {@link Predicate} from the given {@link Part}.
 	 *

spring-data-jpa/src/main/java/org/springframework/data/jpa/repository/query/JpaQueryExecution.java

@@ -15,23 +15,25 @@
  */
 package org.springframework.data.jpa.repository.query;
 
-import java.lang.reflect.Method;
-import java.util.Collection;
-import java.util.List;
-import java.util.Optional;
-
 import jakarta.persistence.EntityManager;
 import jakarta.persistence.NoResultException;
 import jakarta.persistence.Query;
 import jakarta.persistence.StoredProcedureQuery;
 
+import java.lang.reflect.Method;
+import java.util.Collection;
+import java.util.List;
+import java.util.Optional;
+
 import org.springframework.core.convert.ConversionService;
 import org.springframework.core.convert.support.ConfigurableConversionService;
 import org.springframework.core.convert.support.DefaultConversionService;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.ScrollPosition;
 import org.springframework.data.domain.Slice;
 import org.springframework.data.domain.SliceImpl;
+import org.springframework.data.domain.Sort;
 import org.springframework.data.jpa.provider.PersistenceProvider;
 import org.springframework.data.repository.core.support.SurroundingTransactionDetectorMethodInterceptor;
 import org.springframework.data.support.PageableExecutionUtils;
@@ -128,6 +130,33 @@ public abstract class JpaQueryExecution {
 		}
 	}
 
+	/**
+	 * Executes the query to return a {@link org.springframework.data.domain.Window} of entities.
+	 *
+	 * @author Mark Paluch
+	 * @since 3.1
+	 */
+	static class ScrollExecution extends JpaQueryExecution {
+
+		private final Sort sort;
+		private final ScrollDelegate<?> delegate;
+
+		ScrollExecution(Sort sort, ScrollDelegate<?> delegate) {
+			this.sort = sort;
+			this.delegate = delegate;
+		}
+
+		@Override
+		@SuppressWarnings("unchecked")
+		protected Object doExecute(AbstractJpaQuery query, JpaParametersParameterAccessor accessor) {
+
+			ScrollPosition scrollPosition = accessor.getScrollPosition();
+			Query scrollQuery = query.createQuery(accessor);
+
+			return delegate.scroll(scrollQuery, sort.and(accessor.getSort()), scrollPosition);
+		}
+	}
+
 	/**
 	 * Executes the query to return a {@link Slice} of entities.
 	 *

spring-data-jpa/src/main/java/org/springframework/data/jpa/repository/query/JpaQueryFactory.java

@@ -18,6 +18,7 @@ package org.springframework.data.jpa.repository.query;
 import jakarta.persistence.EntityManager;
 
 import org.springframework.data.jpa.repository.QueryRewriter;
+import org.springframework.data.repository.query.QueryCreationException;
 import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
 import org.springframework.data.repository.query.RepositoryQuery;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
@@ -49,6 +50,10 @@ enum JpaQueryFactory {
 			@Nullable String countQueryString, QueryRewriter queryRewriter,
 			QueryMethodEvaluationContextProvider evaluationContextProvider) {
 
+		if (method.isScrollQuery()) {
+			throw QueryCreationException.create(method, "Scroll queries are not supported using String-based queries");
+		}
+
 		return method.isNativeQuery()
 				? new NativeJpaQuery(method, em, queryString, countQueryString, queryRewriter, evaluationContextProvider,
 						PARSER)
@@ -64,6 +69,11 @@ enum JpaQueryFactory {
 	 * @return
 	 */
 	public StoredProcedureJpaQuery fromProcedureAnnotation(JpaQueryMethod method, EntityManager em) {
+
+		if (method.isScrollQuery()) {
+			throw QueryCreationException.create(method, "Scroll queries are not supported using stored procedures");
+		}
+
 		return new StoredProcedureJpaQuery(method, em);
 	}
 }

spring-data-jpa/src/main/java/org/springframework/data/jpa/repository/query/KeysetScrollDelegate.java

@@ -0,0 +1,197 @@
+/*
+ * 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.jpa.repository.query;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import org.springframework.data.domain.KeysetScrollPosition;
+import org.springframework.data.domain.KeysetScrollPosition.Direction;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.domain.Sort.Order;
+import org.springframework.lang.Nullable;
+
+/**
+ * Delegate for keyset scrolling.
+ *
+ * @author Mark Paluch
+ * @since 3.1
+ */
+public class KeysetScrollDelegate {
+
+	private static final KeysetScrollDelegate forward = new KeysetScrollDelegate();
+	private static final KeysetScrollDelegate reverse = new ReverseKeysetScrollDelegate();
+
+	/**
+	 * Factory method to obtain the right {@link KeysetScrollDelegate}.
+	 *
+	 * @param direction
+	 * @return
+	 */
+	public static KeysetScrollDelegate of(Direction direction) {
+		return direction == Direction.Forward ? forward : reverse;
+	}
+
+	@Nullable
+	public <E, P> P createPredicate(KeysetScrollPosition keyset, Sort sort, QueryStrategy<E, P> strategy) {
+
+		Map<String, Object> keysetValues = keyset.getKeys();
+
+		// first query doesn't come with a keyset
+		if (keysetValues.isEmpty()) {
+			return null;
+		}
+
+		List<P> or = new ArrayList<>();
+		int i = 0;
+
+		// progressive query building to reconstruct a query matching sorting rules
+		for (Order order : sort) {
+
+			if (!keysetValues.containsKey(order.getProperty())) {
+				throw new IllegalStateException(String
+						.format("KeysetScrollPosition does not contain all keyset values. Missing key: %s", order.getProperty()));
+			}
+
+			List<P> sortConstraint = new ArrayList<>();
+
+			int j = 0;
+			for (Order inner : sort) {
+
+				E propertyExpression = strategy.createExpression(inner.getProperty());
+				Object o = keysetValues.get(inner.getProperty());
+
+				if (j >= i) { // tail segment
+
+					sortConstraint.add(strategy.compare(inner, propertyExpression, o));
+					break;
+				}
+
+				sortConstraint.add(strategy.compare(propertyExpression, o));
+				j++;
+			}
+
+			if (!sortConstraint.isEmpty()) {
+				or.add(strategy.and(sortConstraint));
+			}
+
+			i++;
+		}
+
+		if (or.isEmpty()) {
+			return null;
+		}
+
+		return strategy.or(or);
+	}
+
+	protected Sort getSortOrders(Sort sort) {
+		return sort;
+	}
+
+	@SuppressWarnings("unchecked")
+	protected <T> List<T> postProcessResults(List<T> result) {
+		return result;
+	}
+
+	protected <T> List<T> getResultWindow(List<T> list, int limit) {
+		return CollectionUtils.getFirst(limit, list);
+	}
+
+	/**
+	 * Reverse scrolling variant applying {@link Direction#Backward}. In reverse scrolling, we need to flip directions for
+	 * the actual query so that we do not get everything from the top position and apply the limit but rather flip the
+	 * sort direction, apply the limit and then reverse the result to restore the actual sort order.
+	 */
+	private static class ReverseKeysetScrollDelegate extends KeysetScrollDelegate {
+
+		protected Sort getSortOrders(Sort sort) {
+
+			List<Order> orders = new ArrayList<>();
+			for (Order order : sort) {
+				orders.add(new Order(order.isAscending() ? Sort.Direction.DESC : Sort.Direction.ASC, order.getProperty()));
+			}
+
+			return Sort.by(orders);
+		}
+
+		@Override
+		protected <T> List<T> postProcessResults(List<T> result) {
+			Collections.reverse(result);
+			return result;
+		}
+
+		@Override
+		protected <T> List<T> getResultWindow(List<T> list, int limit) {
+			return CollectionUtils.getLast(limit, list);
+		}
+	}
+
+	/**
+	 * Adapter to construct scroll queries.
+	 *
+	 * @param <E> property path expression type.
+	 * @param <P> predicate type.
+	 */
+	public interface QueryStrategy<E, P> {
+
+		/**
+		 * Create an expression object from the given {@code property} path.
+		 *
+		 * @param property must not be {@literal null}.
+		 * @return
+		 */
+		E createExpression(String property);
+
+		/**
+		 * Create a comparison object according to the {@link Order}.
+		 *
+		 * @param order must not be {@literal null}.
+		 * @param propertyExpression must not be {@literal null}.
+		 * @param value
+		 * @return
+		 */
+		P compare(Order order, E propertyExpression, Object value);
+
+		/**
+		 * Create an equals-comparison object.
+		 *
+		 * @param propertyExpression must not be {@literal null}.
+		 * @param value
+		 * @return
+		 */
+		P compare(E propertyExpression, @Nullable Object value);
+
+		/**
+		 * AND-combine the {@code intermediate} predicates.
+		 *
+		 * @param intermediate
+		 * @return
+		 */
+		P and(List<P> intermediate);
+
+		/**
+		 * OR-combine the {@code intermediate} predicates.
+		 *
+		 * @param intermediate
+		 * @return
+		 */
+		P or(List<P> intermediate);
+	}
+
+}

spring-data-jpa/src/main/java/org/springframework/data/jpa/repository/query/KeysetScrollSpecification.java

@@ -0,0 +1,128 @@
+/*
+ * 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.jpa.repository.query;
+
+import jakarta.persistence.criteria.CriteriaBuilder;
+import jakarta.persistence.criteria.CriteriaQuery;
+import jakarta.persistence.criteria.Expression;
+import jakarta.persistence.criteria.From;
+import jakarta.persistence.criteria.Predicate;
+import jakarta.persistence.criteria.Root;
+
+import java.util.List;
+
+import org.springframework.data.domain.KeysetScrollPosition;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.domain.Sort.Order;
+import org.springframework.data.jpa.domain.Specification;
+import org.springframework.data.jpa.repository.query.KeysetScrollDelegate.QueryStrategy;
+import org.springframework.data.jpa.repository.support.JpaEntityInformation;
+import org.springframework.data.mapping.PropertyPath;
+import org.springframework.lang.Nullable;
+
+/**
+ * {@link Specification} to create scroll queries using keyset-scrolling.
+ *
+ * @author Mark Paluch
+ * @author Christoph Strobl
+ * @since 3.1
+ */
+public record KeysetScrollSpecification<T> (KeysetScrollPosition position, Sort sort,
+		JpaEntityInformation<?, ?> entity) implements Specification<T> {
+
+	public KeysetScrollSpecification(KeysetScrollPosition position, Sort sort, JpaEntityInformation<?, ?> entity) {
+
+		this.position = position;
+		this.entity = entity;
+		this.sort = createSort(position, sort, entity);
+	}
+
+	/**
+	 * Create a {@link Sort} object to be used with the actual query.
+	 *
+	 * @param position must not be {@literal null}.
+	 * @param sort must not be {@literal null}.
+	 * @param entity must not be {@literal null}.
+	 * @return
+	 */
+	public static Sort createSort(KeysetScrollPosition position, Sort sort, JpaEntityInformation<?, ?> entity) {
+
+		KeysetScrollDelegate delegate = KeysetScrollDelegate.of(position.getDirection());
+
+		Sort sortToUse;
+		if (entity.hasCompositeId()) {
+			sortToUse = sort.and(Sort.by(entity.getIdAttributeNames().toArray(new String[0])));
+		} else {
+			sortToUse = sort.and(Sort.by(entity.getRequiredIdAttribute().getName()));
+		}
+
+		return delegate.getSortOrders(sortToUse);
+	}
+
+	@Override
+	public Predicate toPredicate(Root<T> root, CriteriaQuery<?> query, CriteriaBuilder criteriaBuilder) {
+		return createPredicate(root, criteriaBuilder);
+	}
+
+	@Nullable
+	public Predicate createPredicate(Root<?> root, CriteriaBuilder criteriaBuilder) {
+
+		KeysetScrollDelegate delegate = KeysetScrollDelegate.of(position.getDirection());
+		return delegate.createPredicate(position, sort, new JpaQueryStrategy(root, criteriaBuilder));
+	}
+
+	@SuppressWarnings("rawtypes")
+	private static class JpaQueryStrategy implements QueryStrategy<Expression<Comparable>, Predicate> {
+
+		private final From<?, ?> from;
+		private final CriteriaBuilder cb;
+
+		public JpaQueryStrategy(From<?, ?> from, CriteriaBuilder cb) {
+
+			this.from = from;
+			this.cb = cb;
+		}
+
+		@Override
+		public Expression<Comparable> createExpression(String property) {
+
+			PropertyPath path = PropertyPath.from(property, from.getJavaType());
+			return QueryUtils.toExpressionRecursively(from, path);
+		}
+
+		@Override
+		public Predicate compare(Order order, Expression<Comparable> propertyExpression, Object value) {
+
+			return order.isAscending() ? cb.greaterThan(propertyExpression, (Comparable) value)
+					: cb.lessThan(propertyExpression, (Comparable) value);
+		}
+
+		@Override
+		public Predicate compare(Expression<Comparable> propertyExpression, @Nullable Object value) {
+			return value == null ? cb.isNull(propertyExpression) : cb.equal(propertyExpression, value);
+		}
+
+		@Override
+		public Predicate and(List<Predicate> intermediate) {
+			return cb.and(intermediate.toArray(new Predicate[0]));
+		}
+
+		@Override
+		public Predicate or(List<Predicate> intermediate) {
+			return cb.or(intermediate.toArray(new Predicate[0]));
+		}
+	}
+}

spring-data-jpa/src/main/java/org/springframework/data/jpa/repository/query/NamedQuery.java

@@ -145,6 +145,10 @@ final class NamedQuery extends AbstractJpaQuery {
 			return null;
 		}
 
+		if (method.isScrollQuery()) {
+			throw QueryCreationException.create(method, "Scroll queries are not supported using String-based queries");
+		}
+
 		try {
 
 			RepositoryQuery query = new NamedQuery(method, em);

spring-data-jpa/src/main/java/org/springframework/data/jpa/repository/query/PartTreeJpaQuery.java

@@ -15,19 +15,25 @@
  */
 package org.springframework.data.jpa.repository.query;
 
-import java.util.List;
-
 import jakarta.persistence.EntityManager;
+import jakarta.persistence.PersistenceUnitUtil;
 import jakarta.persistence.Query;
 import jakarta.persistence.TypedQuery;
 import jakarta.persistence.criteria.CriteriaBuilder;
 import jakarta.persistence.criteria.CriteriaQuery;
 
+import java.util.List;
+
+import org.springframework.data.domain.KeysetScrollPosition;
+import org.springframework.data.domain.OffsetScrollPosition;
+import org.springframework.data.domain.ScrollPosition;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.jpa.repository.query.JpaParameters.JpaParameter;
 import org.springframework.data.jpa.repository.query.JpaQueryExecution.DeleteExecution;
 import org.springframework.data.jpa.repository.query.JpaQueryExecution.ExistsExecution;
+import org.springframework.data.jpa.repository.query.JpaQueryExecution.ScrollExecution;
 import org.springframework.data.jpa.repository.query.ParameterMetadataProvider.ParameterMetadata;
+import org.springframework.data.jpa.repository.support.JpaMetamodelEntityInformation;
 import org.springframework.data.repository.query.ResultProcessor;
 import org.springframework.data.repository.query.ReturnedType;
 import org.springframework.data.repository.query.parser.Part;
@@ -55,6 +61,7 @@ public class PartTreeJpaQuery extends AbstractJpaQuery {
 	private final QueryPreparer countQuery;
 	private final EntityManager em;
 	private final EscapeCharacter escape;
+	private final JpaMetamodelEntityInformation<?, Object> entityInformation;
 
 	/**
 	 * Creates a new {@link PartTreeJpaQuery}.
@@ -79,10 +86,14 @@ public class PartTreeJpaQuery extends AbstractJpaQuery {
 
 		this.em = em;
 		this.escape = escape;
-		Class<?> domainClass = method.getEntityInformation().getJavaType();
 		this.parameters = method.getParameters();
 
-		boolean recreationRequired = parameters.hasDynamicProjection() || parameters.potentiallySortsDynamically();
+		Class<?> domainClass = method.getEntityInformation().getJavaType();
+		PersistenceUnitUtil persistenceUnitUtil = em.getEntityManagerFactory().getPersistenceUnitUtil();
+		this.entityInformation = new JpaMetamodelEntityInformation<>(domainClass, em.getMetamodel(), persistenceUnitUtil);
+
+		boolean recreationRequired = parameters.hasDynamicProjection() || parameters.potentiallySortsDynamically()
+				|| method.isScrollQuery();
 
 		try {
 
@@ -111,7 +122,9 @@ public class PartTreeJpaQuery extends AbstractJpaQuery {
 	@Override
 	protected JpaQueryExecution getExecution() {
 
-		if (this.tree.isDelete()) {
+		if (this.getQueryMethod().isScrollQuery()) {
+			return new ScrollExecution(this.tree.getSort(), new ScrollDelegate<>(entityInformation));
+		} else if (this.tree.isDelete()) {
 			return new DeleteExecution(em);
 		} else if (this.tree.isExistsProjection()) {
 			return new ExistsExecution();
@@ -228,7 +241,11 @@ public class PartTreeJpaQuery extends AbstractJpaQuery {
 
 			TypedQuery<?> query = createQuery(criteriaQuery);
 
-			return restrictMaxResultsIfNecessary(invokeBinding(parameterBinder, query, accessor, this.metadataCache));
+			ScrollPosition scrollPosition = accessor.getParameters().hasScrollPositionParameter()
+					? accessor.getScrollPosition()
+					: null;
+			return restrictMaxResultsIfNecessary(invokeBinding(parameterBinder, query, accessor, this.metadataCache),
+					scrollPosition);
 		}
 
 		/**
@@ -236,10 +253,14 @@ public class PartTreeJpaQuery extends AbstractJpaQuery {
 		 * limited.
 		 */
 		@SuppressWarnings("ConstantConditions")
-		private Query restrictMaxResultsIfNecessary(Query query) {
+		private Query restrictMaxResultsIfNecessary(Query query, @Nullable ScrollPosition scrollPosition) {
 
 			if (tree.isLimiting()) {
 
+				if (scrollPosition instanceof OffsetScrollPosition offset) {
+					query.setFirstResult(Math.toIntExact(offset.getOffset()));
+				}
+
 				if (query.getMaxResults() != Integer.MAX_VALUE) {
 					/*
 					 * In order to return the correct results, we have to adjust the first result offset to be returned if:
@@ -298,6 +319,10 @@ public class PartTreeJpaQuery extends AbstractJpaQuery {
 				returnedType = processor.getReturnedType();
 			}
 
+			if (accessor != null && accessor.getScrollPosition()instanceof KeysetScrollPosition keyset) {
+				return new JpaKeysetScrollQueryCreator(tree, returnedType, builder, provider, entityInformation, keyset);
+			}
+
 			return new JpaQueryCreator(tree, returnedType, builder, provider);
 		}
 

spring-data-jpa/src/main/java/org/springframework/data/jpa/repository/query/ScrollDelegate.java

@@ -0,0 +1,105 @@
+/*
+ * 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.jpa.repository.query;
+
+import jakarta.persistence.Query;
+
+import java.util.List;
+import java.util.Map;
+import java.util.function.IntFunction;
+
+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;
+import org.springframework.data.domain.Sort.Order;
+import org.springframework.data.domain.Window;
+import org.springframework.data.jpa.repository.support.JpaEntityInformation;
+import org.springframework.util.Assert;
+
+/**
+ * Delegate to run {@link ScrollPosition scroll queries} and create result {@link Window}.
+ *
+ * @author Mark Paluch
+ * @since 3.1
+ */
+public class ScrollDelegate<T> {
+
+	private final JpaEntityInformation<T, ?> entity;
+
+	protected ScrollDelegate(JpaEntityInformation<T, ?> entity) {
+		this.entity = entity;
+	}
+
+	/**
+	 * Run the {@link Query} and return a scroll {@link Window}.
+	 *
+	 * @param query must not be {@literal null}.
+	 * @param sort must not be {@literal null}.
+	 * @param scrollPosition must not be {@literal null}.
+	 * @return the scroll {@link Window}.
+	 */
+	@SuppressWarnings("unchecked")
+	public Window<T> scroll(Query query, Sort sort, ScrollPosition scrollPosition) {
+
+		Assert.notNull(scrollPosition, "ScrollPosition must not be null");
+
+		int limit = query.getMaxResults();
+		if (limit > 0 && limit != Integer.MAX_VALUE) {
+			query = query.setMaxResults(limit + 1);
+		}
+
+		List<T> result = query.getResultList();
+
+		if (scrollPosition instanceof KeysetScrollPosition keyset) {
+			return createWindow(sort, limit, keyset.getDirection(), entity, result);
+		}
+
+		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(Sort sort, int limit, Direction direction,
+			JpaEntityInformation<T, ?> entity, List<T> result) {
+
+		KeysetScrollDelegate delegate = KeysetScrollDelegate.of(direction);
+		List<T> resultsToUse = delegate.postProcessResults(result);
+
+		IntFunction<KeysetScrollPosition> positionFunction = value -> {
+
+			T object = result.get(value);
+			Map<String, Object> keys = entity.getKeyset(sort.stream().map(Order::getProperty).toList(), object);
+
+			return KeysetScrollPosition.of(keys);
+		};
+
+		return Window.from(delegate.getResultWindow(resultsToUse, limit), positionFunction, hasMoreElements(result, limit));
+	}
+
+	private static <T> Window<T> createWindow(List<T> result, int limit,
+			IntFunction<? extends ScrollPosition> positionFunction) {
+		return Window.from(CollectionUtils.getFirst(limit, result), positionFunction, hasMoreElements(result, limit));
+	}
+
+	private static boolean hasMoreElements(List<?> result, int limit) {
+		return !result.isEmpty() && result.size() > limit;
+	}
+
+}

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

@@ -1,206 +0,0 @@
-/*
- * Copyright 2021-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.jpa.repository.support;
-
-import jakarta.persistence.EntityManager;
-import jakarta.persistence.TypedQuery;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.List;
-import java.util.function.Function;
-import java.util.stream.Stream;
-
-import org.springframework.dao.IncorrectResultSizeDataAccessException;
-import org.springframework.data.domain.Example;
-import org.springframework.data.domain.Page;
-import org.springframework.data.domain.PageImpl;
-import org.springframework.data.domain.Pageable;
-import org.springframework.data.domain.Sort;
-import org.springframework.data.jpa.repository.query.EscapeCharacter;
-import org.springframework.data.jpa.support.PageableUtils;
-import org.springframework.data.repository.query.FluentQuery.FetchableFluentQuery;
-import org.springframework.data.support.PageableExecutionUtils;
-import org.springframework.util.Assert;
-
-/**
- * Immutable implementation of {@link FetchableFluentQuery} based on Query by {@link Example}. All methods that return a
- * {@link FetchableFluentQuery} will return a new instance, not the original.
- *
- * @param <S> Domain type
- * @param <R> Result type
- * @author Greg Turnquist
- * @author Mark Paluch
- * @author Jens Schauder
- * @author J.R. Onyschak
- * @since 2.6
- */
-class FetchableFluentQueryByExample<S, R> extends FluentQuerySupport<S, R> implements FetchableFluentQuery<R> {
-
-	private final Example<S> example;
-	private final Function<Sort, TypedQuery<S>> finder;
-	private final Function<Example<S>, Long> countOperation;
-	private final Function<Example<S>, Boolean> existsOperation;
-	private final EntityManager entityManager;
-	private final EscapeCharacter escapeCharacter;
-
-	public FetchableFluentQueryByExample(Example<S> example, Function<Sort, TypedQuery<S>> finder,
-			Function<Example<S>, Long> countOperation, Function<Example<S>, Boolean> existsOperation,
-			EntityManager entityManager, EscapeCharacter escapeCharacter) {
-		this(example, example.getProbeType(), (Class<R>) example.getProbeType(), Sort.unsorted(), Collections.emptySet(),
-				finder, countOperation, existsOperation, entityManager, escapeCharacter);
-	}
-
-	private FetchableFluentQueryByExample(Example<S> example, Class<S> entityType, Class<R> returnType, Sort sort,
-			Collection<String> properties, Function<Sort, TypedQuery<S>> finder, Function<Example<S>, Long> countOperation,
-			Function<Example<S>, Boolean> existsOperation, EntityManager entityManager, EscapeCharacter escapeCharacter) {
-
-		super(returnType, sort, properties, entityType);
-		this.example = example;
-		this.finder = finder;
-		this.countOperation = countOperation;
-		this.existsOperation = existsOperation;
-		this.entityManager = entityManager;
-		this.escapeCharacter = escapeCharacter;
-	}
-
-	@Override
-	public FetchableFluentQuery<R> sortBy(Sort sort) {
-
-		Assert.notNull(sort, "Sort must not be null");
-
-		return new FetchableFluentQueryByExample<>(example, entityType, resultType, this.sort.and(sort), properties, finder,
-				countOperation, existsOperation, entityManager, escapeCharacter);
-	}
-
-	@Override
-	public <NR> FetchableFluentQuery<NR> as(Class<NR> resultType) {
-
-		Assert.notNull(resultType, "Projection target type must not be null");
-		if (!resultType.isInterface()) {
-			throw new UnsupportedOperationException("Class-based DTOs are not yet supported.");
-		}
-
-		return new FetchableFluentQueryByExample<>(example, entityType, resultType, sort, properties, finder,
-				countOperation, existsOperation, entityManager, escapeCharacter);
-	}
-
-	@Override
-	public FetchableFluentQuery<R> project(Collection<String> properties) {
-
-		return new FetchableFluentQueryByExample<>(example, entityType, resultType, sort, mergeProperties(properties),
-				finder, countOperation, existsOperation, entityManager, escapeCharacter);
-	}
-
-	@Override
-	public R oneValue() {
-
-		TypedQuery<S> limitedQuery = createSortedAndProjectedQuery();
-		limitedQuery.setMaxResults(2); // Never need more than 2 values
-
-		List<S> results = limitedQuery.getResultList();
-
-		if (results.size() > 1) {
-			throw new IncorrectResultSizeDataAccessException(1);
-		}
-
-		return results.isEmpty() ? null : getConversionFunction().apply(results.get(0));
-	}
-
-	@Override
-	public R firstValue() {
-
-		TypedQuery<S> limitedQuery = createSortedAndProjectedQuery();
-		limitedQuery.setMaxResults(1); // Never need more than 1 value
-
-		List<S> results = limitedQuery.getResultList();
-
-		return results.isEmpty() ? null : getConversionFunction().apply(results.get(0));
-	}
-
-	@Override
-	public List<R> all() {
-
-		List<S> resultList = createSortedAndProjectedQuery().getResultList();
-
-		return convert(resultList);
-	}
-
-	@Override
-	public Page<R> page(Pageable pageable) {
-		return pageable.isUnpaged() ? new PageImpl<>(all()) : readPage(pageable);
-	}
-
-	@Override
-	public Stream<R> stream() {
-
-		return createSortedAndProjectedQuery() //
-				.getResultStream() //
-				.map(getConversionFunction());
-	}
-
-	@Override
-	public long count() {
-		return countOperation.apply(example);
-	}
-
-	@Override
-	public boolean exists() {
-		return existsOperation.apply(example);
-	}
-
-	private Page<R> readPage(Pageable pageable) {
-
-		TypedQuery<S> pagedQuery = createSortedAndProjectedQuery();
-
-		if (pageable.isPaged()) {
-			pagedQuery.setFirstResult(PageableUtils.getOffsetAsInteger(pageable));
-			pagedQuery.setMaxResults(pageable.getPageSize());
-		}
-
-		List<R> paginatedResults = convert(pagedQuery.getResultList());
-
-		return PageableExecutionUtils.getPage(paginatedResults, pageable, () -> countOperation.apply(example));
-	}
-
-	private TypedQuery<S> createSortedAndProjectedQuery() {
-
-		TypedQuery<S> query = finder.apply(sort);
-
-		if (!properties.isEmpty()) {
-			query.setHint(EntityGraphFactory.HINT, EntityGraphFactory.create(entityManager, entityType, properties));
-		}
-
-		return query;
-	}
-
-	private List<R> convert(List<S> resultList) {
-
-		Function<Object, R> conversionFunction = getConversionFunction();
-		List<R> mapped = new ArrayList<>(resultList.size());
-
-		for (S s : resultList) {
-			mapped.add(conversionFunction.apply(s));
-		}
-		return mapped;
-	}
-
-	private Function<Object, R> getConversionFunction() {
-		return getConversionFunction(example.getProbeType(), resultType);
-	}
-
-}

spring-data-jpa/src/main/java/org/springframework/data/jpa/repository/support/FetchableFluentQueryByPredicate.java

@@ -16,6 +16,7 @@
 package org.springframework.data.jpa.repository.support;
 
 import jakarta.persistence.EntityManager;
+import jakarta.persistence.Query;
 
 import java.util.ArrayList;
 import java.util.Collection;
@@ -29,7 +30,10 @@ import org.springframework.dao.IncorrectResultSizeDataAccessException;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.PageImpl;
 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.jpa.repository.query.ScrollDelegate;
 import org.springframework.data.repository.query.FluentQuery.FetchableFluentQuery;
 import org.springframework.data.support.PageableExecutionUtils;
 import org.springframework.util.Assert;
@@ -53,27 +57,31 @@ class FetchableFluentQueryByPredicate<S, R> extends FluentQuerySupport<S, R> imp
 
 	private final Predicate predicate;
 	private final Function<Sort, AbstractJPAQuery<?, ?>> finder;
+
+	private final PredicateScrollDelegate<S> scroll;
 	private final BiFunction<Sort, Pageable, AbstractJPAQuery<?, ?>> pagedFinder;
 	private final Function<Predicate, Long> countOperation;
 	private final Function<Predicate, Boolean> existsOperation;
 	private final EntityManager entityManager;
 
 	public FetchableFluentQueryByPredicate(Predicate predicate, Class<S> entityType,
-			Function<Sort, AbstractJPAQuery<?, ?>> finder, BiFunction<Sort, Pageable, AbstractJPAQuery<?, ?>> pagedFinder,
-			Function<Predicate, Long> countOperation, Function<Predicate, Boolean> existsOperation,
-			EntityManager entityManager) {
-		this(predicate, entityType, (Class<R>) entityType, Sort.unsorted(), Collections.emptySet(), finder, pagedFinder,
-				countOperation, existsOperation, entityManager);
+			Function<Sort, AbstractJPAQuery<?, ?>> finder, PredicateScrollDelegate<S> scroll,
+			BiFunction<Sort, Pageable, AbstractJPAQuery<?, ?>> pagedFinder, Function<Predicate, Long> countOperation,
+			Function<Predicate, Boolean> existsOperation, EntityManager entityManager) {
+		this(predicate, entityType, (Class<R>) entityType, Sort.unsorted(), 0, Collections.emptySet(), finder, scroll,
+				pagedFinder, countOperation, existsOperation, entityManager);
 	}
 
 	private FetchableFluentQueryByPredicate(Predicate predicate, Class<S> entityType, Class<R> resultType, Sort sort,
-			Collection<String> properties, Function<Sort, AbstractJPAQuery<?, ?>> finder,
-			BiFunction<Sort, Pageable, AbstractJPAQuery<?, ?>> pagedFinder, Function<Predicate, Long> countOperation,
-			Function<Predicate, Boolean> existsOperation, EntityManager entityManager) {
+			int limit, Collection<String> properties, Function<Sort, AbstractJPAQuery<?, ?>> finder,
+			PredicateScrollDelegate<S> scroll, BiFunction<Sort, Pageable, AbstractJPAQuery<?, ?>> pagedFinder,
+			Function<Predicate, Long> countOperation, Function<Predicate, Boolean> existsOperation,
+			EntityManager entityManager) {
 
-		super(resultType, sort, properties, entityType);
+		super(resultType, sort, limit, properties, entityType);
 		this.predicate = predicate;
 		this.finder = finder;
+		this.scroll = scroll;
 		this.pagedFinder = pagedFinder;
 		this.countOperation = countOperation;
 		this.existsOperation = existsOperation;
@@ -85,8 +93,17 @@ class FetchableFluentQueryByPredicate<S, R> extends FluentQuerySupport<S, R> imp
 
 		Assert.notNull(sort, "Sort must not be null");
 
-		return new FetchableFluentQueryByPredicate<>(predicate, entityType, resultType, this.sort.and(sort), properties,
-				finder, pagedFinder, countOperation, existsOperation, entityManager);
+		return new FetchableFluentQueryByPredicate<>(predicate, entityType, resultType, this.sort.and(sort), limit,
+				properties, finder, scroll, pagedFinder, countOperation, existsOperation, entityManager);
+	}
+
+	@Override
+	public FetchableFluentQuery<R> limit(int limit) {
+
+		Assert.isTrue(limit >= 0, "Limit must not be negative");
+
+		return new FetchableFluentQueryByPredicate<>(predicate, entityType, resultType, sort, limit, properties, finder,
+				scroll, pagedFinder, countOperation, existsOperation, entityManager);
 	}
 
 	@Override
@@ -98,15 +115,15 @@ class FetchableFluentQueryByPredicate<S, R> extends FluentQuerySupport<S, R> imp
 			throw new UnsupportedOperationException("Class-based DTOs are not yet supported.");
 		}
 
-		return new FetchableFluentQueryByPredicate<>(predicate, entityType, resultType, sort, properties, finder,
-				pagedFinder, countOperation, existsOperation, entityManager);
+		return new FetchableFluentQueryByPredicate<>(predicate, entityType, resultType, sort, limit, properties, finder,
+				scroll, pagedFinder, countOperation, existsOperation, entityManager);
 	}
 
 	@Override
 	public FetchableFluentQuery<R> project(Collection<String> properties) {
 
-		return new FetchableFluentQueryByPredicate<>(predicate, entityType, resultType, sort, mergeProperties(properties),
-				finder, pagedFinder, countOperation, existsOperation, entityManager);
+		return new FetchableFluentQueryByPredicate<>(predicate, entityType, resultType, sort, limit,
+				mergeProperties(properties), finder, scroll, pagedFinder, countOperation, existsOperation, entityManager);
 	}
 
 	@Override
@@ -138,6 +155,14 @@ class FetchableFluentQueryByPredicate<S, R> extends FluentQuerySupport<S, R> imp
 		return convert(createSortedAndProjectedQuery().fetch());
 	}
 
+	@Override
+	public Window<R> scroll(ScrollPosition scrollPosition) {
+
+		Assert.notNull(scrollPosition, "ScrollPosition must not be null");
+
+		return scroll.scroll(sort, limit, scrollPosition).map(getConversionFunction());
+	}
+
 	@Override
 	public Page<R> page(Pageable pageable) {
 		return pageable.isUnpaged() ? new PageImpl<>(all()) : readPage(pageable);
@@ -169,6 +194,10 @@ class FetchableFluentQueryByPredicate<S, R> extends FluentQuerySupport<S, R> imp
 			query.setHint(EntityGraphFactory.HINT, EntityGraphFactory.create(entityManager, entityType, properties));
 		}
 
+		if (limit != 0) {
+			query.limit(limit);
+		}
+
 		return query;
 	}
 
@@ -201,4 +230,24 @@ class FetchableFluentQueryByPredicate<S, R> extends FluentQuerySupport<S, R> imp
 		return getConversionFunction(entityType, resultType);
 	}
 
+
+	static class PredicateScrollDelegate<T> extends ScrollDelegate<T> {
+
+		private final ScrollQueryFactory<T> scrollFunction;
+
+		PredicateScrollDelegate(ScrollQueryFactory<T> scrollQueryFactory, JpaEntityInformation<T, ?> entity) {
+			super(entity);
+			this.scrollFunction = scrollQueryFactory;
+		}
+
+		public Window<T> scroll(Sort sort, int limit, ScrollPosition scrollPosition) {
+
+			Query query = scrollFunction.createQuery(sort, scrollPosition);
+			if (limit > 0) {
+				query = query.setMaxResults(limit);
+			}
+			return scroll(query, sort, scrollPosition);
+		}
+	}
+
 }

spring-data-jpa/src/main/java/org/springframework/data/jpa/repository/support/FetchableFluentQueryBySpecification.java

@@ -16,6 +16,7 @@
 package org.springframework.data.jpa.repository.support;
 
 import jakarta.persistence.EntityManager;
+import jakarta.persistence.Query;
 import jakarta.persistence.TypedQuery;
 
 import java.util.ArrayList;
@@ -29,8 +30,11 @@ import org.springframework.dao.IncorrectResultSizeDataAccessException;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.PageImpl;
 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.jpa.domain.Specification;
+import org.springframework.data.jpa.repository.query.ScrollDelegate;
 import org.springframework.data.jpa.support.PageableUtils;
 import org.springframework.data.repository.query.FluentQuery;
 import org.springframework.data.support.PageableExecutionUtils;
@@ -50,26 +54,28 @@ class FetchableFluentQueryBySpecification<S, R> extends FluentQuerySupport<S, R>
 
 	private final Specification<S> spec;
 	private final Function<Sort, TypedQuery<S>> finder;
+	private final SpecificationScrollDelegate<S> scroll;
 	private final Function<Specification<S>, Long> countOperation;
 	private final Function<Specification<S>, Boolean> existsOperation;
 	private final EntityManager entityManager;
 
-	public FetchableFluentQueryBySpecification(Specification<S> spec, Class<S> entityType, Sort sort,
-			Collection<String> properties, Function<Sort, TypedQuery<S>> finder,
+	public FetchableFluentQueryBySpecification(Specification<S> spec, Class<S> entityType,
+			Function<Sort, TypedQuery<S>> finder, SpecificationScrollDelegate<S> scrollDelegate,
 			Function<Specification<S>, Long> countOperation, Function<Specification<S>, Boolean> existsOperation,
 			EntityManager entityManager) {
-		this(spec, entityType, (Class<R>) entityType, Sort.unsorted(), Collections.emptySet(), finder, countOperation,
-				existsOperation, entityManager);
+		this(spec, entityType, (Class<R>) entityType, Sort.unsorted(), 0, Collections.emptySet(), finder, scrollDelegate,
+				countOperation, existsOperation, entityManager);
 	}
 
 	private FetchableFluentQueryBySpecification(Specification<S> spec, Class<S> entityType, Class<R> resultType,
-			Sort sort, Collection<String> properties, Function<Sort, TypedQuery<S>> finder,
-			Function<Specification<S>, Long> countOperation, Function<Specification<S>, Boolean> existsOperation,
-			EntityManager entityManager) {
+			Sort sort, int limit, Collection<String> properties, Function<Sort, TypedQuery<S>> finder,
+			SpecificationScrollDelegate<S> scrollDelegate, Function<Specification<S>, Long> countOperation,
+			Function<Specification<S>, Boolean> existsOperation, EntityManager entityManager) {
 
-		super(resultType, sort, properties, entityType);
+		super(resultType, sort, limit, properties, entityType);
 		this.spec = spec;
 		this.finder = finder;
+		this.scroll = scrollDelegate;
 		this.countOperation = countOperation;
 		this.existsOperation = existsOperation;
 		this.entityManager = entityManager;
@@ -80,8 +86,17 @@ class FetchableFluentQueryBySpecification<S, R> extends FluentQuerySupport<S, R>
 
 		Assert.notNull(sort, "Sort must not be null");
 
-		return new FetchableFluentQueryBySpecification<>(spec, entityType, resultType, this.sort.and(sort), properties,
-				finder, countOperation, existsOperation, entityManager);
+		return new FetchableFluentQueryBySpecification<>(spec, entityType, resultType, this.sort.and(sort), limit,
+				properties, finder, scroll, countOperation, existsOperation, entityManager);
+	}
+
+	@Override
+	public FetchableFluentQuery<R> limit(int limit) {
+
+		Assert.isTrue(limit >= 0, "Limit must not be negative");
+
+		return new FetchableFluentQueryBySpecification<>(spec, entityType, resultType, this.sort.and(sort), limit,
+				properties, finder, scroll, countOperation, existsOperation, entityManager);
 	}
 
 	@Override
@@ -92,15 +107,15 @@ class FetchableFluentQueryBySpecification<S, R> extends FluentQuerySupport<S, R>
 			throw new UnsupportedOperationException("Class-based DTOs are not yet supported.");
 		}
 
-		return new FetchableFluentQueryBySpecification<>(spec, entityType, resultType, sort, properties, finder,
-				countOperation, existsOperation, entityManager);
+		return new FetchableFluentQueryBySpecification<>(spec, entityType, resultType, sort, limit, properties, finder,
+				scroll, countOperation, existsOperation, entityManager);
 	}
 
 	@Override
 	public FetchableFluentQuery<R> project(Collection<String> properties) {
 
-		return new FetchableFluentQueryBySpecification<>(spec, entityType, resultType, sort, properties, finder,
-				countOperation, existsOperation, entityManager);
+		return new FetchableFluentQueryBySpecification<>(spec, entityType, resultType, sort, limit, properties, finder,
+				scroll, countOperation, existsOperation, entityManager);
 	}
 
 	@Override
@@ -132,6 +147,14 @@ class FetchableFluentQueryBySpecification<S, R> extends FluentQuerySupport<S, R>
 		return convert(createSortedAndProjectedQuery().getResultList());
 	}
 
+	@Override
+	public Window<R> scroll(ScrollPosition scrollPosition) {
+
+		Assert.notNull(scrollPosition, "ScrollPosition must not be null");
+
+		return scroll.scroll(sort, limit, scrollPosition).map(getConversionFunction());
+	}
+
 	@Override
 	public Page<R> page(Pageable pageable) {
 		return pageable.isUnpaged() ? new PageImpl<>(all()) : readPage(pageable);
@@ -163,6 +186,10 @@ class FetchableFluentQueryBySpecification<S, R> extends FluentQuerySupport<S, R>
 			query.setHint(EntityGraphFactory.HINT, EntityGraphFactory.create(entityManager, entityType, properties));
 		}
 
+		if (limit != 0) {
+			query.setMaxResults(limit);
+		}
+
 		return query;
 	}
 
@@ -194,4 +221,25 @@ class FetchableFluentQueryBySpecification<S, R> extends FluentQuerySupport<S, R>
 	private Function<Object, R> getConversionFunction() {
 		return getConversionFunction(entityType, resultType);
 	}
+
+	static class SpecificationScrollDelegate<T> extends ScrollDelegate<T> {
+
+		private final ScrollQueryFactory<T> scrollFunction;
+
+		SpecificationScrollDelegate(ScrollQueryFactory<T> scrollQueryFactory, JpaEntityInformation<T, ?> entity) {
+			super(entity);
+			this.scrollFunction = scrollQueryFactory;
+		}
+
+		public Window<T> scroll(Sort sort, int limit, ScrollPosition scrollPosition) {
+
+			Query query = scrollFunction.createQuery(sort, scrollPosition);
+
+			if (limit > 0) {
+				query = query.setMaxResults(limit);
+			}
+
+			return scroll(query, sort, scrollPosition);
+		}
+	}
 }

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

@@ -15,6 +15,8 @@
  */
 package org.springframework.data.jpa.repository.support;
 
+import jakarta.persistence.Query;
+
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
@@ -22,6 +24,7 @@ import java.util.Set;
 import java.util.function.Function;
 
 import org.springframework.core.convert.support.DefaultConversionService;
+import org.springframework.data.domain.ScrollPosition;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
 import org.springframework.lang.Nullable;
@@ -32,21 +35,25 @@ import org.springframework.lang.Nullable;
  * @param <R> The resulting type of the query.
  * @author Greg Turnquist
  * @author Jens Schauder
+ * @author Mark Paluch
  * @since 2.6
  */
 abstract class FluentQuerySupport<S, R> {
 
 	protected final Class<R> resultType;
 	protected final Sort sort;
+	protected final int limit;
 	protected final Set<String> properties;
 	protected final Class<S> entityType;
 
 	private final SpelAwareProxyProjectionFactory projectionFactory = new SpelAwareProxyProjectionFactory();
 
-	FluentQuerySupport(Class<R> resultType, Sort sort, @Nullable Collection<String> properties, Class<S> entityType) {
+	FluentQuerySupport(Class<R> resultType, Sort sort, int limit, @Nullable Collection<String> properties,
+			Class<S> entityType) {
 
 		this.resultType = resultType;
 		this.sort = sort;
+		this.limit = limit;
 
 		if (properties != null) {
 			this.properties = new HashSet<>(properties);
@@ -78,4 +85,9 @@ abstract class FluentQuerySupport<S, R> {
 
 		return o -> DefaultConversionService.getSharedInstance().convert(o, targetType);
 	}
+
+	interface ScrollQueryFactory<T> {
+		Query createQuery(Sort sort, ScrollPosition scrollPosition);
+	}
+
 }

spring-data-jpa/src/main/java/org/springframework/data/jpa/repository/support/JpaEntityInformation.java

@@ -17,6 +17,9 @@ package org.springframework.data.jpa.repository.support;
 
 import jakarta.persistence.metamodel.SingularAttribute;
 
+import java.util.Collection;
+import java.util.Map;
+
 import org.springframework.data.jpa.repository.query.JpaEntityMetadata;
 import org.springframework.data.repository.core.EntityInformation;
 import org.springframework.lang.Nullable;
@@ -70,7 +73,7 @@ public interface JpaEntityInformation<T, ID> extends EntityInformation<T, ID>, J
 	 *
 	 * @return
 	 */
-	Iterable<String> getIdAttributeNames();
+	Collection<String> getIdAttributeNames();
 
 	/**
 	 * Extracts the value for the given id attribute from a composite id
@@ -81,4 +84,14 @@ public interface JpaEntityInformation<T, ID> extends EntityInformation<T, ID>, J
 	 */
 	@Nullable
 	Object getCompositeIdAttributeValue(Object id, String idAttribute);
+
+	/**
+	 * Extract a keyset for {@code propertyPaths} and the primary key (including composite key components if applicable).
+	 *
+	 * @param propertyPaths
+	 * @param entity
+	 * @return
+	 * @since 3.1
+	 */
+	Map<String, Object> getKeyset(Iterable<String> propertyPaths, T entity);
 }

spring-data-jpa/src/main/java/org/springframework/data/jpa/repository/support/JpaMetamodelEntityInformation.java

@@ -26,9 +26,12 @@ import jakarta.persistence.metamodel.SingularAttribute;
 import jakarta.persistence.metamodel.Type;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.Iterator;
+import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 
@@ -61,7 +64,7 @@ public class JpaMetamodelEntityInformation<T, ID> extends JpaEntityInformationSu
 
 	/**
 	 * Creates a new {@link JpaMetamodelEntityInformation} for the given domain class and {@link Metamodel}.
-	 * 
+	 *
 	 * @param domainClass must not be {@literal null}.
 	 * @param metamodel must not be {@literal null}.
 	 * @param persistenceUnitUtil must not be {@literal null}.
@@ -190,7 +193,7 @@ public class JpaMetamodelEntityInformation<T, ID> extends JpaEntityInformationSu
 	}
 
 	@Override
-	public Iterable<String> getIdAttributeNames() {
+	public Collection<String> getIdAttributeNames() {
 
 		List<String> attributeNames = new ArrayList<>(idMetadata.attributes.size());
 
@@ -222,6 +225,30 @@ public class JpaMetamodelEntityInformation<T, ID> extends JpaEntityInformationSu
 		return versionAttribute.map(it -> wrapper.getPropertyValue(it.getName()) == null).orElse(true);
 	}
 
+	@Override
+	public Map<String, Object> getKeyset(Iterable<String> propertyPaths, T entity) {
+
+		// TODO: Proxy handling requires more elaborate refactoring, see
+		// https://github.com/spring-projects/spring-data-jpa/issues/2784
+		BeanWrapper entityWrapper = new DirectFieldAccessFallbackBeanWrapper(entity);
+
+		Map<String, Object> keyset = new LinkedHashMap<>();
+
+		if (hasCompositeId()) {
+			for (String idAttributeName : getIdAttributeNames()) {
+				keyset.put(idAttributeName, entityWrapper.getPropertyValue(idAttributeName));
+			}
+		} else {
+			keyset.put(getIdAttribute().getName(), getId(entity));
+		}
+
+		for (String propertyPath : propertyPaths) {
+			keyset.put(propertyPath, entityWrapper.getPropertyValue(propertyPath));
+		}
+
+		return keyset;
+	}
+
 	/**
 	 * Simple value object to encapsulate id specific metadata.
 	 *

spring-data-jpa/src/main/java/org/springframework/data/jpa/repository/support/Querydsl.java

@@ -15,10 +15,10 @@
  */
 package org.springframework.data.jpa.repository.support;
 
-import java.util.List;
-
 import jakarta.persistence.EntityManager;
 
+import java.util.List;
+
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.domain.Sort.Order;
@@ -245,4 +245,29 @@ public class Querydsl {
 
 		return sortPropertyExpression;
 	}
+
+	/**
+	 * Creates an {@link Expression} for the given {@code property} property.
+	 *
+	 * @param property must not be {@literal null}.
+	 * @return
+	 */
+	Expression<?> createExpression(String property) {
+
+		Assert.notNull(property, "Property must not be null");
+
+		PropertyPath path = PropertyPath.from(property, builder.getType());
+		Expression<?> sortPropertyExpression = builder;
+
+		while (path != null) {
+
+			sortPropertyExpression = !path.hasNext() && String.class.equals(path.getType()) //
+					? Expressions.stringPath((Path<?>) sortPropertyExpression, path.getSegment()) //
+					: Expressions.path(path.getType(), (Path<?>) sortPropertyExpression, path.getSegment());
+
+			path = path.next();
+		}
+
+		return sortPropertyExpression;
+	}
 }

spring-data-jpa/src/main/java/org/springframework/data/jpa/repository/support/QuerydslJpaPredicateExecutor.java

@@ -15,19 +15,27 @@
  */
 package org.springframework.data.jpa.repository.support;
 
+import jakarta.persistence.EntityManager;
+import jakarta.persistence.LockModeType;
+
 import java.util.List;
 import java.util.Optional;
 import java.util.function.BiFunction;
 import java.util.function.Function;
 
-import jakarta.persistence.EntityManager;
-import jakarta.persistence.LockModeType;
-
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
+import org.springframework.data.domain.KeysetScrollPosition;
+import org.springframework.data.domain.OffsetScrollPosition;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
+import org.springframework.data.domain.Sort.Order;
 import org.springframework.data.jpa.repository.EntityGraph;
+import org.springframework.data.jpa.repository.query.KeysetScrollDelegate;
+import org.springframework.data.jpa.repository.query.KeysetScrollDelegate.QueryStrategy;
+import org.springframework.data.jpa.repository.query.KeysetScrollSpecification;
+import org.springframework.data.jpa.repository.support.FetchableFluentQueryByPredicate.PredicateScrollDelegate;
+import org.springframework.data.jpa.repository.support.FluentQuerySupport.ScrollQueryFactory;
 import org.springframework.data.querydsl.EntityPathResolver;
 import org.springframework.data.querydsl.QSort;
 import org.springframework.data.querydsl.QuerydslPredicateExecutor;
@@ -37,9 +45,14 @@ import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
 import com.querydsl.core.NonUniqueResultException;
+import com.querydsl.core.types.ConstantImpl;
 import com.querydsl.core.types.EntityPath;
+import com.querydsl.core.types.Expression;
+import com.querydsl.core.types.NullExpression;
+import com.querydsl.core.types.Ops;
 import com.querydsl.core.types.OrderSpecifier;
 import com.querydsl.core.types.Predicate;
+import com.querydsl.core.types.dsl.BooleanExpression;
 import com.querydsl.core.types.dsl.Expressions;
 import com.querydsl.core.types.dsl.PathBuilder;
 import com.querydsl.jpa.JPQLQuery;
@@ -63,6 +76,7 @@ public class QuerydslJpaPredicateExecutor<T> implements QuerydslPredicateExecuto
 	private final JpaEntityInformation<T, ?> entityInformation;
 	private final EntityPath<T> path;
 	private final Querydsl querydsl;
+	private final QuerydslQueryStrategy scrollQueryAdapter;
 	private final EntityManager entityManager;
 	private final CrudMethodMetadata metadata;
 
@@ -83,6 +97,7 @@ public class QuerydslJpaPredicateExecutor<T> implements QuerydslPredicateExecuto
 		this.path = resolver.createPath(entityInformation.getJavaType());
 		this.querydsl = new Querydsl(entityManager, new PathBuilder<T>(path.getType(), path.getMetadata()));
 		this.entityManager = entityManager;
+		this.scrollQueryAdapter = new QuerydslQueryStrategy();
 	}
 
 	@Override
@@ -160,6 +175,33 @@ public class QuerydslJpaPredicateExecutor<T> implements QuerydslPredicateExecuto
 			return select;
 		};
 
+		ScrollQueryFactory<T> scroll = (sort, scrollPosition) -> {
+
+			Predicate predicateToUse = predicate;
+
+			if (scrollPosition instanceof KeysetScrollPosition keyset) {
+
+				KeysetScrollDelegate delegate = KeysetScrollDelegate.of(keyset.getDirection());
+				sort = KeysetScrollSpecification.createSort(keyset, sort, entityInformation);
+				BooleanExpression keysetPredicate = delegate.createPredicate(keyset, sort, scrollQueryAdapter);
+
+				if (keysetPredicate != null) {
+					predicateToUse = predicate instanceof BooleanExpression be ? be.and(keysetPredicate)
+							: keysetPredicate.and(predicate);
+				}
+			}
+
+			AbstractJPAQuery<?, ?> select = (AbstractJPAQuery<?, ?>) createQuery(predicateToUse).select(path);
+
+			select = (AbstractJPAQuery<?, ?>) querydsl.applySorting(sort, select);
+
+			if (scrollPosition instanceof OffsetScrollPosition offset) {
+				select.offset(offset.getOffset());
+			}
+
+			return select.createQuery();
+		};
+
 		BiFunction<Sort, Pageable, AbstractJPAQuery<?, ?>> pagedFinder = (sort, pageable) -> {
 
 			AbstractJPAQuery<?, ?> select = finder.apply(sort);
@@ -175,6 +217,7 @@ public class QuerydslJpaPredicateExecutor<T> implements QuerydslPredicateExecuto
 				predicate, //
 				this.entityInformation.getJavaType(), //
 				finder, //
+				new PredicateScrollDelegate<>(scroll, entityInformation), //
 				pagedFinder, //
 				this::count, //
 				this::exists, //
@@ -285,4 +328,34 @@ public class QuerydslJpaPredicateExecutor<T> implements QuerydslPredicateExecuto
 	private List<T> executeSorted(JPQLQuery<T> query, Sort sort) {
 		return querydsl.applySorting(sort, query).fetch();
 	}
+
+	class QuerydslQueryStrategy implements QueryStrategy<Expression<?>, BooleanExpression> {
+
+		@Override
+		public Expression<?> createExpression(String property) {
+			return querydsl.createExpression(property);
+		}
+
+		@Override
+		public BooleanExpression compare(Order order, Expression<?> propertyExpression, Object value) {
+			return Expressions.booleanOperation(order.isAscending() ? Ops.GT : Ops.LT, propertyExpression,
+					ConstantImpl.create(value));
+		}
+
+		@Override
+		public BooleanExpression compare(Expression<?> propertyExpression, @Nullable Object value) {
+			return Expressions.booleanOperation(Ops.EQ, propertyExpression,
+					value == null ? NullExpression.DEFAULT : ConstantImpl.create(value));
+		}
+
+		@Override
+		public BooleanExpression and(List<BooleanExpression> intermediate) {
+			return Expressions.allOf(intermediate.toArray(new BooleanExpression[0]));
+		}
+
+		@Override
+		public BooleanExpression or(List<BooleanExpression> intermediate) {
+			return Expressions.anyOf(intermediate.toArray(new BooleanExpression[0]));
+		}
+	}
 }

spring-data-jpa/src/main/java/org/springframework/data/jpa/repository/support/SimpleJpaRepository.java

@@ -43,6 +43,8 @@ import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
 import org.springframework.data.domain.Example;
+import org.springframework.data.domain.KeysetScrollPosition;
+import org.springframework.data.domain.OffsetScrollPosition;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.PageImpl;
 import org.springframework.data.domain.Pageable;
@@ -52,7 +54,10 @@ import org.springframework.data.jpa.domain.Specification;
 import org.springframework.data.jpa.provider.PersistenceProvider;
 import org.springframework.data.jpa.repository.EntityGraph;
 import org.springframework.data.jpa.repository.query.EscapeCharacter;
+import org.springframework.data.jpa.repository.query.KeysetScrollSpecification;
 import org.springframework.data.jpa.repository.query.QueryUtils;
+import org.springframework.data.jpa.repository.support.FetchableFluentQueryBySpecification.SpecificationScrollDelegate;
+import org.springframework.data.jpa.repository.support.FluentQuerySupport.ScrollQueryFactory;
 import org.springframework.data.jpa.repository.support.QueryHints.NoHints;
 import org.springframework.data.jpa.support.PageableUtils;
 import org.springframework.data.repository.query.FluentQuery.FetchableFluentQuery;
@@ -507,10 +512,40 @@ public class SimpleJpaRepository<T, ID> implements JpaRepositoryImplementation<T
 		Assert.notNull(spec, "Specification must not be null");
 		Assert.notNull(queryFunction, "Query function must not be null");
 
-		Function<Sort, TypedQuery<T>> finder = sort -> getQuery(spec, getDomainClass(), sort);
+		return doFindBy(spec, getDomainClass(), queryFunction);
+	}
 
-		FetchableFluentQuery<R> fluentQuery = new FetchableFluentQueryBySpecification<T, R>(spec, getDomainClass(),
-				Sort.unsorted(), null, finder, this::count, this::exists, this.em);
+	private <S extends T, R> R doFindBy(Specification<T> spec, Class<T> domainClass,
+			Function<FetchableFluentQuery<S>, R> queryFunction) {
+
+		Assert.notNull(spec, "Specification must not be null");
+		Assert.notNull(queryFunction, "Query function must not be null");
+
+		ScrollQueryFactory<T> scrollFunction = (sort, scrollPosition) -> {
+
+			Specification<T> specToUse = spec;
+
+			if (scrollPosition instanceof KeysetScrollPosition keyset) {
+				KeysetScrollSpecification<T> keysetSpec = new KeysetScrollSpecification<>(keyset, sort, entityInformation);
+				sort = keysetSpec.sort();
+				specToUse = specToUse.and(keysetSpec);
+			}
+
+			TypedQuery<T> query = getQuery(specToUse, domainClass, sort);
+
+			if (scrollPosition instanceof OffsetScrollPosition offset) {
+				query.setFirstResult(Math.toIntExact(offset.getOffset()));
+			}
+
+			return query;
+		};
+
+		Function<Sort, TypedQuery<T>> finder = sort -> getQuery(spec, domainClass, sort);
+
+		SpecificationScrollDelegate<T> scrollDelegate = new SpecificationScrollDelegate<>(scrollFunction,
+				entityInformation);
+		FetchableFluentQuery<T> fluentQuery = new FetchableFluentQueryBySpecification<>(spec, domainClass, finder,
+				scrollDelegate, this::count, this::exists, this.em);
 
 		return queryFunction.apply((FetchableFluentQuery<S>) fluentQuery);
 	}
@@ -544,7 +579,6 @@ public class SimpleJpaRepository<T, ID> implements JpaRepositoryImplementation<T
 		return query.setMaxResults(1).getResultList().size() == 1;
 	}
 
-
 	@Override
 	public <S extends T> List<S> findAll(Example<S> example) {
 		return getQuery(new ExampleSpecification<>(example, escapeCharacter), example.getProbeType(), Sort.unsorted())
@@ -572,21 +606,12 @@ public class SimpleJpaRepository<T, ID> implements JpaRepositoryImplementation<T
 		Assert.notNull(example, "Sample must not be null");
 		Assert.notNull(queryFunction, "Query function must not be null");
 
-		Function<Sort, TypedQuery<S>> finder = sort -> {
+		ExampleSpecification<S> spec = new ExampleSpecification<>(example, escapeCharacter);
+		Class<S> probeType = example.getProbeType();
 
-			ExampleSpecification<S> spec = new ExampleSpecification<>(example, escapeCharacter);
-			Class<S> probeType = example.getProbeType();
-
-			return getQuery(spec, probeType, sort);
-		};
-
-		FetchableFluentQuery<S> fluentQuery = new FetchableFluentQueryByExample<>(example, finder, this::count,
-				this::exists, this.em, this.escapeCharacter);
-
-		return queryFunction.apply(fluentQuery);
+		return doFindBy((Specification<T>) spec, (Class<T>) probeType, queryFunction);
 	}
 
-
 	@Override
 	public long count() {
 

spring-data-jpa/src/test/java/org/springframework/data/jpa/domain/sample/Item.java

@@ -21,6 +21,8 @@ import jakarta.persistence.Id;
 import jakarta.persistence.IdClass;
 import jakarta.persistence.JoinColumn;
 import jakarta.persistence.Table;
+import lombok.EqualsAndHashCode;
+import lombok.ToString;
 
 /**
  * @author Mark Paluch
@@ -30,12 +32,16 @@ import jakarta.persistence.Table;
 @Entity
 @Table
 @IdClass(ItemId.class)
+@EqualsAndHashCode
+@ToString
 public class Item {
 
 	@Id @Column(columnDefinition = "INT") private Integer id;
 
 	@Id @JoinColumn(name = "manufacturer_id", columnDefinition = "INT") private Integer manufacturerId;
 
+	private String name;
+
 	public Item() {}
 
 	public Item(Integer id, Integer manufacturerId) {
@@ -43,6 +49,12 @@ public class Item {
 		this.manufacturerId = manufacturerId;
 	}
 
+	public Item(Integer id, Integer manufacturerId, String name) {
+		this.id = id;
+		this.manufacturerId = manufacturerId;
+		this.name = name;
+	}
+
 	public Integer getId() {
 		return id;
 	}
@@ -50,4 +62,13 @@ public class Item {
 	public Integer getManufacturerId() {
 		return manufacturerId;
 	}
+
+	public String getName() {
+		return name;
+	}
+
+	public void setName(String name) {
+		this.name = name;
+	}
+
 }

spring-data-jpa/src/test/java/org/springframework/data/jpa/repository/RepositoryWithIdClassKeyTests.java

@@ -15,8 +15,9 @@
  */
 package org.springframework.data.jpa.repository;
 
-import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.*;
 
+import java.util.Arrays;
 import java.util.Optional;
 
 import org.junit.jupiter.api.Test;
@@ -24,6 +25,9 @@ import org.junit.jupiter.api.extension.ExtendWith;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.ImportResource;
+import org.springframework.data.domain.KeysetScrollPosition;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.domain.Window;
 import org.springframework.data.jpa.domain.sample.Item;
 import org.springframework.data.jpa.domain.sample.ItemId;
 import org.springframework.data.jpa.domain.sample.ItemSite;
@@ -75,6 +79,28 @@ class RepositoryWithIdClassKeyTests {
 		assertThat(loaded).isPresent();
 	}
 
+	@Test // GH-2878
+	void shouldScrollWithKeyset() {
+
+		Item item1 = new Item(1, 2, "a");
+		Item item2 = new Item(2, 3, "b");
+		Item item3 = new Item(3, 4, "c");
+
+		itemRepository.saveAllAndFlush(Arrays.asList(item1, item2, item3));
+
+		Window<Item> first = itemRepository.findBy((root, query, criteriaBuilder) -> {
+			return criteriaBuilder.isNotNull(root.get("name"));
+		}, q -> q.limit(1).sortBy(Sort.by("name")).scroll(KeysetScrollPosition.initial()));
+
+		assertThat(first).containsOnly(item1);
+
+		Window<Item> next = itemRepository.findBy((root, query, criteriaBuilder) -> {
+			return criteriaBuilder.isNotNull(root.get("name"));
+		}, q -> q.limit(1).sortBy(Sort.by("name")).scroll(first.positionAt(0)));
+
+		assertThat(next).containsOnly(item2);
+	}
+
 	@Configuration
 	@EnableJpaRepositories(basePackageClasses = SampleConfig.class)
 	static abstract class Config {

spring-data-jpa/src/test/java/org/springframework/data/jpa/repository/UserRepositoryTests.java

@@ -34,6 +34,7 @@ import jakarta.persistence.criteria.Root;
 import lombok.Data;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
@@ -54,14 +55,7 @@ import org.springframework.dao.DataAccessException;
 import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
-import org.springframework.data.domain.Example;
-import org.springframework.data.domain.ExampleMatcher;
-import org.springframework.data.domain.Page;
-import org.springframework.data.domain.PageImpl;
-import org.springframework.data.domain.PageRequest;
-import org.springframework.data.domain.Pageable;
-import org.springframework.data.domain.Slice;
-import org.springframework.data.domain.Sort;
+import org.springframework.data.domain.*;
 import org.springframework.data.domain.Sort.Direction;
 import org.springframework.data.domain.Sort.Order;
 import org.springframework.data.jpa.domain.Specification;
@@ -1235,6 +1229,185 @@ class UserRepositoryTests {
 		assertThat(result).hasSize(1);
 	}
 
+	@Test // GH-2878
+	void scrollByExampleOffset() {
+
+		User jane1 = new User("Jane", "Doe", "jane@doe1.com");
+		User jane2 = new User("Jane", "Doe", "jane@doe2.com");
+		User john1 = new User("John", "Doe", "john@doe1.com");
+		User john2 = new User("John", "Doe", "john@doe2.com");
+
+		repository.saveAllAndFlush(Arrays.asList(john1, john2, jane1, jane2));
+
+		Example<User> example = Example.of(new User("J", null, null),
+				matching().withMatcher("firstname", GenericPropertyMatcher::startsWith).withIgnorePaths("age", "createdAt",
+						"dateOfBirth"));
+		Window<User> firstWindow = repository.findBy(example,
+				q -> q.limit(2).sortBy(Sort.by("firstname")).scroll(OffsetScrollPosition.initial()));
+
+		assertThat(firstWindow).containsExactly(jane1, jane2);
+		assertThat(firstWindow.hasNext()).isTrue();
+
+		Window<User> nextWindow = repository.findBy(example,
+				q -> q.limit(2).sortBy(Sort.by("firstname")).scroll(firstWindow.positionAt(1)));
+
+		assertThat(nextWindow).containsExactly(john1, john2);
+		assertThat(nextWindow.hasNext()).isFalse();
+	}
+
+	@Test // GH-2878
+	void scrollByExampleKeyset() {
+
+		User jane1 = new User("Jane", "Doe", "jane@doe1.com");
+		User jane2 = new User("Jane", "Doe", "jane@doe2.com");
+		User john1 = new User("John", "Doe", "john@doe1.com");
+		User john2 = new User("John", "Doe", "john@doe2.com");
+
+		repository.saveAllAndFlush(Arrays.asList(john1, john2, jane1, jane2));
+
+		Example<User> example = Example.of(new User("J", null, null),
+				matching().withMatcher("firstname", GenericPropertyMatcher::startsWith).withIgnorePaths("age", "createdAt",
+						"dateOfBirth"));
+		Window<User> firstWindow = repository.findBy(example,
+				q -> q.limit(1).sortBy(Sort.by("firstname", "emailAddress")).scroll(KeysetScrollPosition.initial()));
+
+		assertThat(firstWindow).containsOnly(jane1);
+		assertThat(firstWindow.hasNext()).isTrue();
+
+		Window<User> nextWindow = repository.findBy(example,
+				q -> q.limit(2).sortBy(Sort.by("firstname", "emailAddress")).scroll(firstWindow.positionAt(0)));
+
+		assertThat(nextWindow).containsExactly(jane2, john1);
+		assertThat(nextWindow.hasNext()).isTrue();
+	}
+
+	@Test // GH-2878
+	void scrollByExampleKeysetBackward() {
+
+		User jane1 = new User("Jane", "Doe", "jane@doe1.com");
+		User jane2 = new User("Jane", "Doe", "jane@doe2.com");
+		User john1 = new User("John", "Doe", "john@doe1.com");
+		User john2 = new User("John", "Doe", "john@doe2.com");
+
+		repository.saveAllAndFlush(Arrays.asList(john1, john2, jane1, jane2));
+
+		Example<User> example = Example.of(new User("J", null, null),
+				matching().withMatcher("firstname", GenericPropertyMatcher::startsWith).withIgnorePaths("age", "createdAt",
+						"dateOfBirth"));
+		Window<User> firstWindow = repository.findBy(example,
+				q -> q.limit(4).sortBy(Sort.by("firstname", "emailAddress")).scroll(KeysetScrollPosition.initial()));
+
+		KeysetScrollPosition scrollPosition = (KeysetScrollPosition) firstWindow.positionAt(2);
+		Window<User> previousWindow = repository.findBy(example,
+				q -> q.limit(1).sortBy(Sort.by("firstname", "emailAddress"))
+						.scroll(KeysetScrollPosition.of(scrollPosition.getKeys(), KeysetScrollPosition.Direction.Backward)));
+
+		assertThat(previousWindow).containsOnly(jane2);
+		assertThat(previousWindow.hasNext()).isTrue();
+	}
+
+	@Test // GH-2878
+	void scrollByPredicateOffset() {
+
+		User jane1 = new User("Jane", "Doe", "jane@doe1.com");
+		User jane2 = new User("Jane", "Doe", "jane@doe2.com");
+		User john1 = new User("John", "Doe", "john@doe1.com");
+		User john2 = new User("John", "Doe", "john@doe2.com");
+
+		repository.saveAllAndFlush(Arrays.asList(john1, john2, jane1, jane2));
+
+		Window<User> firstWindow = repository.findBy(QUser.user.firstname.startsWith("J"),
+				q -> q.limit(2).sortBy(Sort.by("firstname")).scroll(OffsetScrollPosition.initial()));
+
+		assertThat(firstWindow).containsExactly(jane1, jane2);
+		assertThat(firstWindow.hasNext()).isTrue();
+
+		Window<User> nextWindow = repository.findBy(QUser.user.firstname.startsWith("J"),
+				q -> q.limit(2).sortBy(Sort.by("firstname")).scroll(firstWindow.positionAt(1)));
+
+		assertThat(nextWindow).containsExactly(john1, john2);
+		assertThat(nextWindow.hasNext()).isFalse();
+	}
+
+	@Test // GH-2878
+	void scrollByPredicateKeyset() {
+
+		User jane1 = new User("Jane", "Doe", "jane@doe1.com");
+		User jane2 = new User("Jane", "Doe", "jane@doe2.com");
+		User john1 = new User("John", "Doe", "john@doe1.com");
+		User john2 = new User("John", "Doe", "john@doe2.com");
+
+		repository.saveAllAndFlush(Arrays.asList(john1, john2, jane1, jane2));
+
+		Window<User> firstWindow = repository.findBy(QUser.user.firstname.startsWith("J"),
+				q -> q.limit(1).sortBy(Sort.by("firstname", "emailAddress")).scroll(KeysetScrollPosition.initial()));
+
+		assertThat(firstWindow).containsOnly(jane1);
+		assertThat(firstWindow.hasNext()).isTrue();
+
+		Window<User> nextWindow = repository.findBy(QUser.user.firstname.startsWith("J"),
+				q -> q.limit(2).sortBy(Sort.by("firstname", "emailAddress")).scroll(firstWindow.positionAt(0)));
+
+		assertThat(nextWindow).containsExactly(jane2, john1);
+		assertThat(nextWindow.hasNext()).isTrue();
+	}
+
+	@Test // GH-2878
+	void scrollByPredicateKeysetBackward() {
+
+		User jane1 = new User("Jane", "Doe", "jane@doe1.com");
+		User jane2 = new User("Jane", "Doe", "jane@doe2.com");
+		User john1 = new User("John", "Doe", "john@doe1.com");
+		User john2 = new User("John", "Doe", "john@doe2.com");
+
+		repository.saveAllAndFlush(Arrays.asList(john1, john2, jane1, jane2));
+
+		Window<User> firstWindow = repository.findBy(QUser.user.firstname.startsWith("J"),
+				q -> q.limit(3).sortBy(Sort.by("firstname", "emailAddress")).scroll(KeysetScrollPosition.initial()));
+
+		assertThat(firstWindow).containsExactly(jane1, jane2, john1);
+		assertThat(firstWindow.hasNext()).isTrue();
+
+		KeysetScrollPosition scrollPosition = (KeysetScrollPosition) firstWindow.positionAt(2);
+		KeysetScrollPosition backward = KeysetScrollPosition.of(scrollPosition.getKeys(),
+				KeysetScrollPosition.Direction.Backward);
+		Window<User> previousWindow = repository.findBy(QUser.user.firstname.startsWith("J"),
+				q -> q.limit(3).sortBy(Sort.by("firstname", "emailAddress")).scroll(backward));
+
+		assertThat(previousWindow).containsExactly(jane1, jane2);
+
+		// no more items before this window
+		assertThat(previousWindow.hasNext()).isFalse();
+	}
+
+	@Test // GH-2878
+	void scrollByPartTreeKeysetBackward() {
+
+		User jane1 = new User("Jane", "Doe", "jane@doe1.com");
+		User jane2 = new User("Jane", "Doe", "jane@doe2.com");
+		User john1 = new User("John", "Doe", "john@doe1.com");
+		User john2 = new User("John", "Doe", "john@doe2.com");
+
+		repository.saveAllAndFlush(Arrays.asList(john1, john2, jane1, jane2));
+
+		Window<User> firstWindow = repository.findTop3ByFirstnameStartingWithOrderByFirstnameAscEmailAddressAsc("J",
+				KeysetScrollPosition.initial());
+
+		assertThat(firstWindow).containsExactly(jane1, jane2, john1);
+		assertThat(firstWindow.hasNext()).isTrue();
+
+		KeysetScrollPosition scrollPosition = (KeysetScrollPosition) firstWindow.positionAt(2);
+		KeysetScrollPosition backward = KeysetScrollPosition.of(scrollPosition.getKeys(),
+				KeysetScrollPosition.Direction.Backward);
+		Window<User> previousWindow = repository.findTop3ByFirstnameStartingWithOrderByFirstnameAscEmailAddressAsc("J",
+				backward);
+
+		assertThat(previousWindow).containsExactly(jane1, jane2);
+
+		// no more items before this window
+		assertThat(previousWindow.hasNext()).isFalse();
+	}
+
 	@Test // DATAJPA-491
 	void sortByNestedAssociationPropertyWithSortInPageable() {
 

spring-data-jpa/src/test/java/org/springframework/data/jpa/repository/query/CollectionUtilsUnitTests.java

@@ -0,0 +1,46 @@
+/*
+ * 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.jpa.repository.query;
+
+import static org.assertj.core.api.Assertions.*;
+
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+
+/**
+ * Unit tests for {@link CollectionUtils}.
+ *
+ * @author Mark Paluch
+ */
+class CollectionUtilsUnitTests {
+
+	@Test // GH-2878
+	void shouldReturnFirstItems() {
+
+		assertThat(CollectionUtils.getFirst(2, List.of(1, 2, 3))).containsExactly(1, 2);
+		assertThat(CollectionUtils.getFirst(2, List.of(1, 2))).containsExactly(1, 2);
+		assertThat(CollectionUtils.getFirst(2, List.of(1))).containsExactly(1);
+	}
+
+	@Test // GH-2878
+	void shouldReturnLastItems() {
+
+		assertThat(CollectionUtils.getLast(2, List.of(1, 2, 3))).containsExactly(2, 3);
+		assertThat(CollectionUtils.getLast(2, List.of(1, 2))).containsExactly(1, 2);
+		assertThat(CollectionUtils.getLast(2, List.of(1))).containsExactly(1);
+	}
+}

spring-data-jpa/src/test/java/org/springframework/data/jpa/repository/sample/ItemRepository.java

@@ -18,10 +18,11 @@ package org.springframework.data.jpa.repository.sample;
 import org.springframework.data.jpa.domain.sample.Item;
 import org.springframework.data.jpa.domain.sample.ItemId;
 import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
 
 /**
  * @author Mark Paluch
  * @see <a href="download.oracle.com/otn-pub/jcp/persistence-2_1-fr-eval-spec/JavaPersistence.pdf">Final JPA 2.1
  *      Specification 2.4.1.3 Derived Identities Example 2</a>
  */
-public interface ItemRepository extends JpaRepository<Item, ItemId> {}
+public interface ItemRepository extends JpaRepository<Item, ItemId>, JpaSpecificationExecutor<Item> {}

spring-data-jpa/src/test/java/org/springframework/data/jpa/repository/sample/UserRepository.java

@@ -29,8 +29,10 @@ import java.util.stream.Stream;
 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.jpa.domain.sample.Role;
 import org.springframework.data.jpa.domain.sample.SpecialUser;
 import org.springframework.data.jpa.domain.sample.User;
@@ -150,6 +152,9 @@ public interface UserRepository extends JpaRepository<User, Integer>, JpaSpecifi
 
 	Page<User> findByFirstnameIn(Pageable pageable, String... firstnames);
 
+	Window<User> findTop3ByFirstnameStartingWithOrderByFirstnameAscEmailAddressAsc(String firstname,
+			ScrollPosition position);
+
 	List<User> findByFirstnameNotIn(Collection<String> firstnames);
 
 	// DATAJPA-292

spring-data-jpa/src/test/java/org/springframework/data/jpa/repository/support/FetchableFluentQueryByExampleUnitTests.java

@@ -1,43 +0,0 @@
-/*
- * Copyright 2022-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.jpa.repository.support;
-
-import static org.assertj.core.api.Assertions.*;
-
-import org.junit.jupiter.api.Test;
-import org.springframework.data.domain.Example;
-import org.springframework.data.domain.Sort;
-import org.springframework.data.domain.Sort.Order;
-import org.springframework.data.jpa.repository.support.FetchableFluentQueryByExample;
-
-/**
- * Unit tests for {@link FetchableFluentQueryByExample}.
- *
- * @author J.R. Onyschak
- */
-class FetchableFluentQueryByExampleUnitTests {
-
-	@Test // GH-2438
-	@SuppressWarnings({ "rawtypes", "unchecked" })
-	void multipleSortBy() {
-
-		Sort s1 = Sort.by(Order.by("s1"));
-		Sort s2 = Sort.by(Order.by("s2"));
-		FetchableFluentQueryByExample f = new FetchableFluentQueryByExample(Example.of(""), null, null, null, null, null);
-		f = (FetchableFluentQueryByExample) f.sortBy(s1).sortBy(s2);
-		assertThat(f.sort).isEqualTo(s1.and(s2));
-	}
-}

spring-data-jpa/src/test/java/org/springframework/data/jpa/repository/support/FetchableFluentQueryByPredicateUnitTests.java

@@ -20,7 +20,6 @@ import static org.assertj.core.api.Assertions.*;
 import org.junit.jupiter.api.Test;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.domain.Sort.Order;
-import org.springframework.data.jpa.repository.support.FetchableFluentQueryByPredicate;
 
 /**
  * Unit tests for {@link FetchableFluentQueryByPredicate}.
@@ -35,7 +34,8 @@ class FetchableFluentQueryByPredicateUnitTests {
 
 		Sort s1 = Sort.by(Order.by("s1"));
 		Sort s2 = Sort.by(Order.by("s2"));
-		FetchableFluentQueryByPredicate f = new FetchableFluentQueryByPredicate(null, null, null, null, null, null, null);
+		FetchableFluentQueryByPredicate f = new FetchableFluentQueryByPredicate(null, null, null, null, null, null, null,
+				null);
 		f = (FetchableFluentQueryByPredicate) f.sortBy(s1).sortBy(s2);
 		assertThat(f.sort).isEqualTo(s1.and(s2));
 	}

spring-data-jpa/src/test/java/org/springframework/data/jpa/repository/support/JpaEntityInformationSupportUnitTests.java

@@ -18,9 +18,6 @@ package org.springframework.data.jpa.repository.support;
 import static org.assertj.core.api.Assertions.*;
 import static org.mockito.Mockito.*;
 
-import java.io.Serializable;
-import java.util.Collections;
-
 import jakarta.persistence.Entity;
 import jakarta.persistence.EntityManager;
 import jakarta.persistence.EntityManagerFactory;
@@ -28,6 +25,11 @@ import jakarta.persistence.PersistenceUnitUtil;
 import jakarta.persistence.metamodel.Metamodel;
 import jakarta.persistence.metamodel.SingularAttribute;
 
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Map;
+
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.Mock;
@@ -97,7 +99,7 @@ class JpaEntityInformationSupportUnitTests {
 		}
 
 		@Override
-		public Iterable<String> getIdAttributeNames() {
+		public Collection<String> getIdAttributeNames() {
 			return Collections.emptySet();
 		}
 
@@ -110,6 +112,11 @@ class JpaEntityInformationSupportUnitTests {
 		public Object getCompositeIdAttributeValue(Object id, String idAttribute) {
 			return null;
 		}
+
+		@Override
+		public Map<String, Object> getKeyset(Iterable<String> propertyPaths, T entity) {
+			return null;
+		}
 	}
 
 	@Entity(name = "AnotherNamedUser")

src/main/asciidoc/index.adoc

@@ -5,8 +5,9 @@ Oliver Gierke; Thomas Darimont; Christoph Strobl; Mark Paluch; Jay Bryant; Greg
 ifdef::backend-epub3[:front-cover-image: image:epub-cover.png[Front Cover,1050,1600]]
 :spring-data-commons-docs: ../../../../spring-data-commons/src/main/asciidoc
 :spring-framework-docs: https://docs.spring.io/spring-framework/docs/{springVersion}/spring-framework-reference/
+:feature-scroll: true
 
-(C) 2008-2022 The original authors.
+(C) 2008-2023 The original authors.
 
 NOTE: Copies of this document may be made for your own use and for distribution to others, provided that you do not charge any fee for such copies and further provided that each copy contains this Copyright Notice, whether distributed in print or electronically.
 

src/main/asciidoc/jpa.adoc

@@ -521,20 +521,45 @@ repo.findByAndSort("stark", Sort.by("LENGTH(firstname)"));            <2>
 repo.findByAndSort("targaryen", JpaSort.unsafe("LENGTH(firstname)")); <3>
 repo.findByAsArrayAndSort("bolton", Sort.by("fn_len"));               <4>
 ----
+
 <1> Valid `Sort` expression pointing to property in domain model.
-<2> Invalid `Sort` containing function call. Throws Exception.
+<2> Invalid `Sort` containing function call.
+Throws Exception.
 <3> Valid `Sort` containing explicitly _unsafe_ `Order`.
 <4> Valid `Sort` expression pointing to aliased function.
 ====
 
+[[jpa.query-methods.scroll]]
+=== Scrolling Large Query Results
+
+When working with large data sets, <<repositories.scrolling,scrolling>> can help to process those results efficiently without loading all results into memory.
+
+You have multiple options to consume large query results:
+
+1. <<repositories.paging-and-sorting,Paging>>.
+You have learned in the previous chapter about `Pageable` and `PageRequest`.
+2. <<repositories.scrolling.offset,Offset-based scrolling>>.
+This is a lighter variant than paging because it does not require the total result count.
+3. <<repositories.scrolling.keyset,Keyset-baset scrolling>>.
+This method avoids https://use-the-index-luke.com/no-offset[the shortcomings of offset-based result retrieval by leveraging database indexes].
+
+Read more on <<repositories.scrolling.guidance,which method to use best>> for your particular arrangement.
+
+You can use the Scroll API with query methods, <<query-by-example.running,Query-by-Example>>, and <<core.extensions.querydsl,Querydsl>>.
+
+NOTE: Scrolling with String-based query methods is not yet supported.
+Scrolling is also not supported using stored `@Procedure` query methods.
+
 [[jpa.named-parameters]]
 === Using Named Parameters
 
-By default, Spring Data JPA uses position-based parameter binding, as described in all the preceding examples. This makes query methods a little error-prone when refactoring regarding the parameter position. To solve this issue, you can use `@Param` annotation to give a method parameter a concrete name and bind the name in the query, as shown in the following example:
+By default, Spring Data JPA uses position-based parameter binding, as described in all the preceding examples.
+This makes query methods a little error-prone when refactoring regarding the parameter position.
+To solve this issue, you can use `@Param` annotation to give a method parameter a concrete name and bind the name in the query, as shown in the following example:
 
 .Using named parameters
 ====
-[source, java]
+[source,java]
 ----
 public interface UserRepository extends JpaRepository<User, Long> {