From 90bfa5317ba1cbbb97f76edcdb02a46e415b75ac Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Fri, 1 Sep 2017 13:00:18 +0200 Subject: [PATCH] DATACASS-56 - Cassandra paging support. We now support forward-only paging with Cassandra through the Template API and Repositories. Results in Cassandra are paged by navigating forward-only through pages described by a binary paging state encapsulated by CassandraPageRequest and accessible via the returned Slice. Spring Data Page's do not fit to Cassandra's paging concept because Cassandra paging is not based on limit/offset. Page requests are applicable to a Query and as parameter of query methods. Query query = Query.empty().pageRequest(CassandraPageRequest.first(10)); Slice slice = template.slice(query, User.class); do { // consume slice if (slice.hasNext()) { slice = template.select(query, slice.nextPageable(), User.class); } else { break; } } while (!slice.getContent().isEmpty()); assertThat(ids).hasSize(100); assertThat(iterations).isEqualTo(10); interface UserRepository implements Repository { Slice findAllByName(String name, Pageable pageRequest); } --- .../core/AsyncCassandraOperations.java | 25 ++ .../core/AsyncCassandraTemplate.java | 157 +++++++---- .../cassandra/core/CassandraOperations.java | 27 ++ .../cassandra/core/CassandraTemplate.java | 106 +++++--- .../data/cassandra/core/QueryUtils.java | 37 +++ .../core/ReactiveCassandraOperations.java | 30 +-- .../core/ReactiveCassandraTemplate.java | 67 ++--- .../core/query/CassandraPageRequest.java | 255 ++++++++++++++++++ .../data/cassandra/core/query/Query.java | 37 ++- .../repository/CassandraRepository.java | 22 +- .../query/AbstractCassandraQuery.java | 21 +- .../query/CassandraQueryExecution.java | 31 +++ .../query/CassandraQueryMethod.java | 7 +- .../support/SimpleCassandraRepository.java | 13 + ...syncCassandraTemplateIntegrationTests.java | 41 +++ .../CassandraTemplateIntegrationTests.java | 40 +++ .../query/CassandraPageRequestUnitTests.java | 115 ++++++++ .../cassandra/core/query/QueryUnitTests.java | 23 ++ .../QueryDerivationIntegrationTests.java | 22 ++ ...leCassandraRepositoryIntegrationTests.java | 12 + .../SimpleCassandraRepositoryUnitTests.java | 21 ++ .../reference/cassandra-repositories.adoc | 53 +++- src/main/asciidoc/reference/cassandra.adoc | 4 +- 23 files changed, 990 insertions(+), 176 deletions(-) create mode 100644 spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/query/CassandraPageRequest.java create mode 100644 spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/query/CassandraPageRequestUnitTests.java diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraOperations.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraOperations.java index cf2a907d4..a9b7962ad 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraOperations.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraOperations.java @@ -23,8 +23,10 @@ import org.springframework.data.cassandra.core.convert.CassandraConverter; import org.springframework.data.cassandra.core.cql.AsyncCqlOperations; import org.springframework.data.cassandra.core.cql.QueryOptions; import org.springframework.data.cassandra.core.cql.WriteOptions; +import org.springframework.data.cassandra.core.query.CassandraPageRequest; import org.springframework.data.cassandra.core.query.Query; import org.springframework.data.cassandra.core.query.Update; +import org.springframework.data.domain.Slice; import org.springframework.util.concurrent.ListenableFuture; import com.datastax.driver.core.Statement; @@ -110,6 +112,18 @@ public interface AsyncCassandraOperations { */ ListenableFuture> select(Statement statement, Class entityClass) throws DataAccessException; + /** + * Execute a {@code SELECT} query with paging and convert the resulting items to a {@link Slice} of entities. A sliced + * query translates the effective {@link Statement#getFetchSize() fetch size} to the page size. + * + * @param statement the CQL statement, must not be {@literal null}. + * @param entityClass The entity type must not be {@literal null}. + * @return the converted results + * @throws DataAccessException if there is any problem executing the query. + * @see CassandraPageRequest + */ + ListenableFuture> slice(Statement statement, Class entityClass) throws DataAccessException; + /** * Execute a {@code SELECT} query and convert the resulting items notifying {@link Consumer} for each entity. * @@ -147,6 +161,17 @@ public interface AsyncCassandraOperations { */ ListenableFuture> select(Query query, Class entityClass) throws DataAccessException; + /** + * Execute a {@code SELECT} query with paging and convert the resulting items to a {@link Slice} of entities. + * + * @param query the query object used to create a CQL statement, must not be {@literal null}. + * @param entityClass The entity type must not be {@literal null}. + * @return the converted results + * @throws DataAccessException if there is any problem executing the query. + * @see CassandraPageRequest + */ + ListenableFuture> slice(Query query, Class entityClass) throws DataAccessException; + /** * Execute a {@code SELECT} query and convert the resulting items notifying {@link Consumer} for each entity. * diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraTemplate.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraTemplate.java index 7cc2382d2..adeb72daf 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraTemplate.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraTemplate.java @@ -29,6 +29,7 @@ import org.springframework.data.cassandra.core.convert.UpdateMapper; import org.springframework.data.cassandra.core.cql.AsyncCqlOperations; import org.springframework.data.cassandra.core.cql.AsyncCqlTemplate; import org.springframework.data.cassandra.core.cql.AsyncSessionCallback; +import org.springframework.data.cassandra.core.cql.CassandraAccessor; import org.springframework.data.cassandra.core.cql.CqlExceptionTranslator; import org.springframework.data.cassandra.core.cql.CqlIdentifier; import org.springframework.data.cassandra.core.cql.CqlProvider; @@ -39,8 +40,10 @@ import org.springframework.data.cassandra.core.mapping.CassandraMappingContext; import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity; import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty; import org.springframework.data.cassandra.core.query.Query; +import org.springframework.data.domain.Slice; import org.springframework.data.mapping.context.MappingContext; import org.springframework.lang.Nullable; +import org.springframework.scheduling.annotation.AsyncResult; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; import org.springframework.util.concurrent.ListenableFuture; @@ -147,8 +150,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { this.statementFactory = new StatementFactory(new QueryMapper(converter), new UpdateMapper(converter)); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.AsyncCassandraOperations#getAsyncCqlOperations() */ @Override @@ -156,8 +158,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { return this.cqlOperations; } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.AsyncCassandraOperations#getConverter() */ @Override @@ -165,7 +166,6 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { return this.converter; } - /* (non-Javadoc) */ private static MappingCassandraConverter newConverter() { MappingCassandraConverter converter = new MappingCassandraConverter(); @@ -204,8 +204,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { // Methods dealing with static CQL // ------------------------------------------------------------------------- - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.AsyncCassandraOperations#select(java.lang.String, java.lang.Class) */ @Override @@ -216,6 +215,9 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { return select(new SimpleStatement(cql), entityClass); } + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.AsyncCassandraOperations#select(java.lang.String, java.util.function.Consumer, java.lang.Class) + */ @Override public ListenableFuture select(String cql, Consumer entityConsumer, Class entityClass) throws DataAccessException { @@ -227,8 +229,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { return select(new SimpleStatement(cql), entityConsumer, entityClass); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.AsyncCassandraOperations#selectOne(java.lang.String, java.lang.Class) */ @Override @@ -244,8 +245,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { // Methods dealing with com.datastax.driver.core.Statement // ------------------------------------------------------------------------- - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.AsyncCassandraOperations#select(com.datastax.driver.core.Statement, java.lang.Class) */ @Override @@ -257,6 +257,25 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { return getAsyncCqlOperations().query(statement, (row, rowNum) -> getConverter().read(entityClass, row)); } + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.AsyncCassandraOperations#slice(com.datastax.driver.core.Statement, java.lang.Class) + */ + @Override + public ListenableFuture> slice(Statement statement, Class entityClass) { + + Assert.notNull(statement, "Statement must not be null"); + Assert.notNull(entityClass, "Entity type must not be null"); + + ListenableFuture resultSet = getAsyncCqlOperations().queryForResultSet(statement); + CassandraConverter converter = getConverter(); + + return new MappingListenableFutureAdapter<>(resultSet, rs -> QueryUtils.readSlice(rs, + (row, rowNum) -> converter.read(entityClass, row), 0, getEffectiveFetchSize(statement))); + } + + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.AsyncCassandraOperations#select(com.datastax.driver.core.Statement, java.util.function.Consumer, java.lang.Class) + */ @Override public ListenableFuture select(Statement statement, Consumer entityConsumer, Class entityClass) throws DataAccessException { @@ -270,8 +289,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { }); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.AsyncCassandraOperations#selectOne(com.datastax.driver.core.Statement, java.lang.Class) */ @Override @@ -294,8 +312,21 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { Assert.notNull(query, "Query must not be null"); Assert.notNull(entityClass, "Entity type must not be null"); - return select(getStatementFactory().select(query, - getMappingContext().getRequiredPersistentEntity(entityClass)), entityClass); + return select(getStatementFactory().select(query, getMappingContext().getRequiredPersistentEntity(entityClass)), + entityClass); + } + + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.AsyncCassandraOperations#slice(org.springframework.data.cassandra.core.query.Query, java.lang.Class) + */ + @Override + public ListenableFuture> slice(Query query, Class entityClass) throws DataAccessException { + + Assert.notNull(query, "Query must not be null"); + Assert.notNull(entityClass, "Entity type must not be null"); + + return slice(statementFactory.select(query, getMappingContext().getRequiredPersistentEntity(entityClass)), + entityClass); } /* (non-Javadoc) @@ -309,8 +340,8 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { Assert.notNull(entityConsumer, "Entity Consumer must not be empty"); Assert.notNull(entityClass, "Entity type must not be null"); - return select(getStatementFactory().select(query, - getMappingContext().getRequiredPersistentEntity(entityClass)), entityConsumer, entityClass); + return select(getStatementFactory().select(query, getMappingContext().getRequiredPersistentEntity(entityClass)), + entityConsumer, entityClass); } /* (non-Javadoc) @@ -322,8 +353,8 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { Assert.notNull(query, "Query must not be null"); Assert.notNull(entityClass, "Entity type must not be null"); - return selectOne(getStatementFactory().select(query, - getMappingContext().getRequiredPersistentEntity(entityClass)), entityClass); + return selectOne(getStatementFactory().select(query, getMappingContext().getRequiredPersistentEntity(entityClass)), + entityClass); } /* (non-Javadoc) @@ -337,8 +368,8 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { Assert.notNull(update, "Update must not be null"); Assert.notNull(entityClass, "Entity type must not be null"); - return getAsyncCqlOperations().execute(getStatementFactory().update(query, update, - getMappingContext().getRequiredPersistentEntity(entityClass))); + return getAsyncCqlOperations().execute( + getStatementFactory().update(query, update, getMappingContext().getRequiredPersistentEntity(entityClass))); } /* (non-Javadoc) @@ -350,16 +381,15 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { Assert.notNull(query, "Query must not be null"); Assert.notNull(entityClass, "Entity type must not be null"); - return getAsyncCqlOperations().execute(getStatementFactory().delete(query, - getMappingContext().getRequiredPersistentEntity(entityClass))); + return getAsyncCqlOperations() + .execute(getStatementFactory().delete(query, getMappingContext().getRequiredPersistentEntity(entityClass))); } // ------------------------------------------------------------------------- // Methods dealing with entities // ------------------------------------------------------------------------- - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.AsyncCassandraOperations#count(java.lang.Class) */ @Override @@ -373,8 +403,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { return getAsyncCqlOperations().queryForObject(select, Long.class); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.AsyncCassandraOperations#exists(java.lang.Object, java.lang.Class) */ @Override @@ -393,8 +422,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { resultSet -> resultSet.iterator().hasNext()); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.AsyncCassandraOperations#selectOneById(java.lang.Object, java.lang.Class) */ @Override @@ -412,8 +440,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { return selectOne(select, entityClass); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.AsyncCassandraOperations#insert(java.lang.Object) */ @Override @@ -421,8 +448,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { return new MappingListenableFutureAdapter<>(insert(entity, InsertOptions.empty()), writeResult -> entity); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.AsyncCassandraOperations#insert(java.lang.Object, org.springframework.data.cassandra.core.InsertOptions) */ @Override @@ -433,12 +459,11 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { Insert insert = QueryUtils.createInsertQuery(getTableName(entity).toCql(), entity, options, getConverter()); - return new MappingListenableFutureAdapter<>( - getAsyncCqlOperations().execute(new AsyncStatementCallback(insert)), WriteResult::of); + return new MappingListenableFutureAdapter<>(getAsyncCqlOperations().execute(new AsyncStatementCallback(insert)), + WriteResult::of); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.AsyncCassandraOperations#update(java.lang.Object) */ @Override @@ -446,8 +471,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { return new MappingListenableFutureAdapter<>(update(entity, UpdateOptions.empty()), writeResult -> entity); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.AsyncCassandraOperations#update(java.lang.Object, org.springframework.data.cassandra.core.UpdateOptions) */ @Override @@ -458,12 +482,11 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { Update update = QueryUtils.createUpdateQuery(getTableName(entity).toCql(), entity, options, getConverter()); - return new MappingListenableFutureAdapter<>( - getAsyncCqlOperations().execute(new AsyncStatementCallback(update)), WriteResult::of); + return new MappingListenableFutureAdapter<>(getAsyncCqlOperations().execute(new AsyncStatementCallback(update)), + WriteResult::of); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.AsyncCassandraOperations#delete(java.lang.Object) */ @Override @@ -471,8 +494,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { return new MappingListenableFutureAdapter<>(delete(entity, QueryOptions.empty()), writeResult -> entity); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.AsyncCassandraOperations#delete(java.lang.Object, org.springframework.data.cassandra.core.cql.QueryOptions) */ @Override @@ -483,12 +505,11 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { Delete delete = QueryUtils.createDeleteQuery(getTableName(entity).toCql(), entity, options, getConverter()); - return new MappingListenableFutureAdapter<>( - getAsyncCqlOperations().execute(new AsyncStatementCallback(delete)), WriteResult::of); + return new MappingListenableFutureAdapter<>(getAsyncCqlOperations().execute(new AsyncStatementCallback(delete)), + WriteResult::of); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.AsyncCassandraOperations#deleteById(java.lang.Object, java.lang.Class) */ @Override @@ -506,8 +527,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { return getAsyncCqlOperations().execute(delete); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.AsyncCassandraOperations#truncate(java.lang.Class) */ @Override @@ -515,12 +535,34 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { Assert.notNull(entityClass, "Entity type must not be null"); - Truncate truncate = QueryBuilder.truncate( - getMappingContext().getRequiredPersistentEntity(entityClass).getTableName().toCql()); + Truncate truncate = QueryBuilder + .truncate(getMappingContext().getRequiredPersistentEntity(entityClass).getTableName().toCql()); return new MappingListenableFutureAdapter<>(getAsyncCqlOperations().execute(truncate), aBoolean -> null); } + // ------------------------------------------------------------------------- + // Implementation hooks and helper methods + // ------------------------------------------------------------------------- + + @SuppressWarnings("ConstantConditions") + private int getEffectiveFetchSize(Statement statement) { + + if (statement.getFetchSize() > 0) { + return statement.getFetchSize(); + } + + if (getAsyncCqlOperations() instanceof CassandraAccessor) { + CassandraAccessor accessor = (CassandraAccessor) getAsyncCqlOperations(); + if (accessor.getFetchSize() != -1) { + return accessor.getFetchSize(); + } + } + + return getAsyncCqlOperations().execute((AsyncSessionCallback) session -> AsyncResult + .forValue(session.getCluster().getConfiguration().getQueryOptions().getFetchSize())).completable().join(); + } + static class MappingListenableFutureAdapter extends org.springframework.util.concurrent.ListenableFutureAdapter { @@ -531,6 +573,9 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { this.mapper = mapper; } + /* (non-Javadoc) + * @see org.springframework.util.concurrent.FutureAdapter#adapt(java.lang.Object) + */ @Override protected T adapt(@Nullable S adapteeResult) throws ExecutionException { return mapper.apply(adapteeResult); @@ -545,6 +590,9 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { this.statement = statement; } + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.cql.AsyncSessionCallback#doInSession(com.datastax.driver.core.Session) + */ @Override public ListenableFuture doInSession(Session session) throws DriverException, DataAccessException { return new GuavaListenableFutureAdapter<>(session.executeAsync(statement), @@ -553,6 +601,9 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { : exceptionTranslator.translateExceptionIfPossible(e))); } + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.cql.CqlProvider#getCql() + */ @Override public String getCql() { return statement.toString(); diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraOperations.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraOperations.java index 1a42f50f1..a135ffb46 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraOperations.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraOperations.java @@ -25,8 +25,10 @@ import org.springframework.data.cassandra.core.cql.CqlIdentifier; import org.springframework.data.cassandra.core.cql.CqlOperations; import org.springframework.data.cassandra.core.cql.QueryOptions; import org.springframework.data.cassandra.core.cql.WriteOptions; +import org.springframework.data.cassandra.core.query.CassandraPageRequest; import org.springframework.data.cassandra.core.query.Query; import org.springframework.data.cassandra.core.query.Update; +import org.springframework.data.domain.Slice; import org.springframework.lang.Nullable; import com.datastax.driver.core.Statement; @@ -131,6 +133,18 @@ public interface CassandraOperations { */ List select(Statement statement, Class entityClass) throws DataAccessException; + /** + * Execute a {@code SELECT} query with paging and convert the resulting items to a {@link Slice} of entities. A sliced + * query translates the effective {@link Statement#getFetchSize() fetch size} to the page size. + * + * @param statement the CQL statement, must not be {@literal null}. + * @param entityClass The entity type must not be {@literal null}. + * @return the converted results + * @throws DataAccessException if there is any problem executing the query. + * @since 2.0 + */ + Slice slice(Statement statement, Class entityClass) throws DataAccessException; + /** * Execute a {@code SELECT} query and convert the resulting items to a {@link Iterator} of entities. *

@@ -159,6 +173,7 @@ public interface CassandraOperations { // ------------------------------------------------------------------------- // Methods dealing with org.springframework.data.cassandra.core.query.Query // ------------------------------------------------------------------------- + /** * Execute a {@code SELECT} query and convert the resulting items to a {@link List} of entities. * @@ -170,6 +185,18 @@ public interface CassandraOperations { */ List select(Query query, Class entityClass) throws DataAccessException; + /** + * Execute a {@code SELECT} query with paging and convert the resulting items to a {@link Slice} of entities. + * + * @param query the query object used to create a CQL statement, must not be {@literal null}. + * @param entityClass The entity type must not be {@literal null}. + * @return the converted results + * @throws DataAccessException if there is any problem executing the query. + * @since 2.0 + * @see CassandraPageRequest + */ + Slice slice(Query query, Class entityClass) throws DataAccessException; + /** * Execute a {@code SELECT} query and convert the resulting items to a {@link Iterator} of entities. *

diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java index 10f987b14..fb6c575a4 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java @@ -28,6 +28,7 @@ import org.springframework.data.cassandra.core.convert.CassandraConverter; import org.springframework.data.cassandra.core.convert.MappingCassandraConverter; import org.springframework.data.cassandra.core.convert.QueryMapper; import org.springframework.data.cassandra.core.convert.UpdateMapper; +import org.springframework.data.cassandra.core.cql.CassandraAccessor; import org.springframework.data.cassandra.core.cql.CqlIdentifier; import org.springframework.data.cassandra.core.cql.CqlOperations; import org.springframework.data.cassandra.core.cql.CqlProvider; @@ -39,6 +40,7 @@ import org.springframework.data.cassandra.core.mapping.CassandraMappingContext; import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity; import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty; import org.springframework.data.cassandra.core.query.Query; +import org.springframework.data.domain.Slice; import org.springframework.data.mapping.context.MappingContext; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; @@ -150,7 +152,6 @@ public class CassandraTemplate implements CassandraOperations { return this.converter; } - /* (non-Javadoc) */ private static MappingCassandraConverter newConverter() { MappingCassandraConverter converter = new MappingCassandraConverter(); @@ -160,8 +161,7 @@ public class CassandraTemplate implements CassandraOperations { return converter; } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.CassandraOperations#CqlOperations() */ @Override @@ -194,8 +194,7 @@ public class CassandraTemplate implements CassandraOperations { // Methods dealing with static CQL // ------------------------------------------------------------------------- - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.CassandraOperations#select(java.lang.String, java.lang.Class) */ @Override @@ -218,8 +217,7 @@ public class CassandraTemplate implements CassandraOperations { return stream(new SimpleStatement(cql), entityClass); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.CassandraOperations#selectOne(java.lang.String, java.lang.Class) */ @Override @@ -235,8 +233,7 @@ public class CassandraTemplate implements CassandraOperations { // Methods dealing with com.datastax.driver.core.Statement // ------------------------------------------------------------------------- - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.CassandraOperations#select(com.datastax.driver.core.Statement, java.lang.Class) */ @Override @@ -248,6 +245,22 @@ public class CassandraTemplate implements CassandraOperations { return getCqlOperations().query(statement, (row, rowNum) -> getConverter().read(entityClass, row)); } + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.CassandraOperations#slice(com.datastax.driver.core.Statement, java.lang.Class) + */ + @Override + public Slice slice(Statement statement, Class entityClass) { + + Assert.notNull(statement, "Statement must not be null"); + Assert.notNull(entityClass, "Entity type must not be null"); + + ResultSet resultSet = getCqlOperations().queryForResultSet(statement); + CassandraConverter converter = getConverter(); + + return QueryUtils.readSlice(resultSet, (row, rowNum) -> converter.read(entityClass, row), 0, + getEffectiveFetchSize(statement)); + } + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.CassandraOperations#stream(com.datastax.driver.core.Statement, java.lang.Class) */ @@ -261,8 +274,7 @@ public class CassandraTemplate implements CassandraOperations { .map(row -> getConverter().read(entityClass, row)); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.CassandraOperations#selectOne(com.datastax.driver.core.Statement, java.lang.Class) */ @Override @@ -287,6 +299,19 @@ public class CassandraTemplate implements CassandraOperations { entityClass); } + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.CassandraOperations#slice(org.springframework.data.cassandra.core.query.Query, java.lang.Class) + */ + @Override + public Slice slice(Query query, Class entityClass) throws DataAccessException { + + Assert.notNull(query, "Query must not be null"); + Assert.notNull(entityClass, "Entity type must not be null"); + + return slice(statementFactory.select(query, getMappingContext().getRequiredPersistentEntity(entityClass)), + entityClass); + } + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.CassandraOperations#stream(org.springframework.data.cassandra.core.query.Query, java.lang.Class) */ @@ -343,8 +368,7 @@ public class CassandraTemplate implements CassandraOperations { // Methods dealing with entities // ------------------------------------------------------------------------- - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.CassandraOperations#count(java.lang.Class) */ @Override @@ -360,8 +384,7 @@ public class CassandraTemplate implements CassandraOperations { return count != null ? count : 0L; } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.CassandraOperations#exists(java.lang.Object, java.lang.Class) */ @Override @@ -379,8 +402,7 @@ public class CassandraTemplate implements CassandraOperations { return getCqlOperations().queryForResultSet(select).iterator().hasNext(); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.CassandraOperations#selectOneById(java.lang.Object, java.lang.Class) */ @Override @@ -398,8 +420,7 @@ public class CassandraTemplate implements CassandraOperations { return selectOne(select, entityClass); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.CassandraOperations#insert(java.lang.Object) */ @Override @@ -407,8 +428,7 @@ public class CassandraTemplate implements CassandraOperations { insert(entity, InsertOptions.empty()); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.CassandraOperations#insert(java.lang.Object, org.springframework.data.cassandra.core.InsertOptions) */ @Override @@ -423,8 +443,7 @@ public class CassandraTemplate implements CassandraOperations { return getCqlOperations().execute(new StatementCallback(insert)); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.CassandraOperations#update(java.lang.Object) */ @Override @@ -432,8 +451,7 @@ public class CassandraTemplate implements CassandraOperations { update(entity, UpdateOptions.empty()); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.CassandraOperations#update(java.lang.Object, org.springframework.data.cassandra.core.UpdateOptions) */ @Override @@ -448,8 +466,7 @@ public class CassandraTemplate implements CassandraOperations { return getCqlOperations().execute(new StatementCallback(update)); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.CassandraOperations#delete(java.lang.Object) */ @Override @@ -457,8 +474,7 @@ public class CassandraTemplate implements CassandraOperations { delete(entity, QueryOptions.empty()); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.CassandraOperations#delete(java.lang.Object, org.springframework.data.cassandra.core.cql.QueryOptions) */ @Override @@ -473,8 +489,7 @@ public class CassandraTemplate implements CassandraOperations { return getCqlOperations().execute(new StatementCallback(delete)); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.CassandraOperations#deleteById(java.lang.Object, java.lang.Class) */ @Override @@ -492,8 +507,7 @@ public class CassandraTemplate implements CassandraOperations { return getCqlOperations().execute(delete); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.CassandraOperations#truncate(java.lang.Class) */ @Override @@ -511,9 +525,7 @@ public class CassandraTemplate implements CassandraOperations { // Implementation hooks and helper methods // ------------------------------------------------------------------------- - /* - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.CassandraOperations#getTableName(java.lang.Class) */ @Override @@ -521,8 +533,26 @@ public class CassandraTemplate implements CassandraOperations { return getMappingContext().getRequiredPersistentEntity(ClassUtils.getUserClass(entityClass)).getTableName(); } - /* - * (non-Javadoc) + @SuppressWarnings("ConstantConditions") + private int getEffectiveFetchSize(Statement statement) { + + if (statement.getFetchSize() > 0) { + return statement.getFetchSize(); + } + + if (getCqlOperations() instanceof CassandraAccessor) { + + CassandraAccessor accessor = (CassandraAccessor) getCqlOperations(); + if (accessor.getFetchSize() != -1) { + return accessor.getFetchSize(); + } + } + + return getCqlOperations().execute( + (SessionCallback) session -> session.getCluster().getConfiguration().getQueryOptions().getFetchSize()); + } + + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.CassandraOperations#batchOps() */ @Override diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/QueryUtils.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/QueryUtils.java index 16a0b015a..23e34e9b2 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/QueryUtils.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/QueryUtils.java @@ -15,12 +15,23 @@ */ package org.springframework.data.cassandra.core; +import java.util.ArrayList; +import java.util.List; + import org.springframework.data.cassandra.core.cql.QueryOptions; import org.springframework.data.cassandra.core.cql.QueryOptionsUtil; +import org.springframework.data.cassandra.core.cql.RowMapper; import org.springframework.data.cassandra.core.cql.WriteOptions; +import org.springframework.data.cassandra.core.query.CassandraPageRequest; import org.springframework.data.convert.EntityWriter; +import org.springframework.data.domain.PageRequest; +import org.springframework.data.domain.Pageable; +import org.springframework.data.domain.Slice; +import org.springframework.data.domain.SliceImpl; import org.springframework.util.Assert; +import com.datastax.driver.core.PagingState; +import com.datastax.driver.core.ResultSet; import com.datastax.driver.core.querybuilder.Delete; import com.datastax.driver.core.querybuilder.Delete.Where; import com.datastax.driver.core.querybuilder.Insert; @@ -127,4 +138,30 @@ class QueryUtils { return delete; } + + /** + * Read a {@link Slice} of data from the {@link ResultSet} for a {@link Pageable}. + * + * @param resultSet must not be {@literal null}. + * @param mapper must not be {@literal null}. + * @param page + * @param pageSize + * @return the resulting {@link Slice}. + */ + static Slice readSlice(ResultSet resultSet, RowMapper mapper, int page, int pageSize) { + + int toRead = resultSet.getAvailableWithoutFetching(); + List result = new ArrayList<>(toRead); + + for (int i = 0; i < toRead; i++) { + + T element = mapper.mapRow(resultSet.one(), i); + result.add(element); + } + + PagingState pagingState = resultSet.getExecutionInfo().getPagingState(); + CassandraPageRequest pageRequest = CassandraPageRequest.of(PageRequest.of(page, pageSize), pagingState); + + return new SliceImpl<>(result, pageRequest, pagingState != null); + } } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraOperations.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraOperations.java index 513ea417a..ccb155345 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraOperations.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraOperations.java @@ -44,6 +44,21 @@ import com.datastax.driver.core.Statement; */ public interface ReactiveCassandraOperations { + /** + * Returns the underlying {@link CassandraConverter}. + * + * @return the underlying {@link CassandraConverter}. + */ + CassandraConverter getConverter(); + + /** + * Expose the underlying {@link ReactiveCqlOperations} to allow CQL operations. + * + * @return the underlying {@link ReactiveCqlOperations}. + * @see ReactiveCqlOperations + */ + ReactiveCqlOperations getReactiveCqlOperations(); + // ------------------------------------------------------------------------- // Methods dealing with static CQL // ------------------------------------------------------------------------- @@ -252,19 +267,4 @@ public interface ReactiveCassandraOperations { * @throws DataAccessException if there is any problem issuing the execution. */ Mono truncate(Class entityClass) throws DataAccessException; - - /** - * Returns the underlying {@link CassandraConverter}. - * - * @return the underlying {@link CassandraConverter}. - */ - CassandraConverter getConverter(); - - /** - * Expose the underlying {@link ReactiveCqlOperations} to allow CQL operations. - * - * @return the underlying {@link ReactiveCqlOperations}. - * @see ReactiveCqlOperations - */ - ReactiveCqlOperations getReactiveCqlOperations(); } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java index 15a1b9fc9..b6ddd518c 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java @@ -159,7 +159,6 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { return this.converter; } - /* (non-Javadoc) */ private static MappingCassandraConverter newConverter() { MappingCassandraConverter converter = new MappingCassandraConverter(); @@ -208,8 +207,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { // Methods dealing with static CQL // ------------------------------------------------------------------------- - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#select(java.lang.String, java.lang.Class) */ @Override @@ -220,8 +218,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { return select(new SimpleStatement(cql), entityClass); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#selectOne(java.lang.String, java.lang.Class) */ @Override @@ -246,8 +243,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { return getReactiveCqlOperations().query(cql, (row, rowNum) -> getConverter().read(entityClass, row)); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#selectOne(com.datastax.driver.core.Statement, java.lang.Class) */ @Override @@ -268,8 +264,8 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { Assert.notNull(query, "Query must not be null"); Assert.notNull(entityClass, "Entity type must not be null"); - return select(getStatementFactory().select(query, - getMappingContext().getRequiredPersistentEntity(entityClass)), entityClass); + return select(getStatementFactory().select(query, getMappingContext().getRequiredPersistentEntity(entityClass)), + entityClass); } /* (non-Javadoc) @@ -281,8 +277,8 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { Assert.notNull(query, "Query must not be null"); Assert.notNull(entityClass, "Entity type must not be null"); - return selectOne(getStatementFactory().select(query, - getMappingContext().getRequiredPersistentEntity(entityClass)), entityClass); + return selectOne(getStatementFactory().select(query, getMappingContext().getRequiredPersistentEntity(entityClass)), + entityClass); } /* (non-Javadoc) @@ -296,8 +292,8 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { Assert.notNull(update, "Update must not be null"); Assert.notNull(entityClass, "Entity type must not be null"); - return getReactiveCqlOperations().execute(getStatementFactory().update(query, update, - getMappingContext().getRequiredPersistentEntity(entityClass))); + return getReactiveCqlOperations().execute( + getStatementFactory().update(query, update, getMappingContext().getRequiredPersistentEntity(entityClass))); } /* (non-Javadoc) @@ -309,16 +305,15 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { Assert.notNull(query, "Query must not be null"); Assert.notNull(entityClass, "Entity type must not be null"); - return getReactiveCqlOperations().execute(getStatementFactory().delete(query, - getMappingContext().getRequiredPersistentEntity(entityClass))); + return getReactiveCqlOperations() + .execute(getStatementFactory().delete(query, getMappingContext().getRequiredPersistentEntity(entityClass))); } // ------------------------------------------------------------------------- // Methods dealing with entities // ------------------------------------------------------------------------- - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#count(java.lang.Class) */ @Override @@ -327,13 +322,12 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { Assert.notNull(entityClass, "Entity type must not be null"); Select select = QueryBuilder.select().countAll() - .from(getMappingContext().getRequiredPersistentEntity(entityClass).getTableName().toCql()); + .from(getMappingContext().getRequiredPersistentEntity(entityClass).getTableName().toCql()); return getReactiveCqlOperations().queryForObject(select, Long.class); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#exists(java.lang.Object, java.lang.Class) */ @Override @@ -351,8 +345,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { return getReactiveCqlOperations().queryForRows(select).hasElements(); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#selectOneById(java.lang.Object, java.lang.Class) */ @Override @@ -370,8 +363,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { return selectOne(select, entityClass); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#insert(java.lang.Object) */ @Override @@ -379,8 +371,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { return insert(entity, InsertOptions.empty()).map(writeResult -> entity); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#insert(java.lang.Object, org.springframework.data.cassandra.core.InsertOptions) */ @Override @@ -394,8 +385,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { return getReactiveCqlOperations().execute(new StatementCallback(insert)).next(); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#update(java.lang.Object) */ @Override @@ -403,8 +393,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { return update(entity, UpdateOptions.empty()).map(writeResult -> entity); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#update(java.lang.Object, org.springframework.data.cassandra.core.UpdateOptions) */ @Override @@ -418,8 +407,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { return getReactiveCqlOperations().execute(new StatementCallback(update)).next(); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#delete(java.lang.Object) */ @Override @@ -427,8 +415,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { return delete(entity, QueryOptions.empty()).map(reactiveWriteResult -> entity); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#delete(java.lang.Object, org.springframework.data.cql.core.QueryOptions) */ @Override @@ -442,8 +429,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { return getReactiveCqlOperations().execute(new StatementCallback(delete)).next(); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#deleteById(java.lang.Object, java.lang.Class) */ @Override @@ -461,8 +447,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { return getReactiveCqlOperations().execute(delete); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#truncate(java.lang.Class) */ @Override @@ -481,11 +466,17 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { @NonNull Statement statement; + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.cql.ReactiveSessionCallback#doInSession(org.springframework.data.cassandra.ReactiveSession) + */ @Override public Publisher doInSession(ReactiveSession session) throws DriverException, DataAccessException { return session.execute(statement).flatMap(StatementCallback::toWriteResult); } + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.cql.CqlProvider#getCql() + */ @Override public String getCql() { return statement.toString(); 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 new file mode 100644 index 000000000..c17359a5e --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/query/CassandraPageRequest.java @@ -0,0 +1,255 @@ +/* + * Copyright 2017 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 + * + * http://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 org.springframework.data.domain.PageRequest; +import org.springframework.data.domain.Pageable; +import org.springframework.data.domain.Sort; +import org.springframework.data.domain.Sort.Direction; +import org.springframework.lang.Nullable; +import org.springframework.util.Assert; + +import com.datastax.driver.core.PagingState; + +/** + * Cassandra-specific {@link PageRequest} implementation providing access to {@link PagingState}. This class allows + * creation of the first page request and represents through Cassandra paging is based on the progress of fetched pages + * and allows forward-only navigation. Accessing a particular page requires fetching of all pages until the desired page + * is reached. + *

+ * The fetching progress is represented as {@link PagingState}. Query {@link com.datastax.driver.core.ResultSet results} + * are associated with a {@link com.datastax.driver.core.ExecutionInfo#getPagingState paging state} that is used on the + * next query as input parameter to continue page fetching. + * + * @author Mark Paluch + * @since 2.0 + */ +public class CassandraPageRequest extends PageRequest { + + private final @Nullable PagingState pagingState; + + private final boolean nextAllowed; + + private CassandraPageRequest(int page, int size, Sort sort, @Nullable PagingState pagingState, boolean nextAllowed) { + + super(page, size, sort); + + this.pagingState = pagingState; + this.nextAllowed = nextAllowed; + } + + /** + * Creates a new unsorted {@link PageRequest}. + * + * @param page zero-based page index. + * @param size the size of the page to be returned. + * @throws IllegalArgumentException for page requests other than the first page. + */ + public static CassandraPageRequest of(int page, int size) { + + Assert.isTrue(page == 0, + "Cannot create a Cassandra page request for an indexed page other than the first page (0)."); + + return of(page, size, Sort.unsorted()); + } + + /** + * Creates a new {@link PageRequest} with sort parameters applied. + * + * @param page zero-based page index. + * @param size the size of the page to be returned. + * @param sort must not be {@literal null}. + * @throws IllegalArgumentException for page requests other than the first page. + */ + public static CassandraPageRequest of(int page, int size, Sort sort) { + + Assert.isTrue(page == 0, + "Cannot create a Cassandra page request for an indexed page other than the first page (0)."); + + return new CassandraPageRequest(page, size, sort, null, false); + } + + /** + * Creates a new {@link PageRequest} with sort direction and properties applied. + * + * @param page zero-based page index. + * @param size the size of the page to be returned. + * @param direction must not be {@literal null}. + * @param properties must not be {@literal null}. + * @throws IllegalArgumentException for page requests other than the first page. + */ + public static CassandraPageRequest of(int page, int size, Direction direction, String... properties) { + + Assert.isTrue(page == 0, + "Cannot create a Cassandra page request for an indexed page other than the first page (0)."); + + return of(page, size, Sort.by(direction, properties)); + } + + /** + * Creates a 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 PagingState pagingState) { + return new CassandraPageRequest(current.getPageNumber(), current.getPageSize(), current.getSort(), pagingState, + pagingState != null); + } + + /** + * Creates a new unsorted {@link PageRequest} for the first page. + * + * @param size the size of the page to be returned. + */ + public static CassandraPageRequest first(int size) { + return of(0, size, Sort.unsorted()); + } + + /** + * Creates a new {@link PageRequest} with sort parameters applied for the first page. + * + * @param size the size of the page to be returned. + * @param sort must not be {@literal null}. + */ + public static CassandraPageRequest first(int size, Sort sort) { + return new CassandraPageRequest(0, size, sort, null, false); + } + + /** + * Creates a new {@link PageRequest} with sort direction and properties applied for the first page. + * + * @param size the size of the page to be returned. + * @param direction must not be {@literal null}. + * @param properties must not be {@literal null}. + */ + public static CassandraPageRequest first(int size, Direction direction, String... properties) { + return first(size, Sort.by(direction, properties)); + } + + /** + * Validate the {@link Pageable} whether it can be used for querying. Valid pageables are either: + *

    + *
  • Unpaged
    • + *
  • Request the first page
    • + *
  • {@link CassandraPageRequest} with a {@link PagingState}
    • + *
+ * + * @param pageable + * @throws IllegalArgumentException if the {@link Pageable} is not valid. + */ + public static void validatePageable(Pageable pageable) { + + if (pageable.isUnpaged() || pageable.getPageNumber() == 0) { + return; + } + + if (pageable instanceof CassandraPageRequest) { + + CassandraPageRequest pageRequest = (CassandraPageRequest) pageable; + + if (pageRequest.getPagingState() != null) { + return; + } + } + + throw new IllegalArgumentException( + "Paging queries for pages other than the first one require a CassandraPageRequest with a valid paging state"); + } + + /** + * @return the {@link PagingState} for the current {@link CassandraPageRequest} or {@literal null} if the current + * {@link Pageable} represents the last page. + */ + @Nullable + public PagingState getPagingState() { + return pagingState; + } + + /** + * Returns whether there's a next {@link Pageable} we can access from the current one. Will return {@literal false} in + * case the current {@link Pageable} already refers to the next page. + * + * @return {@literal true } if there's a next {@link Pageable} we can access from the current one. + */ + public boolean hasNext() { + return getPagingState() != null && nextAllowed; + } + + /* (non-Javadoc) + * @see org.springframework.data.domain.PageRequest#next() + */ + @Override + public CassandraPageRequest next() { + + Assert.state(hasNext(), "Cannot create a next page request without a PagingState"); + + return new CassandraPageRequest(getPageNumber() + 1, getPageSize(), getSort(), pagingState, false); + } + + /* (non-Javadoc) + * @see org.springframework.data.domain.PageRequest#previous() + */ + @Override + public PageRequest previous() { + + Assert.state(getPageNumber() < 2, "Cannot navigate to an intermediate page"); + + return super.previous(); + } + + /* (non-Javadoc) + * @see org.springframework.data.domain.PageRequest#equals(java.lang.Object) + */ + @Override + public boolean equals(@Nullable Object o) { + + if (this == o) + return true; + if (!(o instanceof CassandraPageRequest)) + return false; + if (!super.equals(o)) + return false; + + CassandraPageRequest that = (CassandraPageRequest) o; + + if (nextAllowed != that.nextAllowed) + return false; + return pagingState != null ? pagingState.equals(that.pagingState) : that.pagingState == null; + } + + /* (non-Javadoc) + * @see org.springframework.data.domain.PageRequest#hashCode() + */ + @Override + public int hashCode() { + + int result = super.hashCode(); + result = 31 * result + (pagingState != null ? pagingState.hashCode() : 0); + result = 31 * result + (nextAllowed ? 1 : 0); + return result; + } + + /* (non-Javadoc) + * @see java.lang.Object#toString() + */ + @Override + public String toString() { + return String.format("Cassandra page request [number: %d, size %d, sort: %s, paging state: %s]", getPageNumber(), + getPageSize(), getSort(), getPagingState()); + } +} 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 3ca5141bc..44f796b53 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 @@ -15,8 +15,7 @@ */ package org.springframework.data.cassandra.core.query; -import static org.springframework.util.ObjectUtils.nullSafeEquals; -import static org.springframework.util.ObjectUtils.nullSafeHashCode; +import static org.springframework.util.ObjectUtils.*; import java.util.ArrayList; import java.util.Arrays; @@ -27,6 +26,8 @@ import java.util.stream.Collectors; import java.util.stream.StreamSupport; import org.springframework.data.cassandra.core.cql.QueryOptions; +import org.springframework.data.domain.PageRequest; +import org.springframework.data.domain.Pageable; import org.springframework.data.domain.Sort; import org.springframework.data.domain.Sort.Order; import org.springframework.util.Assert; @@ -191,6 +192,34 @@ public class Query implements Filter { return this.sort; } + /** + * Create a {@link Query} initialized with a {@link PageRequest} to fetch the first page of results or advance in + * paging along with sorting. Reads (and overrides, if set) {@link Pageable#getPageSize() page size} into + * {@link QueryOptions#getFetchSize()} and sets {@link PagingState} and {@link Sort}. + * + * @param pageable must not be {@literal null}. + * @return a new {@link Query} object containing the former settings with {@link PageRequest} applied. + * @see CassandraPageRequest + */ + public Query pageRequest(Pageable pageable) { + + Assert.notNull(pageable, "Pageable must not be null"); + + CassandraPageRequest.validatePageable(pageable); + + PagingState pagingState = this.pagingState.orElse(null); + + if (pageable instanceof CassandraPageRequest) { + pagingState = ((CassandraPageRequest) pageable).getPagingState(); + } + + QueryOptions queryOptions = this.queryOptions.map(QueryOptions::mutate).orElse(QueryOptions.builder()) + .fetchSize(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); + } + /** * Set the {@link PagingState} to skip rows. * @@ -257,8 +286,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.pagingState, this.queryOptions, this.limit, + true); } /** diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/CassandraRepository.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/CassandraRepository.java index 2c3b033e1..31a8b2673 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/CassandraRepository.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/CassandraRepository.java @@ -19,7 +19,10 @@ import java.util.List; import org.springframework.data.cassandra.core.mapping.MapId; import org.springframework.data.cassandra.core.mapping.Table; +import org.springframework.data.cassandra.core.query.CassandraPageRequest; +import org.springframework.data.domain.Pageable; import org.springframework.data.domain.Persistable; +import org.springframework.data.domain.Slice; import org.springframework.data.repository.CrudRepository; import org.springframework.data.repository.NoRepositoryBean; @@ -39,27 +42,34 @@ import org.springframework.data.repository.NoRepositoryBean; @NoRepositoryBean public interface CassandraRepository extends CrudRepository { - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.repository.CrudRepository#saveAll(java.lang.Iterable) */ @Override List saveAll(Iterable entites); - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.repository.CrudRepository#findAll() */ @Override List findAll(); - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.repository.CrudRepository#findAllById(java.lang.Iterable) */ @Override List findAllById(Iterable ids); + /** + * Returns a {@link Slice} of entities meeting the paging restriction provided in the {@code Pageable} object. + * + * @param pageable must not be {@literal null}. + * @return a {@link Slice} of entities. + * @since 2.0 + * @see CassandraPageRequest + */ + Slice findAll(Pageable pageable); + /** * Inserts the given entity. Assumes the instance to be new to be able to apply insertion optimizations. Use the * returned instance for further operations as the save operation might have changed the entity instance completely. 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 3f85aec84..6c8141352 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 @@ -22,6 +22,7 @@ import org.springframework.data.cassandra.repository.query.CassandraQueryExecuti import org.springframework.data.cassandra.repository.query.CassandraQueryExecution.ResultProcessingExecution; import org.springframework.data.cassandra.repository.query.CassandraQueryExecution.ResultSetQuery; 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.repository.query.ParameterAccessor; import org.springframework.data.repository.query.RepositoryQuery; @@ -76,8 +77,8 @@ public abstract class AbstractCassandraQuery extends CassandraRepositoryQuerySup Statement statement = createQuery(parameterAccessor); - CassandraQueryExecution queryExecution = getExecution(new ResultProcessingConverter(resultProcessor, - getOperations().getConverter().getMappingContext(), getEntityInstantiators())); + CassandraQueryExecution queryExecution = getExecution(parameterAccessor, new ResultProcessingConverter( + resultProcessor, getOperations().getConverter().getMappingContext(), getEntityInstantiators())); Class resultType = resolveResultType(resultProcessor); @@ -102,15 +103,21 @@ public abstract class AbstractCassandraQuery extends CassandraRepositoryQuerySup /** * Returns the execution instance to use. * - * @param resultProcessing must not be {@literal null}. @return + * @param parameterAccessor must not be {@literal null}. + * @param resultProcessing must not be {@literal null}. + * @return a wrapped {@link CassandraQueryExecution} to execute this query method. */ - private CassandraQueryExecution getExecution(Converter resultProcessing) { - return new ResultProcessingExecution(getExecutionToWrap(resultProcessing), resultProcessing); + private CassandraQueryExecution getExecution(CassandraParameterAccessor parameterAccessor, + Converter resultProcessing) { + return new ResultProcessingExecution(getExecutionToWrap(parameterAccessor, resultProcessing), resultProcessing); } - private CassandraQueryExecution getExecutionToWrap(Converter resultProcessing) { + private CassandraQueryExecution getExecutionToWrap(CassandraParameterAccessor parameterAccessor, + Converter resultProcessing) { - if (getQueryMethod().isCollectionQuery()) { + if (getQueryMethod().isSliceQuery()) { + return new SlicedExecution(getOperations(), parameterAccessor.getPageable()); + } else if (getQueryMethod().isCollectionQuery()) { return new CollectionExecution(getOperations()); } else if (getQueryMethod().isResultSetQuery()) { return new ResultSetQuery(getOperations()); 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 e9d9d1cdb..5e40b4ff4 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 @@ -22,7 +22,9 @@ import org.springframework.core.convert.converter.Converter; 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.convert.EntityInstantiators; +import org.springframework.data.domain.Pageable; import org.springframework.data.mapping.context.MappingContext; import org.springframework.data.repository.query.ResultProcessor; import org.springframework.data.repository.query.ReturnedType; @@ -63,6 +65,35 @@ interface CassandraQueryExecution { } } + /** + * {@link CassandraQueryExecution} for a {@link Slice}. + * + * @author Mark Paluch + */ + @RequiredArgsConstructor + final class SlicedExecution implements CassandraQueryExecution { + + private final @NonNull CassandraOperations operations; + private final @NonNull Pageable pageable; + + /* (non-Javadoc) + * @see org.springframework.data.cassandra.repository.query.CassandraQueryExecution#execute(java.lang.String, java.lang.Class) + */ + @Override + public Object execute(Statement statement, Class type) { + + CassandraPageRequest.validatePageable(pageable); + + Statement statementToUse = statement.setFetchSize(pageable.getPageSize()); + + if (pageable instanceof CassandraPageRequest) { + statementToUse = statementToUse.setPagingState(((CassandraPageRequest) pageable).getPagingState()); + } + + return operations.slice(statementToUse, type); + } + } + /** * {@link CassandraQueryExecution} for collection returning queries. * diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraQueryMethod.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraQueryMethod.java index 918f00701..b36e5f2f1 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraQueryMethod.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraQueryMethod.java @@ -82,14 +82,13 @@ public class CassandraQueryMethod extends QueryMethod { } /** - * Validates that this query is not a page or slice query. + * Validates that this query is not a page query. */ @SuppressWarnings("unused") public void verify(Method method, RepositoryMetadata metadata) { - // TODO: support Page & Slice queries - if (isSliceQuery() || isPageQuery()) { - throw new InvalidDataAccessApiUsageException("Slice and Page queries are not supported"); + if (isPageQuery()) { + throw new InvalidDataAccessApiUsageException("Page queries are not supported. Use a Slice query."); } } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepository.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepository.java index d4594ec1a..5f6f52c77 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepository.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepository.java @@ -26,6 +26,8 @@ import org.springframework.data.cassandra.core.CassandraTemplate; import org.springframework.data.cassandra.core.query.Query; import org.springframework.data.cassandra.repository.CassandraRepository; import org.springframework.data.cassandra.repository.query.CassandraEntityInformation; +import org.springframework.data.domain.Pageable; +import org.springframework.data.domain.Slice; import org.springframework.data.util.StreamUtils; import org.springframework.data.util.Streamable; import org.springframework.util.Assert; @@ -193,6 +195,17 @@ public class SimpleCassandraRepository implements CassandraRepository findAll(Pageable pageable) { + + Assert.notNull(pageable, "Pageable must not be null"); + + return operations.slice(Query.empty().pageRequest(pageable), entityInformation.getJavaType()); + } + /* (non-Javadoc) * @see org.springframework.data.repository.CrudRepository#deleteById(java.lang.Object) */ diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/AsyncCassandraTemplateIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/AsyncCassandraTemplateIntegrationTests.java index 8f4616f02..34f4f3856 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/AsyncCassandraTemplateIntegrationTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/AsyncCassandraTemplateIntegrationTests.java @@ -17,12 +17,16 @@ package org.springframework.data.cassandra.core; import static org.assertj.core.api.Assertions.*; +import java.util.HashSet; +import java.util.LinkedHashSet; +import java.util.Set; import java.util.concurrent.Future; import org.junit.Before; import org.junit.Test; import org.springframework.data.cassandra.core.convert.MappingCassandraConverter; import org.springframework.data.cassandra.core.cql.AsyncCqlTemplate; +import org.springframework.data.cassandra.core.query.CassandraPageRequest; import org.springframework.data.cassandra.core.query.Columns; import org.springframework.data.cassandra.core.query.Criteria; import org.springframework.data.cassandra.core.query.Query; @@ -31,6 +35,7 @@ import org.springframework.data.cassandra.domain.User; import org.springframework.data.cassandra.domain.UserToken; import org.springframework.data.cassandra.repository.support.SchemaTestUtils; import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest; +import org.springframework.data.domain.Slice; import org.springframework.data.domain.Sort; import org.springframework.util.concurrent.ListenableFuture; @@ -255,6 +260,42 @@ public class AsyncCassandraTemplateIntegrationTests extends AbstractKeyspaceCrea assertThat(getUser(user.getId())).isNull(); } + @Test // DATACASS-56 + public void shouldPageRequests() { + + Set expectedIds = new LinkedHashSet<>(); + + for (int i = 0; i < 100; i++) { + + User user = new User("heisenberg" + i, "Walter", "White"); + expectedIds.add(user.getId()); + template.insert(user); + } + + Set ids = new HashSet<>(); + + Query query = Query.empty(); + Slice slice = getUninterruptibly( + template.slice(query.pageRequest(CassandraPageRequest.first(10)), User.class)); + int iterations = 0; + do { + + iterations++; + assertThat(slice).hasSize(10); + + slice.stream().map(User::getId).forEach(ids::add); + + if (slice.hasNext()) { + slice = getUninterruptibly(template.slice(query.pageRequest(slice.nextPageable()), User.class)); + } else { + break; + } + } while (!slice.getContent().isEmpty()); + + assertThat(ids).containsAll(expectedIds); + assertThat(iterations).isEqualTo(10); + } + private User getUser(String id) { return getUninterruptibly(template.selectOneById(id, User.class)); } diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/CassandraTemplateIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/CassandraTemplateIntegrationTests.java index 5bf10badc..3f88c6246 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/CassandraTemplateIntegrationTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/CassandraTemplateIntegrationTests.java @@ -20,7 +20,10 @@ import static org.junit.Assume.*; import java.util.Arrays; import java.util.Collections; +import java.util.HashSet; +import java.util.LinkedHashSet; import java.util.List; +import java.util.Set; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -29,6 +32,7 @@ import org.junit.Test; import org.springframework.data.cassandra.core.convert.MappingCassandraConverter; import org.springframework.data.cassandra.core.cql.CqlTemplate; import org.springframework.data.cassandra.core.mapping.BasicMapId; +import org.springframework.data.cassandra.core.query.CassandraPageRequest; import org.springframework.data.cassandra.core.query.Columns; import org.springframework.data.cassandra.core.query.Criteria; import org.springframework.data.cassandra.core.query.Query; @@ -39,6 +43,7 @@ import org.springframework.data.cassandra.domain.UserToken; import org.springframework.data.cassandra.repository.support.SchemaTestUtils; import org.springframework.data.cassandra.support.CassandraVersion; import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest; +import org.springframework.data.domain.Slice; import org.springframework.data.domain.Sort; import org.springframework.data.util.Version; @@ -391,4 +396,39 @@ public class CassandraTemplateIntegrationTests extends AbstractKeyspaceCreatingI assertThat(loadAfterDelete).isNull(); } + + @Test // DATACASS-56 + public void shouldPageRequests() { + + Set expectedIds = new LinkedHashSet<>(); + + for (int i = 0; i < 100; i++) { + + User user = new User("heisenberg" + i, "Walter", "White"); + expectedIds.add(user.getId()); + template.insert(user); + } + + Set ids = new HashSet<>(); + + Query query = Query.empty().pageRequest(CassandraPageRequest.first(10)); + Slice slice = template.slice(query, User.class); + int iterations = 0; + do { + + iterations++; + assertThat(slice).hasSize(10); + + slice.stream().map(User::getId).forEach(ids::add); + + if (slice.hasNext()) { + slice = template.slice(query.pageRequest(slice.nextPageable()), User.class); + } else { + break; + } + } while (!slice.getContent().isEmpty()); + + assertThat(ids).containsAll(expectedIds); + assertThat(iterations).isEqualTo(10); + } } diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/query/CassandraPageRequestUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/query/CassandraPageRequestUnitTests.java new file mode 100644 index 000000000..582de8c55 --- /dev/null +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/query/CassandraPageRequestUnitTests.java @@ -0,0 +1,115 @@ +/* + * Copyright 2017 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 + * + * http://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 static org.assertj.core.api.Assertions.*; +import static org.springframework.data.domain.Sort.Order.*; + +import org.junit.Test; +import org.springframework.data.domain.PageRequest; +import org.springframework.data.domain.Sort; +import org.springframework.data.domain.Sort.Direction; + +import com.datastax.driver.core.PagingState; + +/** + * Unit tests for {@link CassandraPageRequest}. + * + * @author Mark Paluch + */ +public class CassandraPageRequestUnitTests { + + PagingState pagingState = PagingState + .fromString("001400100c68656973656e62657267313600f07ffffff5006f934c985d6110148e1385ca793a75780004"); + + @Test // DATACASS-56 + public void shouldNotAllowNonZeroPageConstruction() { + + assertThatThrownBy(() -> CassandraPageRequest.of(1, 1)).isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> CassandraPageRequest.of(1, 1, Sort.unsorted())) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> CassandraPageRequest.of(1, 1, Direction.ASC, "foo")) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test // DATACASS-56 + public void shouldCreateFirstUnsortedPageRequest() { + + CassandraPageRequest pageRequest = CassandraPageRequest.first(10); + + assertThat(pageRequest.hasNext()).isFalse(); + assertThat(pageRequest.getPageSize()).isEqualTo(10); + assertThat(pageRequest.getSort()).isEqualTo(Sort.unsorted()); + } + + @Test // DATACASS-56 + public void shouldCreateFirstSortedPageRequest() { + + CassandraPageRequest pageRequest = CassandraPageRequest.first(10, Direction.ASC, "foo"); + + assertThat(pageRequest.hasNext()).isFalse(); + assertThat(pageRequest.getPageSize()).isEqualTo(10); + assertThat(pageRequest.getSort()).isEqualTo(Sort.by(asc("foo"))); + } + + @Test // DATACASS-56 + public void shouldFailIfNoNextPageIsAvailable() { + + CassandraPageRequest pageRequest = CassandraPageRequest.first(10, Direction.ASC, "foo"); + + assertThatThrownBy(pageRequest::next).isInstanceOf(IllegalStateException.class); + } + + @Test // DATACASS-56 + public void shouldCreateNextPageRequest() { + + CassandraPageRequest pageRequest = CassandraPageRequest.first(10, Direction.ASC, "foo"); + CassandraPageRequest next = CassandraPageRequest.of(pageRequest, pagingState).next(); + + assertThat(next.hasNext()).isFalse(); + assertThat(next.getPageSize()).isEqualTo(10); + } + + @Test // DATACASS-56 + public void shouldNotAllowPreviousPageNavigationToIntermediatePages() { + + CassandraPageRequest next = CassandraPageRequest.of(PageRequest.of(5, 10), pagingState); + + assertThatThrownBy(next::previous).isInstanceOf(IllegalStateException.class); + } + + @Test // DATACASS-56 + public void shouldCheckEquality() { + + CassandraPageRequest first = CassandraPageRequest.first(10); + CassandraPageRequest anotherFirst = CassandraPageRequest.first(10); + + assertThat(first.hashCode()).isEqualTo(anotherFirst.hashCode()); + + assertThat(first).isEqualTo(anotherFirst); + assertThat(first).isEqualTo(first); + + CassandraPageRequest withPaging = CassandraPageRequest.of(first, pagingState).next(); + CassandraPageRequest anotherWithPaging = CassandraPageRequest.of(anotherFirst, pagingState).next(); + + assertThat(withPaging.hashCode()).isEqualTo(anotherWithPaging.hashCode()); + assertThat(withPaging.hashCode()).isNotEqualTo(first.hashCode()); + + assertThat(withPaging).isEqualTo(anotherWithPaging); + assertThat(withPaging).isEqualTo(withPaging); + assertThat(withPaging).isNotEqualTo(anotherFirst); + } +} diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/query/QueryUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/query/QueryUnitTests.java index 42bc2832e..0dfad20ae 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/query/QueryUnitTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/query/QueryUnitTests.java @@ -16,9 +16,14 @@ package org.springframework.data.cassandra.core.query; import static org.assertj.core.api.Assertions.*; +import static org.springframework.data.domain.Sort.Order.*; import org.junit.Test; +import org.springframework.data.domain.PageRequest; import org.springframework.data.domain.Sort; +import org.springframework.data.domain.Sort.Direction; + +import com.datastax.driver.core.PagingState; /** * Unit tests for {@link Query}. @@ -64,4 +69,22 @@ public class QueryUnitTests { assertThat(query.getLimit()).isEqualTo(10); assertThat(query.isAllowFiltering()).isTrue(); } + + @Test // DATACASS-56 + public void shouldApplyPageRequests() { + + PagingState pagingState = PagingState + .fromString("001400100c68656973656e62657267313600f07ffffff5006f934c985d6110148e1385ca793a75780004"); + + CassandraPageRequest pageRequest = CassandraPageRequest.of(PageRequest.of(0, 42, Direction.ASC, "foo"), pagingState) + .next(); + + Query query = Query.empty().pageRequest(pageRequest); + + assertThat(query.getSort()).isEqualTo(Sort.by(asc("foo"))); + assertThat(query.getPagingState()).contains(pagingState); + assertThat(query.getQueryOptions()).hasValueSatisfying(actual -> { + assertThat(actual).extracting("fetchSize").contains(42); + }); + } } 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 5da36b8e8..0d5eb5dbb 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 @@ -19,6 +19,7 @@ import static org.assertj.core.api.Assertions.*; import static org.junit.Assume.*; import java.time.LocalDate; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; @@ -37,6 +38,7 @@ import org.springframework.data.cassandra.config.SchemaAction; import org.springframework.data.cassandra.core.CassandraOperations; import org.springframework.data.cassandra.core.cql.generator.CreateIndexCqlGenerator; import org.springframework.data.cassandra.core.cql.keyspace.CreateIndexSpecification; +import org.springframework.data.cassandra.core.query.CassandraPageRequest; import org.springframework.data.cassandra.domain.AddressType; import org.springframework.data.cassandra.domain.Person; import org.springframework.data.cassandra.repository.QueryDerivationIntegrationTests.PersonRepository.NumberOfChildren; @@ -46,6 +48,8 @@ import org.springframework.data.cassandra.repository.config.EnableCassandraRepos import org.springframework.data.cassandra.repository.support.AbstractSpringDataEmbeddedCassandraIntegrationTest; import org.springframework.data.cassandra.repository.support.IntegrationTestConfig; import org.springframework.data.cassandra.support.CassandraVersion; +import org.springframework.data.domain.Pageable; +import org.springframework.data.domain.Slice; import org.springframework.data.domain.Sort; import org.springframework.data.util.Version; import org.springframework.test.context.ContextConfiguration; @@ -298,6 +302,22 @@ public class QueryDerivationIntegrationTests extends AbstractSpringDataEmbeddedC assertThat(personRepository.findByNicknameContains("eisenber")).isEqualTo(walter); } + @Test // DATACASS-56 + public void shouldSelectSliced() { + + List result = new ArrayList<>(); + + Slice firstPage = personRepository.findAllSlicedByLastname("White", CassandraPageRequest.first(2)); + Slice nextPage = personRepository.findAllSlicedByLastname("White", firstPage.nextPageable()); + + result.addAll(firstPage.getContent()); + result.addAll(nextPage.getContent()); + + assertThat(firstPage).hasSize(2); + assertThat(nextPage).hasSize(1); + assertThat(result).contains(walter, skyler, flynn); + } + /** * @author Mark Paluch */ @@ -324,6 +344,8 @@ public class QueryDerivationIntegrationTests extends AbstractSpringDataEmbeddedC Person findByNumberOfChildren(NumberOfChildren numberOfChildren); + Slice findAllSlicedByLastname(String lastname, Pageable pageable); + Collection findPersonProjectedBy(); Collection findPersonDtoBy(); diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepositoryIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepositoryIntegrationTests.java index 6fcf524e8..0f84a62a8 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepositoryIntegrationTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepositoryIntegrationTests.java @@ -31,9 +31,11 @@ import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Configuration; import org.springframework.data.cassandra.core.CassandraOperations; +import org.springframework.data.cassandra.core.query.CassandraPageRequest; import org.springframework.data.cassandra.domain.User; import org.springframework.data.cassandra.repository.CassandraRepository; import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest; +import org.springframework.data.domain.Slice; import org.springframework.data.repository.query.DefaultEvaluationContextProvider; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -153,6 +155,16 @@ public class SimpleCassandraRepositoryIntegrationTests extends AbstractKeyspaceC assertThat(Users).hasSize(2); } + @Test // DATACASS-56 + public void findAllWithPaging() { + + Slice slice = repository.findAll(CassandraPageRequest.first(2)); + + assertThat(slice).hasSize(2); + + assertThat(repository.findAll(slice.nextPageable())).hasSize(2); + } + @Test // DATACASS-396 public void countShouldReturnNumberOfRecords() { diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepositoryUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepositoryUnitTests.java index 38bd4fd67..c9d54750c 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepositoryUnitTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepositoryUnitTests.java @@ -34,10 +34,14 @@ import org.springframework.data.cassandra.core.CassandraOperations; import org.springframework.data.cassandra.core.convert.MappingCassandraConverter; import org.springframework.data.cassandra.core.cql.CqlIdentifier; import org.springframework.data.cassandra.core.cql.CqlOperations; +import org.springframework.data.cassandra.core.cql.QueryOptions; import org.springframework.data.cassandra.core.mapping.CassandraMappingContext; import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity; import org.springframework.data.cassandra.core.mapping.UserTypeResolver; +import org.springframework.data.cassandra.core.query.CassandraPageRequest; +import org.springframework.data.cassandra.core.query.Query; import org.springframework.data.cassandra.domain.Person; +import org.springframework.data.domain.Sort.Direction; import com.datastax.driver.core.UserType; import com.datastax.driver.core.querybuilder.Insert; @@ -139,6 +143,23 @@ public class SimpleCassandraRepositoryUnitTests { verify(cassandraOperations).insert(person); } + @Test // DATACASS-56 + public void shouldSelectWithPaging() { + + CassandraPageRequest pageRequest = CassandraPageRequest.first(10, Direction.ASC, "foo"); + + repository = new SimpleCassandraRepository( + new MappingCassandraEntityInformation( + converter.getMappingContext().getRequiredPersistentEntity(SimplePerson.class), converter), + cassandraOperations); + + repository.findAll(pageRequest); + + verify(cassandraOperations).slice( + Query.empty().sort(pageRequest.getSort()).queryOptions(QueryOptions.builder().fetchSize(10).build()), + SimplePerson.class); + } + @Data static class SimplePerson { diff --git a/src/main/asciidoc/reference/cassandra-repositories.adoc b/src/main/asciidoc/reference/cassandra-repositories.adoc index bbf4f4b68..d2e2a4994 100644 --- a/src/main/asciidoc/reference/cassandra-repositories.adoc +++ b/src/main/asciidoc/reference/cassandra-repositories.adoc @@ -111,7 +111,7 @@ class ApplicationConfig extends AbstractCassandraConfiguration { As our domain Repository extends `CrudRepository` it provides you with basic CRUD operations. Working with the Repository instance is just a matter of injecting the Repository as a dependency into a client. -.Paging access to Person entities +.Basic access to Person entities ==== [source,java] ---- @@ -131,6 +131,35 @@ public class PersonRepositoryTests { ---- ==== +Cassandra repositories support paging and sorting for paginated and sorted access to the entities. Cassandra paging requires a paging state to forward-only navigate through pages. A `Slice` keeps track of the current paging state and allows creation of a `Pageable` to request the next page. + +.Paging access to Person entities +==== +[source,java] +---- +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration +public class PersonRepositoryTests { + + @Autowired PersonRepository repository; + + @Test + public void readsPagesCorrectly() { + + Slice firstBatch = repository.findAll(CassandraPageRequest.first(10)); + + assertThat(firstBatch).hasSize(10); + + Page nextBatch = repository.findAll(firstBatch.nextPageable()); + + // … + } +} +---- +==== + +NOTE: Cassandra repositories do not extend `PagingAndSortingRepository` because classic paging patterns using limit/offset are not applicable to Cassandra. + The sample creates an application context with Spring's unit test support, which will perform annotation-based dependency injection into the test class. Inside the test cases (test methods) we simply use the Repository to query the data store. We invoke the Repository query method that requests the all `Person` instances. @@ -147,26 +176,30 @@ the Apache Cassandra database. Defining such a query is just a matter of declari ---- public interface PersonRepository extends CrudRepository { - List findByLastname(String lastname); <1> + List findByLastname(String lastname); <1> - List findByFirstname(String firstname, Sort sort); <2> + Slice findByFirstname(String firstname, Pageable pageRequest); <2> - List findByFirstname(String firstname, QueryOptions opts); <3> + List findByFirstname(String firstname, QueryOptions opts); <3> - Person findByShippingAddress(Address address); <4> + List findByFirstname(String firstname, Sort sort); <4> - Stream findAllBy(); <5> + Person findByShippingAddress(Address address); <5> + + Stream findAllBy(); <6> +>>>>>>> 18c314d... DATACASS-56 - Cassandra paging support. } ---- <1> The method shows a query for all people with the given `lastname`. The query will be derived from parsing the method name for constraints which can be concatenated with `And`. Thus the method name will result in a query expression of `SELECT * from person WHERE lastname = 'lastname'`. -<2> Applies dynamic sorting to a query. Just add a `Sort` parameter to your method signature and Spring Data. -will automatically apply ordering to the query accordingly. +<2> Applies pagination to a query. Just equip your method signature with a `Pageable` parameter and let the method return a `Slice` instance and we will automatically page the query accordingly. <3> Passing a `QueryOptions` object will apply the query options to the resulting query before it's execution. -<4> Shows that you can query based on properties which are not a primitive type using registered `Converter`'s. +<4> Applies dynamic sorting to a query. Just add a `Sort` parameter to your method signature and Spring Data +will automatically apply ordering to the query accordingly. +<5> Shows that you can query based on properties which are not a primitive type using registered `Converter`'s in `CustomConversions`. -<5> Uses a Java 8 `Stream` which reads and converts individual elements while iterating the stream. +<6> Uses a Java 8 `Stream` which reads and converts individual elements while iterating the stream. ==== NOTE: Querying non-primary key properties requires secondary indexes. diff --git a/src/main/asciidoc/reference/cassandra.adoc b/src/main/asciidoc/reference/cassandra.adoc index 6ba2c03d4..758efc12c 100644 --- a/src/main/asciidoc/reference/cassandra.adoc +++ b/src/main/asciidoc/reference/cassandra.adoc @@ -1137,6 +1137,7 @@ The `Query` class has some additional methods used to provide options for the qu * `Query` *and* `(CriteriaDefinition criteria)` used to add additional criteria to the query. * `Query` *columns* `(Columns columns)` used to define columns to be included in the query results. * `Query` *limit* `(long limit)` used to limit the size of the returned results to the provided limit (used for paging). +* `Query` *pageRequest* `(Pageable pageRequest)` used to associate `Sort`, `PagingState` and `fetchSize` with the query (used for paging). * `Query` *pagingState* `(PagingState pagingState)` used to associate a `PagingState` with the query (used for paging). * `Query` *queryOptions* `(QueryOptions queryOptions)` used to associate `QueryOptions` with the query. * `Query` *sort* `(Sort sort)` used to provide sort definition for the results. @@ -1151,8 +1152,9 @@ The query methods need to specify the target type T that will be returned. * `List` *select* `(Query query, Class entityClass)` Query for a list of objects of type T from the table. * `T` *selectOneById* `(Query query, Class entityClass)` Query for a single object of type T from the table. +* `Slice` *slice* `(Query query, Class entityClass)` Start or continue paging by querying for a `Slice` of objects of type T from the table. +* `T` *selectOne* `(Query query, Class entityClass)` Query for a single object of type T from the table. * `Stream` *stream* `(Query query, Class entityClass)` Query for a stream of objects of type T from the table. - * `List` *select* `(String cql, Class entityClass)` Ad-hoc query for a list of objects of type T from the table providing a CQL statement. * `T` *selectOneById* `(String cql, Class entityClass)` Ad-hoc query for a single object of type T from the table providing a CQL statement. * `Stream` *stream* `(String cql, Class entityClass)` Ad-hoc query for a stream of objects of type T from the table providing a CQL statement.