From befa0ff2ef6fe7b52ed59005bb27a467029bc535 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Wed, 5 Jul 2023 14:59:05 +0200 Subject: [PATCH] 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 --- .../core/query/CassandraPageRequest.java | 42 +++++-- .../core/query/CassandraScrollPosition.java | 116 ++++++++++++++++++ .../data/cassandra/core/query/Query.java | 77 ++++++++---- .../query/AbstractCassandraQuery.java | 3 + .../query/AbstractReactiveCassandraQuery.java | 4 + .../query/CassandraParameterAccessor.java | 4 + .../CassandraParametersParameterAccessor.java | 19 +++ .../query/CassandraQueryExecution.java | 35 +++++- .../query/ConvertingParameterAccessor.java | 5 +- .../query/QueryStatementCreator.java | 36 ++++-- .../ReactiveCassandraQueryExecution.java | 36 ++++++ .../repository/query/WindowUtil.java | 38 ++++++ .../QueryDerivationIntegrationTests.java | 52 ++++++++ ...veCassandraRepositoryIntegrationTests.java | 54 +++++++- .../query/StubParameterAccessor.java | 4 +- .../reference/cassandra-repositories.adoc | 35 +++--- 16 files changed, 488 insertions(+), 72 deletions(-) create mode 100644 spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/query/CassandraScrollPosition.java create mode 100644 spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/WindowUtil.java 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.