Introduce support for CassandraScrollPosition.

We now represent the PagingState as ScrollPosition to provide scrolling queries that are closer to Cassandra's internal pagination. In combination with Limit, queries are now as flexible as using the Statement API directly.

Closes #1408

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/query/CassandraPageRequest.java

@@ -16,6 +16,7 @@
 package org.springframework.data.cassandra.core.query;
 
 import java.nio.ByteBuffer;
+import java.util.Objects;
 
 import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Pageable;
@@ -39,15 +40,18 @@ import org.springframework.util.Assert;
  */
 public class CassandraPageRequest extends PageRequest {
 
-	private final @Nullable ByteBuffer pagingState;
+	private final CassandraScrollPosition scrollPosition;
 
 	private final boolean nextAllowed;
 
-	private CassandraPageRequest(int page, int size, Sort sort, @Nullable ByteBuffer pagingState, boolean nextAllowed) {
+	private CassandraPageRequest(int page, int size, Sort sort, CassandraScrollPosition scrollPosition,
+			boolean nextAllowed) {
 
 		super(page, size, sort);
 
-		this.pagingState = pagingState;
+		Assert.notNull(scrollPosition, "ScrollPosition must not be null");
+
+		this.scrollPosition = scrollPosition;
 		this.nextAllowed = nextAllowed;
 	}
 
@@ -79,7 +83,7 @@ public class CassandraPageRequest extends PageRequest {
 		Assert.isTrue(page == 0,
 				"Cannot create a Cassandra page request for an indexed page other than the first page (0)");
 
-		return new CassandraPageRequest(page, size, sort, null, false);
+		return new CassandraPageRequest(page, size, sort, CassandraScrollPosition.initial(), false);
 	}
 
 	/**
@@ -100,14 +104,15 @@ public class CassandraPageRequest extends PageRequest {
 	}
 
 	/**
-	 * Creates a a {@link PageRequest} with sort direction and properties applied.
+	 * Creates a {@link PageRequest} with sort direction and properties applied.
 	 *
 	 * @param current the current {@link Pageable}, must not be {@literal null}.
 	 * @param pagingState the paging state associated with the current {@link Pageable}. Can be {@literal null} if there
 	 *          is no paging state associated.
 	 */
 	public static CassandraPageRequest of(Pageable current, @Nullable ByteBuffer pagingState) {
-		return new CassandraPageRequest(current.getPageNumber(), current.getPageSize(), current.getSort(), pagingState,
+		return new CassandraPageRequest(current.getPageNumber(), current.getPageSize(), current.getSort(),
+				pagingState != null ? CassandraScrollPosition.of(pagingState) : CassandraScrollPosition.initial(),
 				pagingState != null);
 	}
 
@@ -127,7 +132,7 @@ public class CassandraPageRequest extends PageRequest {
 	 * @param sort must not be {@literal null}.
 	 */
 	public static CassandraPageRequest first(int size, Sort sort) {
-		return new CassandraPageRequest(0, size, sort, null, false);
+		return new CassandraPageRequest(0, size, sort, CassandraScrollPosition.initial(), false);
 	}
 
 	/**
@@ -178,11 +183,11 @@ public class CassandraPageRequest extends PageRequest {
 	@Nullable
 	public ByteBuffer getPagingState() {
 
-		if (this.pagingState == null) {
+		if (this.scrollPosition.isInitial()) {
 			return null;
 		}
 
-		return this.pagingState.asReadOnlyBuffer();
+		return this.scrollPosition.getPagingState();
 	}
 
 	/**
@@ -200,7 +205,7 @@ public class CassandraPageRequest extends PageRequest {
 
 		Assert.state(hasNext(), "Cannot create a next page request without a PagingState");
 
-		return new CassandraPageRequest(getPageNumber() + 1, getPageSize(), getSort(), getPagingState(), false);
+		return new CassandraPageRequest(getPageNumber() + 1, getPageSize(), getSort(), this.scrollPosition, false);
 	}
 
 	/**
@@ -214,7 +219,8 @@ public class CassandraPageRequest extends PageRequest {
 
 		Assert.notNull(sort, "Sort must not be null");
 
-		return new CassandraPageRequest(this.getPageNumber(), this.getPageSize(), sort, getPagingState(), this.nextAllowed);
+		return new CassandraPageRequest(this.getPageNumber(), this.getPageSize(), sort, this.scrollPosition,
+				this.nextAllowed);
 	}
 
 	@Override
@@ -225,6 +231,16 @@ public class CassandraPageRequest extends PageRequest {
 		return super.previous();
 	}
 
+	/**
+	 * Returns the underlying {@link CassandraScrollPosition}.
+	 *
+	 * @return the underlying {@link CassandraScrollPosition}.
+	 * @since 4.2
+	 */
+	public CassandraScrollPosition getScrollPosition() {
+		return this.scrollPosition;
+	}
+
 	@Override
 	public boolean equals(@Nullable Object obj) {
 
@@ -244,7 +260,7 @@ public class CassandraPageRequest extends PageRequest {
 			return false;
 		}
 
-		return (pagingState != null ? pagingState.equals(that.pagingState) : that.pagingState == null);
+		return Objects.equals(this.scrollPosition, that.scrollPosition);
 	}
 
 	@Override
@@ -252,7 +268,7 @@ public class CassandraPageRequest extends PageRequest {
 
 		int result = super.hashCode();
 
-		result = 31 * result + (pagingState != null ? pagingState.hashCode() : 0);
+		result = 31 * result + (scrollPosition != null ? scrollPosition.hashCode() : 0);
 		result = 31 * result + (nextAllowed ? 1 : 0);
 
 		return result;

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/query/CassandraScrollPosition.java

@@ -0,0 +1,116 @@
+/*
+ * 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.cassandra.core.query;
+
+import java.nio.ByteBuffer;
+
+import org.springframework.data.cassandra.core.query.CassandraScrollPosition.Initial;
+import org.springframework.data.cassandra.core.query.CassandraScrollPosition.PagingState;
+import org.springframework.data.domain.ScrollPosition;
+import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
+
+/**
+ * Cassandra-specific implementation of {@link ScrollPosition} using
+ * {@link com.datastax.oss.driver.api.core.cql.PagingState}.
+ *
+ * @author Mark Paluch
+ * @since 4.2
+ */
+public abstract sealed class CassandraScrollPosition implements ScrollPosition permits Initial,PagingState {
+
+	/**
+	 * Returns an initial {@link CassandraScrollPosition}.
+	 *
+	 * @return an initial {@link CassandraScrollPosition}.
+	 */
+	public static CassandraScrollPosition initial() {
+		return Initial.INSTANCE;
+	}
+
+	/**
+	 * Creates a continuation {@link CassandraScrollPosition} given {@code pagingState}.
+	 *
+	 * @return a continuation {@link CassandraScrollPosition} given {@code pagingState}.
+	 */
+	public static CassandraScrollPosition of(ByteBuffer pagingState) {
+
+		Assert.notNull(pagingState, "PagingState must not be null");
+
+		return new PagingState(pagingState);
+	}
+
+	/**
+	 * Returns the underlying binary representation of the paging state as read-only buffer if the scroll position is not
+	 * {@link #initial()}.
+	 *
+	 * @return the underlying binary representation of the paging state.
+	 * @throws IllegalStateException if the scroll position {@link #initial()} (i.e. the scroll position isn't associated
+	 *           with a continuation).
+	 */
+	public abstract ByteBuffer getPagingState();
+
+	static final class Initial extends CassandraScrollPosition {
+
+		private final static Initial INSTANCE = new Initial();
+
+		@Override
+		public boolean isInitial() {
+			return true;
+		}
+
+		@Override
+		public ByteBuffer getPagingState() {
+			throw new IllegalStateException("Initial scroll position does not provide a PagingState");
+		}
+	}
+
+	static final class PagingState extends CassandraScrollPosition {
+
+		private final ByteBuffer pagingState;
+
+		PagingState(ByteBuffer pagingState) {
+			this.pagingState = pagingState;
+		}
+
+		@Override
+		public boolean isInitial() {
+			return false;
+		}
+
+		@Override
+		public ByteBuffer getPagingState() {
+			return pagingState.asReadOnlyBuffer();
+		}
+
+		@Override
+		public boolean equals(Object o) {
+			if (this == o)
+				return true;
+			if (o == null || getClass() != o.getClass())
+				return false;
+
+			PagingState that = (PagingState) o;
+
+			return ObjectUtils.nullSafeEquals(pagingState, that.pagingState);
+		}
+
+		@Override
+		public int hashCode() {
+			return ObjectUtils.nullSafeHashCode(pagingState);
+		}
+	}
+}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/query/Query.java

@@ -47,7 +47,7 @@ import org.springframework.util.Assert;
 public class Query implements Filter {
 
 	private static final Query EMPTY = new Query(Collections.emptyList(), Columns.empty(), Sort.unsorted(),
-			Optional.empty(), Optional.empty(), Limit.unlimited(), false);
+			CassandraScrollPosition.initial(), Optional.empty(), Limit.unlimited(), false);
 
 	private final boolean allowFiltering;
 
@@ -57,21 +57,21 @@ public class Query implements Filter {
 
 	private final Limit limit;
 
-	private final Optional<ByteBuffer> pagingState;
+	private final CassandraScrollPosition scrollPosition;
 
 	private final Optional<QueryOptions> queryOptions;
 
 	private final Sort sort;
 
 	private Query(List<CriteriaDefinition> criteriaDefinitions, Columns columns, Sort sort,
-			Optional<ByteBuffer> pagingState, Optional<QueryOptions> queryOptions, Limit limit, boolean allowFiltering) {
+			CassandraScrollPosition pagingState, Optional<QueryOptions> queryOptions, Limit limit, boolean allowFiltering) {
 
 		Assert.notNull(limit, "Limit must not be null");
 
 		this.criteriaDefinitions = criteriaDefinitions;
 		this.columns = columns;
 		this.sort = sort;
-		this.pagingState = pagingState;
+		this.scrollPosition = pagingState;
 		this.queryOptions = queryOptions;
 		this.limit = limit;
 		this.allowFiltering = allowFiltering;
@@ -112,8 +112,8 @@ public class Query implements Filter {
 		List<CriteriaDefinition> collect = StreamSupport.stream(criteriaDefinitions.spliterator(), false)
 				.collect(Collectors.toList());
 
-		return new Query(collect, Columns.empty(), Sort.unsorted(), Optional.empty(), Optional.empty(), Limit.unlimited(),
-				false);
+		return new Query(collect, Columns.empty(), Sort.unsorted(), CassandraScrollPosition.initial(), Optional.empty(),
+				Limit.unlimited(), false);
 	}
 
 	/**
@@ -134,7 +134,7 @@ public class Query implements Filter {
 			criteriaDefinitions.add(criteriaDefinition);
 		}
 
-		return new Query(criteriaDefinitions, this.columns, this.sort, this.pagingState, this.queryOptions, this.limit,
+		return new Query(criteriaDefinitions, this.columns, this.sort, this.scrollPosition, this.queryOptions, this.limit,
 				this.allowFiltering);
 	}
 
@@ -154,7 +154,7 @@ public class Query implements Filter {
 
 		Assert.notNull(columns, "Columns must not be null");
 
-		return new Query(this.criteriaDefinitions, this.columns.and(columns), this.sort, this.pagingState,
+		return new Query(this.criteriaDefinitions, this.columns.and(columns), this.sort, this.scrollPosition,
 				this.queryOptions, this.limit, this.allowFiltering);
 	}
 
@@ -182,8 +182,8 @@ public class Query implements Filter {
 			}
 		}
 
-		return new Query(this.criteriaDefinitions, this.columns, this.sort.and(sort), this.pagingState, this.queryOptions,
-				this.limit, this.allowFiltering);
+		return new Query(this.criteriaDefinitions, this.columns, this.sort.and(sort), this.scrollPosition,
+				this.queryOptions, this.limit, this.allowFiltering);
 	}
 
 	/**
@@ -208,17 +208,31 @@ public class Query implements Filter {
 
 		CassandraPageRequest.validatePageable(pageable);
 
-		ByteBuffer pagingState = getPagingState().orElse(null);
+		CassandraScrollPosition scrollPosition = getScrollPosition();
 
-		if (pageable instanceof CassandraPageRequest) {
-			pagingState = ((CassandraPageRequest) pageable).getPagingState();
+		if (pageable instanceof CassandraPageRequest cpr) {
+			scrollPosition = cpr.getScrollPosition();
 		}
 
 		QueryOptions queryOptions = this.queryOptions.map(QueryOptions::mutate).orElse(QueryOptions.builder())
 				.pageSize(pageable.getPageSize()).build();
 
-		return new Query(this.criteriaDefinitions, this.columns, this.sort.and(pageable.getSort()),
-				Optional.ofNullable(pagingState), Optional.of(queryOptions), this.limit, this.allowFiltering);
+		return new Query(this.criteriaDefinitions, this.columns, this.sort.and(pageable.getSort()), scrollPosition,
+				Optional.of(queryOptions), this.limit, this.allowFiltering);
+	}
+
+	/**
+	 * Set the {@code paging state} to skip rows.
+	 *
+	 * @param scrollPosition must not be {@literal null}.
+	 * @return a new {@link Query} object containing the former settings with paging state applied.
+	 */
+	public Query pagingState(CassandraScrollPosition scrollPosition) {
+
+		Assert.notNull(scrollPosition, "CassandraScrollPosition must not be null");
+
+		return new Query(this.criteriaDefinitions, this.columns, this.sort, scrollPosition, this.queryOptions, this.limit,
+				this.allowFiltering);
 	}
 
 	/**
@@ -231,15 +245,20 @@ public class Query implements Filter {
 
 		Assert.notNull(pagingState, "PagingState must not be null");
 
-		return new Query(this.criteriaDefinitions, this.columns, this.sort, Optional.of(pagingState), this.queryOptions,
-				this.limit, this.allowFiltering);
+		return new Query(this.criteriaDefinitions, this.columns, this.sort, CassandraScrollPosition.of(pagingState),
+				this.queryOptions, this.limit, this.allowFiltering);
 	}
 
 	/**
 	 * @return the optional {@link ByteBuffer paging state}.
 	 */
 	public Optional<ByteBuffer> getPagingState() {
-		return this.pagingState.map(ByteBuffer::asReadOnlyBuffer);
+
+		if (this.scrollPosition.isInitial()) {
+			return Optional.empty();
+		}
+
+		return Optional.of(this.scrollPosition.getPagingState());
 	}
 
 	/**
@@ -252,7 +271,7 @@ public class Query implements Filter {
 
 		Assert.notNull(queryOptions, "QueryOptions must not be null");
 
-		return new Query(this.criteriaDefinitions, this.columns, this.sort, this.pagingState, Optional.of(queryOptions),
+		return new Query(this.criteriaDefinitions, this.columns, this.sort, this.scrollPosition, Optional.of(queryOptions),
 				this.limit, this.allowFiltering);
 	}
 
@@ -270,7 +289,7 @@ public class Query implements Filter {
 	 * @return a new {@link Query} object containing the former settings with {@code limit} applied.
 	 */
 	public Query limit(long limit) {
-		return new Query(this.criteriaDefinitions, this.columns, this.sort, this.pagingState, this.queryOptions,
+		return new Query(this.criteriaDefinitions, this.columns, this.sort, this.scrollPosition, this.queryOptions,
 				Limit.of(Math.toIntExact(limit)), this.allowFiltering);
 	}
 
@@ -281,7 +300,7 @@ public class Query implements Filter {
 	 * @return a new {@link Query} object containing the former settings with {@code limit} applied.
 	 */
 	public Query limit(Limit limit) {
-		return new Query(this.criteriaDefinitions, this.columns, this.sort, this.pagingState, this.queryOptions, limit,
+		return new Query(this.criteriaDefinitions, this.columns, this.sort, this.scrollPosition, this.queryOptions, limit,
 				this.allowFiltering);
 	}
 
@@ -305,8 +324,8 @@ public class Query implements Filter {
 	 * @return a new {@link Query} object containing the former settings with {@code allowFiltering} applied.
 	 */
 	public Query withAllowFiltering() {
-		return new Query(this.criteriaDefinitions, this.columns, this.sort, this.pagingState, this.queryOptions, this.limit,
-				true);
+		return new Query(this.criteriaDefinitions, this.columns, this.sort, this.scrollPosition, this.queryOptions,
+				this.limit, true);
 	}
 
 	/**
@@ -316,6 +335,14 @@ public class Query implements Filter {
 		return this.allowFiltering;
 	}
 
+	/**
+	 * @return the paging state as {@link CassandraScrollPosition}.
+	 * @since 4.2
+	 */
+	private CassandraScrollPosition getScrollPosition() {
+		return scrollPosition;
+	}
+
 	@Override
 	public boolean equals(@Nullable Object obj) {
 
@@ -341,7 +368,7 @@ public class Query implements Filter {
 		boolean criteriaEqual = this.criteriaDefinitions.equals(that.criteriaDefinitions);
 		boolean columnsEqual = nullSafeEquals(this.columns, that.columns);
 		boolean sortEqual = nullSafeEquals(this.sort, that.sort);
-		boolean pagingStateEqual = nullSafeEquals(this.pagingState, that.pagingState);
+		boolean pagingStateEqual = nullSafeEquals(this.scrollPosition, that.scrollPosition);
 		boolean queryOptionsEqual = nullSafeEquals(this.queryOptions, that.queryOptions);
 		boolean limitEqual = this.limit == that.limit;
 		boolean allowFilteringEqual = this.allowFiltering == that.allowFiltering;
@@ -356,7 +383,7 @@ public class Query implements Filter {
 		result += 31 * criteriaDefinitions.hashCode();
 		result += 31 * nullSafeHashCode(columns);
 		result += 31 * nullSafeHashCode(sort);
-		result += 31 * nullSafeHashCode(pagingState);
+		result += 31 * nullSafeHashCode(scrollPosition);
 		result += 31 * nullSafeHashCode(queryOptions);
 		result += 31 * nullSafeHashCode(limit);
 		result += (allowFiltering ? 0 : 1);

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractCassandraQuery.java

@@ -25,6 +25,7 @@ import org.springframework.data.cassandra.repository.query.CassandraQueryExecuti
 import org.springframework.data.cassandra.repository.query.CassandraQueryExecution.SingleEntityExecution;
 import org.springframework.data.cassandra.repository.query.CassandraQueryExecution.SlicedExecution;
 import org.springframework.data.cassandra.repository.query.CassandraQueryExecution.StreamExecution;
+import org.springframework.data.cassandra.repository.query.CassandraQueryExecution.WindowExecution;
 import org.springframework.data.repository.query.ParameterAccessor;
 import org.springframework.data.repository.query.RepositoryQuery;
 import org.springframework.data.repository.query.ResultProcessor;
@@ -118,6 +119,8 @@ public abstract class AbstractCassandraQuery extends CassandraRepositoryQuerySup
 
 		if (getQueryMethod().isSliceQuery()) {
 			return new SlicedExecution(getOperations(), parameterAccessor.getPageable());
+		} else if (getQueryMethod().isScrollQuery()) {
+			return new WindowExecution(getOperations(), parameterAccessor.getScrollPosition(), parameterAccessor.getLimit());
 		} else if (getQueryMethod().isCollectionQuery()) {
 			return new CollectionExecution(getOperations());
 		} else if (getQueryMethod().isResultSetQuery()) {

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractReactiveCassandraQuery.java

@@ -28,6 +28,7 @@ import org.springframework.data.cassandra.repository.query.ReactiveCassandraQuer
 import org.springframework.data.cassandra.repository.query.ReactiveCassandraQueryExecution.ResultProcessingExecution;
 import org.springframework.data.cassandra.repository.query.ReactiveCassandraQueryExecution.SingleEntityExecution;
 import org.springframework.data.cassandra.repository.query.ReactiveCassandraQueryExecution.SlicedExecution;
+import org.springframework.data.cassandra.repository.query.ReactiveCassandraQueryExecution.WindowExecution;
 import org.springframework.data.repository.query.ParameterAccessor;
 import org.springframework.data.repository.query.RepositoryQuery;
 import org.springframework.data.repository.query.ResultProcessor;
@@ -122,6 +123,9 @@ public abstract class AbstractReactiveCassandraQuery extends CassandraRepository
 
 		if (getQueryMethod().isSliceQuery()) {
 			return new SlicedExecution(getReactiveCassandraOperations(), parameterAccessor.getPageable());
+		} else if (getQueryMethod().isScrollQuery()) {
+			return new WindowExecution(getReactiveCassandraOperations(), parameterAccessor.getScrollPosition(),
+					parameterAccessor.getLimit());
 		} else if (getQueryMethod().isCollectionQuery()) {
 			return new CollectionExecution(getReactiveCassandraOperations());
 		} else if (isCountQuery()) {

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraParameterAccessor.java

@@ -17,6 +17,7 @@ package org.springframework.data.cassandra.repository.query;
 
 import org.springframework.data.cassandra.core.cql.QueryOptions;
 import org.springframework.data.cassandra.core.mapping.CassandraType;
+import org.springframework.data.cassandra.core.query.CassandraScrollPosition;
 import org.springframework.data.repository.query.ParameterAccessor;
 import org.springframework.lang.Nullable;
 
@@ -71,6 +72,9 @@ public interface CassandraParameterAccessor extends ParameterAccessor {
 	 */
 	Object[] getValues();
 
+	@Override
+	CassandraScrollPosition getScrollPosition();
+
 	/**
 	 * Returns the {@link QueryOptions} associated with the associated Repository query method.
 	 *

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraParametersParameterAccessor.java

@@ -18,7 +18,9 @@ package org.springframework.data.cassandra.repository.query;
 import org.springframework.data.cassandra.core.cql.QueryOptions;
 import org.springframework.data.cassandra.core.mapping.CassandraSimpleTypeHolder;
 import org.springframework.data.cassandra.core.mapping.CassandraType;
+import org.springframework.data.cassandra.core.query.CassandraScrollPosition;
 import org.springframework.data.domain.Limit;
+import org.springframework.data.domain.ScrollPosition;
 import org.springframework.data.repository.query.ParameterAccessor;
 import org.springframework.data.repository.query.ParametersParameterAccessor;
 import org.springframework.lang.Nullable;
@@ -57,6 +59,7 @@ public class CassandraParametersParameterAccessor extends ParametersParameterAcc
 	}
 
 	@Nullable
+	@Override
 	public CassandraType findCassandraType(int index) {
 		return getParameters().getParameter(index).getCassandraType();
 	}
@@ -76,6 +79,22 @@ public class CassandraParametersParameterAccessor extends ParametersParameterAcc
 		return super.getValues();
 	}
 
+	@Override
+	public CassandraScrollPosition getScrollPosition() {
+
+		ScrollPosition scrollPosition = super.getScrollPosition();
+		if (scrollPosition instanceof CassandraScrollPosition csp) {
+			return csp;
+		}
+
+		if (scrollPosition == null) {
+			return CassandraScrollPosition.initial();
+		}
+
+		throw new IllegalArgumentException(
+				"Unsupported scroll position " + scrollPosition + ". Only CassandraScrollPosition supported.");
+	}
+
 	@Override
 	public Limit getLimit() {
 

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraQueryExecution.java

@@ -23,7 +23,9 @@ import org.springframework.data.cassandra.core.CassandraOperations;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
 import org.springframework.data.cassandra.core.query.CassandraPageRequest;
+import org.springframework.data.cassandra.core.query.CassandraScrollPosition;
 import org.springframework.data.convert.DtoInstantiatingConverter;
+import org.springframework.data.domain.Limit;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Slice;
 import org.springframework.data.domain.SliceImpl;
@@ -108,6 +110,37 @@ interface CassandraQueryExecution {
 		}
 	}
 
+	/**
+	 * {@link CassandraQueryExecution} for a {@link org.springframework.data.domain.Window}.
+	 *
+	 * @author Mark Paluch
+	 * @since 4.2
+	 */
+	final class WindowExecution implements CassandraQueryExecution {
+
+		private final CassandraOperations operations;
+		private final CassandraScrollPosition scrollPosition;
+		private final Limit limit;
+
+		public WindowExecution(CassandraOperations operations, CassandraScrollPosition scrollPosition, Limit limit) {
+			this.operations = operations;
+			this.scrollPosition = scrollPosition;
+			this.limit = limit;
+		}
+
+		@Override
+		public Object execute(Statement<?> statement, Class<?> type) {
+
+			Statement<?> statementToUse = limit.isLimited() ? statement.setPageSize(limit.max()) : statement;
+
+			if (!this.scrollPosition.isInitial()) {
+				statementToUse = statementToUse.setPagingState(this.scrollPosition.getPagingState());
+			}
+
+			return WindowUtil.of(operations.slice(statementToUse, type));
+		}
+	}
+
 	/**
 	 * {@link CassandraQueryExecution} for collection returning queries.
 	 *
@@ -197,7 +230,7 @@ interface CassandraQueryExecution {
 	}
 
 	/**
-	 * {@link CassandraQueryExecution} to return a {@link ResultSet}.
+	 * {@link CassandraQueryExecution} to return a {@link com.datastax.oss.driver.api.core.cql.ResultSet}.
 	 *
 	 * @author Mark Paluch
 	 */

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ConvertingParameterAccessor.java

@@ -20,10 +20,10 @@ import java.util.Iterator;
 import org.springframework.data.cassandra.core.convert.CassandraConverter;
 import org.springframework.data.cassandra.core.cql.QueryOptions;
 import org.springframework.data.cassandra.core.mapping.CassandraType;
+import org.springframework.data.cassandra.core.query.CassandraScrollPosition;
 import org.springframework.data.domain.Limit;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Range;
-import org.springframework.data.domain.ScrollPosition;
 import org.springframework.data.domain.Sort;
 import org.springframework.lang.Nullable;
 
@@ -49,9 +49,8 @@ class ConvertingParameterAccessor implements CassandraParameterAccessor {
 		this.delegate = delegate;
 	}
 
-	@Nullable
 	@Override
-	public ScrollPosition getScrollPosition() {
+	public CassandraScrollPosition getScrollPosition() {
 		return delegate.getScrollPosition();
 	}
 

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/QueryStatementCreator.java

@@ -15,7 +15,6 @@
  */
 package org.springframework.data.cassandra.repository.query;
 
-
 import java.util.Optional;
 import java.util.function.Function;
 
@@ -24,6 +23,7 @@ import org.apache.commons.logging.LogFactory;
 import org.springframework.data.cassandra.core.StatementFactory;
 import org.springframework.data.cassandra.core.cql.QueryExtractorDelegate;
 import org.springframework.data.cassandra.core.cql.QueryOptions;
+import org.springframework.data.cassandra.core.cql.QueryOptions.QueryOptionsBuilder;
 import org.springframework.data.cassandra.core.cql.QueryOptionsUtil;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
@@ -196,13 +196,16 @@ class QueryStatementCreator {
 
 		try {
 
-			if (tree.isLimiting()) {
-				query = query.limit(tree.getMaxResults());
-			}
-
 			Limit limit = parameterAccessor.getLimit();
-			if (limit.isLimited()) {
-				query = query.limit(limit);
+
+			if (!queryMethod.isScrollQuery()) {
+
+				if (limit.isLimited()) {
+					query = query.limit(limit);
+				}
+				if (tree.isLimiting()) {
+					query = query.limit(tree.getMaxResults());
+				}
 			}
 
 			if (allowsFiltering()) {
@@ -210,14 +213,23 @@ class QueryStatementCreator {
 			}
 
 			Optional<QueryOptions> queryOptions = Optional.ofNullable(parameterAccessor.getQueryOptions());
+			QueryOptionsBuilder optionsBuilder = queryOptions.orElseGet(QueryOptions::empty).mutate();
 
-			if (queryOptions.isPresent()) {
-				query = Optional.ofNullable(parameterAccessor.getQueryOptions()).map(query::queryOptions).orElse(query);
-			} else if (this.queryMethod.hasConsistencyLevel()) {
-				query = query.queryOptions(
-						QueryOptions.builder().consistencyLevel(this.queryMethod.getRequiredAnnotatedConsistencyLevel()).build());
+			if (queryMethod.isScrollQuery()) {
+
+				if (limit.isLimited()) {
+					optionsBuilder.pageSize(limit.max());
+				} else if (tree.isLimiting()) {
+					optionsBuilder.pageSize(tree.getMaxResults());
+				}
 			}
 
+			if (this.queryMethod.hasConsistencyLevel()) {
+				optionsBuilder.consistencyLevel(this.queryMethod.getRequiredAnnotatedConsistencyLevel());
+			}
+
+			query = query.queryOptions(optionsBuilder.build());
+
 			return function.apply(query);
 		} catch (RuntimeException cause) {
 			throw QueryCreationException.create(this.queryMethod, cause);

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraQueryExecution.java

@@ -27,7 +27,9 @@ import org.springframework.data.cassandra.core.ReactiveCassandraOperations;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
 import org.springframework.data.cassandra.core.query.CassandraPageRequest;
+import org.springframework.data.cassandra.core.query.CassandraScrollPosition;
 import org.springframework.data.convert.DtoInstantiatingConverter;
+import org.springframework.data.domain.Limit;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Slice;
 import org.springframework.data.domain.SliceImpl;
@@ -94,6 +96,40 @@ interface ReactiveCassandraQueryExecution {
 		}
 	}
 
+	/**
+	 * {@link ReactiveCassandraQueryExecution} for a {@link org.springframework.data.domain.Window}.
+	 *
+	 * @author Mark Paluch
+	 * @since 4.2
+	 */
+	final class WindowExecution implements ReactiveCassandraQueryExecution {
+
+		private final ReactiveCassandraOperations operations;
+		private final CassandraScrollPosition scrollPosition;
+		private final Limit limit;
+
+		public WindowExecution(ReactiveCassandraOperations operations, CassandraScrollPosition scrollPosition,
+				Limit limit) {
+			this.operations = operations;
+			this.scrollPosition = scrollPosition;
+			this.limit = limit;
+		}
+
+		@Override
+		public Publisher<? extends Object> execute(Statement<?> statement, Class<?> type) {
+
+			Statement<?> statementToUse = limit.isLimited() ? statement.setPageSize(limit.max()) : statement;
+
+			if (!this.scrollPosition.isInitial()) {
+				statementToUse = statementToUse.setPagingState(this.scrollPosition.getPagingState());
+			}
+
+			Mono<? extends Slice<?>> slice = operations.slice(statementToUse, type);
+
+			return slice.map(WindowUtil::of);
+		}
+	}
+
 	/**
 	 * {@link ReactiveCassandraQueryExecution} for collection returning queries.
 	 *

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/WindowUtil.java

@@ -0,0 +1,38 @@
+/*
+ * 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.cassandra.repository.query;
+
+import java.util.List;
+
+import org.springframework.data.cassandra.core.query.CassandraPageRequest;
+import org.springframework.data.domain.Slice;
+import org.springframework.data.domain.Window;
+
+/**
+ * Utility to create {@link org.springframework.data.domain.Window} objects.
+ *
+ * @author Mark Paluch
+ */
+class WindowUtil {
+
+	static <T> Window<T> of(Slice<T> slice) {
+
+		List<T> content = slice.getContent();
+		CassandraPageRequest pageable = (CassandraPageRequest) slice.getPageable();
+
+		return Window.from(content, value -> pageable.getScrollPosition(), slice.hasNext());
+	}
+}

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/QueryDerivationIntegrationTests.java

@@ -43,6 +43,7 @@ import org.springframework.data.cassandra.core.mapping.Embedded;
 import org.springframework.data.cassandra.core.mapping.Indexed;
 import org.springframework.data.cassandra.core.mapping.Table;
 import org.springframework.data.cassandra.core.query.CassandraPageRequest;
+import org.springframework.data.cassandra.core.query.CassandraScrollPosition;
 import org.springframework.data.cassandra.domain.AddressType;
 import org.springframework.data.cassandra.domain.Person;
 import org.springframework.data.cassandra.repository.QueryDerivationIntegrationTests.PersonRepository.NumberOfChildren;
@@ -54,8 +55,12 @@ import org.springframework.data.cassandra.repository.support.IntegrationTestConf
 import org.springframework.data.cassandra.support.CassandraVersion;
 import org.springframework.data.domain.Limit;
 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.support.WindowIterator;
+import org.springframework.data.util.Streamable;
 import org.springframework.data.util.Version;
 import org.springframework.lang.Nullable;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
@@ -341,6 +346,49 @@ class QueryDerivationIntegrationTests extends AbstractSpringDataEmbeddedCassandr
 		assertThat(result).contains(walter, skyler, flynn);
 	}
 
+	@Test // GH-1408
+	public void shouldSelectWindow() {
+
+		List<Person> result = new ArrayList<>();
+
+		Window<Person> firstWindow = personRepository.findAllWindowByLastname("White", CassandraScrollPosition.initial(),
+				Limit.of(2));
+		Window<Person> nextWindow = personRepository.findAllWindowByLastname("White",
+				firstWindow.positionAt(firstWindow.size() - 1), Limit.of(10));
+
+		result.addAll(firstWindow.getContent());
+		result.addAll(nextWindow.getContent());
+
+		assertThat(firstWindow).hasSize(2);
+
+		assertThat(nextWindow).hasSize(1);
+		assertThat(result).contains(walter, skyler, flynn);
+
+		WindowIterator<Person> iterator = WindowIterator
+				.of(scrollPosition -> personRepository.findAllWindowByLastname("White", scrollPosition, Limit.of(2)))
+				.startingAt(CassandraScrollPosition.initial());
+
+		List<Person> people = Streamable.of(() -> iterator).toList();
+		assertThat(people).containsOnly(walter, skyler, flynn);
+	}
+
+	@Test // GH-1408
+	public void shouldSelectWindowWithTopKeyword() {
+
+		List<Person> result = new ArrayList<>();
+
+		Window<Person> firstWindow = personRepository.findTop2ByLastname("White", CassandraScrollPosition.initial());
+		Window<Person> nextWindow = personRepository.findTop2ByLastname("White",
+				firstWindow.positionAt(firstWindow.size() - 1));
+
+		result.addAll(firstWindow.getContent());
+		result.addAll(nextWindow.getContent());
+
+		assertThat(firstWindow).hasSize(2);
+		assertThat(nextWindow).hasSize(1);
+		assertThat(result).contains(walter, skyler, flynn);
+	}
+
 	@Test // GH-1407
 	public void shouldSelectWithLimit() {
 
@@ -436,6 +484,10 @@ class QueryDerivationIntegrationTests extends AbstractSpringDataEmbeddedCassandr
 
 		Slice<Person> findAllSlicedByLastname(String lastname, Pageable pageable);
 
+		Window<Person> findAllWindowByLastname(String lastname, ScrollPosition scrollPosition, Limit limit);
+
+		Window<Person> findTop2ByLastname(String lastname, ScrollPosition scrollPosition);
+
 		List<Person> findAllLimitedByLastname(String lastname, Limit limit);
 
 		Collection<PersonProjection> findPersonProjectedBy();

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/ReactiveCassandraRepositoryIntegrationTests.java

@@ -15,12 +15,17 @@
  */
 package org.springframework.data.cassandra.repository;
 
+import static org.assertj.core.api.Assertions.*;
+
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
+import java.time.Duration;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
 
 import org.junit.jupiter.api.BeforeEach;
@@ -35,6 +40,7 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
 import org.springframework.data.cassandra.core.ReactiveCassandraOperations;
 import org.springframework.data.cassandra.core.query.CassandraPageRequest;
+import org.springframework.data.cassandra.core.query.CassandraScrollPosition;
 import org.springframework.data.cassandra.domain.Group;
 import org.springframework.data.cassandra.domain.GroupKey;
 import org.springframework.data.cassandra.domain.User;
@@ -44,8 +50,10 @@ import org.springframework.data.cassandra.repository.support.ReactiveCassandraRe
 import org.springframework.data.cassandra.repository.support.SimpleReactiveCassandraRepository;
 import org.springframework.data.domain.Limit;
 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.util.Streamable;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
@@ -130,7 +138,47 @@ class ReactiveCassandraRepositoryIntegrationTests extends AbstractSpringDataEmbe
 		repository.findByLastname(dave.getLastname()).as(StepVerifier::create).expectNextCount(2).verifyComplete();
 	}
 
-	@Test // DATACASS-529
+	@Test // GH-1408
+	void shouldSelectWindow() {
+
+		List<User> result = new ArrayList<>();
+
+		Window<User> firstWindow = repository
+				.findAllWindowByLastname("Matthews", CassandraScrollPosition.initial(), Limit.of(1))
+				.block(Duration.ofSeconds(10));
+
+		Window<User> nextWindow = repository
+				.findAllWindowByLastname("Matthews", firstWindow.positionAt(firstWindow.size() - 1), Limit.of(10))
+				.block(Duration.ofSeconds(10));
+
+		result.addAll(firstWindow.getContent());
+		result.addAll(nextWindow.getContent());
+
+		assertThat(firstWindow).hasSize(1);
+		assertThat(nextWindow).hasSize(1);
+		assertThat(result).contains(dave, oliver);
+	}
+
+	@Test // GH-1408
+	void shouldSelectWindowWithTopKeyword() {
+
+		List<User> result = new ArrayList<>();
+
+		Window<User> firstWindow = repository.findTop1ByLastname("Matthews", CassandraScrollPosition.initial())
+				.block(Duration.ofSeconds(10));
+
+		Window<User> nextWindow = repository.findTop1ByLastname("Matthews", firstWindow.positionAt(firstWindow.size() - 1))
+				.block(Duration.ofSeconds(10));
+
+		result.addAll(firstWindow.getContent());
+		result.addAll(nextWindow.getContent());
+
+		assertThat(firstWindow).hasSize(1);
+		assertThat(nextWindow).hasSize(1);
+		assertThat(result).contains(dave, oliver);
+	}
+
+	@Test // GH-1408
 	void shouldFindSliceByLastName() {
 		repository.findByLastname(carter.getLastname(), CassandraPageRequest.first(1)).as(StepVerifier::create)
 				.expectNextMatches(users -> users.getSize() == 1 && users.hasNext()).verifyComplete();
@@ -248,6 +296,10 @@ class ReactiveCassandraRepositoryIntegrationTests extends AbstractSpringDataEmbe
 
 		Mono<Slice<User>> findByLastname(String firstname, Pageable pageable);
 
+		Mono<Window<User>> findAllWindowByLastname(String lastname, ScrollPosition scrollPosition, Limit limit);
+
+		Mono<Window<User>> findTop1ByLastname(String lastname, ScrollPosition scrollPosition);
+
 		Flux<User> findByLastname(String lastname, Limit limit);
 
 		Mono<User> findFirstByLastname(String lastname);

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/StubParameterAccessor.java

@@ -21,8 +21,8 @@ import java.util.Iterator;
 import org.springframework.data.cassandra.core.convert.CassandraConverter;
 import org.springframework.data.cassandra.core.cql.QueryOptions;
 import org.springframework.data.cassandra.core.mapping.CassandraType;
+import org.springframework.data.cassandra.core.query.CassandraScrollPosition;
 import org.springframework.data.domain.Pageable;
-import org.springframework.data.domain.ScrollPosition;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.repository.query.ParameterAccessor;
 import org.springframework.lang.Nullable;
@@ -73,7 +73,7 @@ class StubParameterAccessor implements CassandraParameterAccessor {
 
 	@Nullable
 	@Override
-	public ScrollPosition getScrollPosition() {
+	public CassandraScrollPosition getScrollPosition() {
 		return null;
 	}
 

src/main/asciidoc/reference/cassandra-repositories.adoc

@@ -151,7 +151,7 @@ public class PersonRepositoryTests {
 
       assertThat(firstBatch).hasSize(10);
 
-      Page<Person> nextBatch = repository.findAll(firstBatch.nextPageable());
+      Slice<Person> nextBatch = repository.findAll(firstBatch.nextPageable());
 
       // …
     }
@@ -182,20 +182,22 @@ public interface PersonRepository extends CrudRepository<Person, String> {
 
     Slice<Person> findByFirstname(String firstname, Pageable pageRequest);  <2>
 
-    List<Person> findByFirstname(String firstname, QueryOptions opts);      <3>
+    Window<Person> findByFirstname(String firstname, CassandraScrollPosition pos, Limit limit);  <3>
 
-    List<Person> findByFirstname(String firstname, Sort sort);              <4>
+    List<Person> findByFirstname(String firstname, QueryOptions opts);      <4>
 
-    List<Person> findByFirstname(String firstname, Limit limit);            <5>
+    List<Person> findByFirstname(String firstname, Sort sort);              <5>
 
-    Person findByShippingAddress(Address address);                          <6>
+    List<Person> findByFirstname(String firstname, Limit limit);            <6>
 
-    Person findFirstByShippingAddress(Address address);                     <7>
+    Person findByShippingAddress(Address address);                          <7>
 
-    Stream<Person> findAllBy();                                             <8>
+    Person findFirstByShippingAddress(Address address);                     <8>
+
+    Stream<Person> findAllBy();                                             <9>
 
     @AllowFiltering
-    List<Person> findAllByAge(int age);                                     <9>
+    List<Person> findAllByAge(int age);                                     <10>
 }
 ----
 
@@ -204,17 +206,20 @@ The query is derived from parsing the method name for constraints, which can be
 Thus, the method name results in a query expression of `SELECT * FROM person WHERE lastname = 'lastname'`.
 <2> Applies pagination to a query.
 You can equip your method signature with a `Pageable` parameter and let the method return a `Slice` instance, and we automatically page the query accordingly.
-<3> Passing a `QueryOptions` object applies the query options to the resulting query before its execution.
-<4> Applies dynamic sorting to a query.
+<3> Applies scrolling to a query.
+Scrolling wraps Cassandra's `PagingState` into `CassandraScrollPosition` and allows dynamic limiting.
+You can also use `findTop…` for a static limit.
+<4> Passing a `QueryOptions` object applies the query options to the resulting query before its execution.
+<5> Applies dynamic sorting to a query.
 You can add a `Sort` parameter to your method signature, and Spring Data automatically applies ordering to the query.
-<5> Applies dynamic result limiting to a query.
+<6> Applies dynamic result limiting to a query.
 Query results can be limited using `SELECT … LIMIT`.
-<6> Shows that you can query based on properties that are not a primitive type by using `Converter` instances registered in `CustomConversions`.
+<7> Shows that you can query based on properties that are not a primitive type by using `Converter` instances registered in `CustomConversions`.
 Throws `IncorrectResultSizeDataAccessException` if more than one match is found.
-<7> Uses the `First` keyword to restrict the query to only the first result.
+<8> Uses the `First` keyword to restrict the query to only the first result.
 Unlike the preceding method, this method does not throw an exception if more than one match is found.
-<8> Uses a Java 8 `Stream` to read and convert individual elements while iterating the stream.
-<9> Shows a query method annotated with `@AllowFiltering`, to allow server-side filtering.
+<9> Uses a Java 8 `Stream` to read and convert individual elements while iterating the stream.
+<10> Shows a query method annotated with `@AllowFiltering`, to allow server-side filtering.
 ====
 
 NOTE: Querying non-primary key properties requires secondary indexes.