diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/query/CassandraPageRequest.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/query/CassandraPageRequest.java index 4458a1da4..eba46fa2c 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/query/CassandraPageRequest.java +++ b/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; diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/query/CassandraScrollPosition.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/query/CassandraScrollPosition.java new file mode 100644 index 000000000..84338d60a --- /dev/null +++ b/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); + } + } +} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/query/Query.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/query/Query.java index 3b6a65ceb..9698c1b89 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/query/Query.java +++ b/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 pagingState; + private final CassandraScrollPosition scrollPosition; private final Optional queryOptions; private final Sort sort; private Query(List criteriaDefinitions, Columns columns, Sort sort, - Optional pagingState, Optional queryOptions, Limit limit, boolean allowFiltering) { + CassandraScrollPosition pagingState, Optional 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 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 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); diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractCassandraQuery.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractCassandraQuery.java index 097ccdee3..d54f04435 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractCassandraQuery.java +++ b/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()) { diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractReactiveCassandraQuery.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractReactiveCassandraQuery.java index d9f3851fa..5f83d5749 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractReactiveCassandraQuery.java +++ b/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()) { diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraParameterAccessor.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraParameterAccessor.java index 6e297cbe8..cba94df93 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraParameterAccessor.java +++ b/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. * diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraParametersParameterAccessor.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraParametersParameterAccessor.java index 6e4b12a0e..6d0780257 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraParametersParameterAccessor.java +++ b/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() { diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraQueryExecution.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraQueryExecution.java index b2dc56e4a..b92bd8bc4 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraQueryExecution.java +++ b/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 */ diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ConvertingParameterAccessor.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ConvertingParameterAccessor.java index cbc1cd777..588fdb9b6 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ConvertingParameterAccessor.java +++ b/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(); } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/QueryStatementCreator.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/QueryStatementCreator.java index a3b2f9369..708c4c24f 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/QueryStatementCreator.java +++ b/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 = 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); diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraQueryExecution.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraQueryExecution.java index 99bc79266..3abef2744 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraQueryExecution.java +++ b/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 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> slice = operations.slice(statementToUse, type); + + return slice.map(WindowUtil::of); + } + } + /** * {@link ReactiveCassandraQueryExecution} for collection returning queries. * diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/WindowUtil.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/WindowUtil.java new file mode 100644 index 000000000..521874641 --- /dev/null +++ b/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 Window of(Slice slice) { + + List content = slice.getContent(); + CassandraPageRequest pageable = (CassandraPageRequest) slice.getPageable(); + + return Window.from(content, value -> pageable.getScrollPosition(), slice.hasNext()); + } +} diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/QueryDerivationIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/QueryDerivationIntegrationTests.java index 437216178..d0658fa68 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/QueryDerivationIntegrationTests.java +++ b/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 result = new ArrayList<>(); + + Window firstWindow = personRepository.findAllWindowByLastname("White", CassandraScrollPosition.initial(), + Limit.of(2)); + Window 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 iterator = WindowIterator + .of(scrollPosition -> personRepository.findAllWindowByLastname("White", scrollPosition, Limit.of(2))) + .startingAt(CassandraScrollPosition.initial()); + + List people = Streamable.of(() -> iterator).toList(); + assertThat(people).containsOnly(walter, skyler, flynn); + } + + @Test // GH-1408 + public void shouldSelectWindowWithTopKeyword() { + + List result = new ArrayList<>(); + + Window firstWindow = personRepository.findTop2ByLastname("White", CassandraScrollPosition.initial()); + Window 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 findAllSlicedByLastname(String lastname, Pageable pageable); + Window findAllWindowByLastname(String lastname, ScrollPosition scrollPosition, Limit limit); + + Window findTop2ByLastname(String lastname, ScrollPosition scrollPosition); + List findAllLimitedByLastname(String lastname, Limit limit); Collection findPersonProjectedBy(); diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/ReactiveCassandraRepositoryIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/ReactiveCassandraRepositoryIntegrationTests.java index 6c7110534..df6dd989e 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/ReactiveCassandraRepositoryIntegrationTests.java +++ b/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 result = new ArrayList<>(); + + Window firstWindow = repository + .findAllWindowByLastname("Matthews", CassandraScrollPosition.initial(), Limit.of(1)) + .block(Duration.ofSeconds(10)); + + Window 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 result = new ArrayList<>(); + + Window firstWindow = repository.findTop1ByLastname("Matthews", CassandraScrollPosition.initial()) + .block(Duration.ofSeconds(10)); + + Window 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> findByLastname(String firstname, Pageable pageable); + Mono> findAllWindowByLastname(String lastname, ScrollPosition scrollPosition, Limit limit); + + Mono> findTop1ByLastname(String lastname, ScrollPosition scrollPosition); + Flux findByLastname(String lastname, Limit limit); Mono findFirstByLastname(String lastname); diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/StubParameterAccessor.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/StubParameterAccessor.java index e514743dc..4a50a68da 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/StubParameterAccessor.java +++ b/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; } diff --git a/src/main/asciidoc/reference/cassandra-repositories.adoc b/src/main/asciidoc/reference/cassandra-repositories.adoc index 01e7d53d7..0387b6c0a 100644 --- a/src/main/asciidoc/reference/cassandra-repositories.adoc +++ b/src/main/asciidoc/reference/cassandra-repositories.adoc @@ -151,7 +151,7 @@ public class PersonRepositoryTests { assertThat(firstBatch).hasSize(10); - Page nextBatch = repository.findAll(firstBatch.nextPageable()); + Slice nextBatch = repository.findAll(firstBatch.nextPageable()); // … } @@ -182,20 +182,22 @@ public interface PersonRepository extends CrudRepository { Slice findByFirstname(String firstname, Pageable pageRequest); <2> - List findByFirstname(String firstname, QueryOptions opts); <3> + Window findByFirstname(String firstname, CassandraScrollPosition pos, Limit limit); <3> - List findByFirstname(String firstname, Sort sort); <4> + List findByFirstname(String firstname, QueryOptions opts); <4> - List findByFirstname(String firstname, Limit limit); <5> + List findByFirstname(String firstname, Sort sort); <5> - Person findByShippingAddress(Address address); <6> + List findByFirstname(String firstname, Limit limit); <6> - Person findFirstByShippingAddress(Address address); <7> + Person findByShippingAddress(Address address); <7> - Stream findAllBy(); <8> + Person findFirstByShippingAddress(Address address); <8> + + Stream findAllBy(); <9> @AllowFiltering - List findAllByAge(int age); <9> + List 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.