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

Author: mp911de

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 @@
  */
 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 static CassandraPageRequest of(int page, int size, Sort sort) {
 		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 static CassandraPageRequest of(int page, int size, Direction direction, S
 	}
 
 	/**
-	 * 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 static CassandraPageRequest first(int size) {
 	 * @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 static void validatePageable(Pageable pageable) {
 	@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 CassandraPageRequest next() {
 
 		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 CassandraPageRequest withSort(Sort sort) {
 
 		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 PageRequest previous() {
 		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,15 +260,15 @@ public boolean equals(@Nullable Object obj) {
 			return false;
 		}
 
-		return (pagingState != null ? pagingState.equals(that.pagingState) : that.pagingState == null);
+		return Objects.equals(this.scrollPosition, that.scrollPosition);
 	}
 
 	@Override
 	public int hashCode() {
 
 		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);
+		}
+	}
+}