Spring/spring-data-cassandra
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATACASS-56 - Cassandra paging support.

Browse Source 
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<User> 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<User, String> {

    Slice<User> findAllByName(String name, Pageable pageRequest);
}
This commit is contained in:
Mark Paluch
2017-09-01 13:00:18 +02:00
committed by John Blum
parent 62e6d33314
commit 90bfa5317b
23 changed files with 990 additions and 176 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraOperations.java
View File

@@ -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 {
*/
<T> ListenableFuture<List<T>> select(Statement statement, Class<T> 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
*/
<T> ListenableFuture<Slice<T>> slice(Statement statement, Class<T> 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 {
*/
<T> ListenableFuture<List<T>> select(Query query, Class<T> 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
*/
<T> ListenableFuture<Slice<T>> slice(Query query, Class<T> entityClass) throws DataAccessException;
/**
* Execute a {@code SELECT} query and convert the resulting items notifying {@link Consumer} for each entity.
*

157
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraTemplate.java
View File

@@ -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 <T> ListenableFuture<Void> select(String cql, Consumer<T> entityConsumer, Class<T> 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 <T> ListenableFuture<Slice<T>> slice(Statement statement, Class<T> entityClass) {
Assert.notNull(statement, "Statement must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
ListenableFuture<ResultSet> 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 <T> ListenableFuture<Void> select(Statement statement, Consumer<T> entityConsumer, Class<T> 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 <T> ListenableFuture<Slice<T>> slice(Query query, Class<T> 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<Integer>) session -> AsyncResult
.forValue(session.getCluster().getConfiguration().getQueryOptions().getFetchSize())).completable().join();
}
static class MappingListenableFutureAdapter<T, S>
extends org.springframework.util.concurrent.ListenableFutureAdapter<T, S> {
@@ -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<ResultSet> 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();

27
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraOperations.java
View File

@@ -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 {
*/
<T> List<T> select(Statement statement, Class<T> 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
*/
<T> Slice<T> slice(Statement statement, Class<T> entityClass) throws DataAccessException;
/**
* Execute a {@code SELECT} query and convert the resulting items to a {@link Iterator} of entities.
* <p>
@@ -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 {
*/
<T> List<T> select(Query query, Class<T> 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
*/
<T> Slice<T> slice(Query query, Class<T> entityClass) throws DataAccessException;
/**
* Execute a {@code SELECT} query and convert the resulting items to a {@link Iterator} of entities.
* <p>

106
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java
View File

@@ -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 <T> Slice<T> slice(Statement statement, Class<T> 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 <T> Slice<T> slice(Query query, Class<T> 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<Integer>) session -> session.getCluster().getConfiguration().getQueryOptions().getFetchSize());
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#batchOps()
*/
@Override

37
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/QueryUtils.java
View File

@@ -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 <T> Slice<T> readSlice(ResultSet resultSet, RowMapper<T> mapper, int page, int pageSize) {
int toRead = resultSet.getAvailableWithoutFetching();
List<T> 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);
}
}

30
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraOperations.java
View File

@@ -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<Void> 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();
}

67
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java
View File

@@ -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<WriteResult> 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();

255
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/query/CassandraPageRequest.java Normal file
View File

@@ -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.
* <p/>
* 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:
* <ul>
* <li>Unpaged</li>
* <li>Request the first page</li>
* <li>{@link CassandraPageRequest} with a {@link PagingState}</li>
* </ul>
*
* @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());
}
}

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

@@ -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);
}
/**

22
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/CassandraRepository.java
View File

@@ -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<T, ID> extends CrudRepository<T, ID> {
/*
* (non-Javadoc)
/* (non-Javadoc)
* @see org.springframework.data.repository.CrudRepository#saveAll(java.lang.Iterable)
*/
@Override
<S extends T> List<S> saveAll(Iterable<S> entites);
/*
* (non-Javadoc)
/* (non-Javadoc)
* @see org.springframework.data.repository.CrudRepository#findAll()
*/
@Override
List<T> findAll();
/*
* (non-Javadoc)
/* (non-Javadoc)
* @see org.springframework.data.repository.CrudRepository#findAllById(java.lang.Iterable)
*/
@Override
List<T> findAllById(Iterable<ID> 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<T> 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.

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

@@ -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<Object, Object> resultProcessing) {
return new ResultProcessingExecution(getExecutionToWrap(resultProcessing), resultProcessing);
private CassandraQueryExecution getExecution(CassandraParameterAccessor parameterAccessor,
Converter<Object, Object> resultProcessing) {
return new ResultProcessingExecution(getExecutionToWrap(parameterAccessor, resultProcessing), resultProcessing);
}
private CassandraQueryExecution getExecutionToWrap(Converter<Object, Object> resultProcessing) {
private CassandraQueryExecution getExecutionToWrap(CassandraParameterAccessor parameterAccessor,
Converter<Object, Object> 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());

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

@@ -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.
*

7
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraQueryMethod.java
View File

@@ -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.");
}
}

13
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepository.java
View File

@@ -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<T, ID> implements CassandraRepository<T,
entityInformation.getJavaType());
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.repository.CassandraRepository#findAll(org.springframework.data.domain.Pageable)
*/
@Override
public Slice<T> 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)
*/

41
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/AsyncCassandraTemplateIntegrationTests.java
View File

@@ -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<String> 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<String> ids = new HashSet<>();
Query query = Query.empty();
Slice<User> 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));
}

40
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/CassandraTemplateIntegrationTests.java
View File

@@ -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<String> 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<String> ids = new HashSet<>();
Query query = Query.empty().pageRequest(CassandraPageRequest.first(10));
Slice<User> 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);
}
}

115
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/query/CassandraPageRequestUnitTests.java Normal file
View File

@@ -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);
}
}

23
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/query/QueryUnitTests.java
View File

@@ -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);
});
}
}

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

@@ -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<Person> result = new ArrayList<>();
Slice<Person> firstPage = personRepository.findAllSlicedByLastname("White", CassandraPageRequest.first(2));
Slice<Person> 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<Person> findAllSlicedByLastname(String lastname, Pageable pageable);
Collection<PersonProjection> findPersonProjectedBy();
Collection<PersonDto> findPersonDtoBy();

12
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepositoryIntegrationTests.java
View File

@@ -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<User> slice = repository.findAll(CassandraPageRequest.first(2));
assertThat(slice).hasSize(2);
assertThat(repository.findAll(slice.nextPageable())).hasSize(2);
}
@Test // DATACASS-396
public void countShouldReturnNumberOfRecords() {

21
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepositoryUnitTests.java
View File

@@ -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<Object, String>(
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 {

53
src/main/asciidoc/reference/cassandra-repositories.adoc
View File

@@ -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<Person> firstBatch = repository.findAll(CassandraPageRequest.first(10));
assertThat(firstBatch).hasSize(10);
Page<Person> 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<Person, String> {
List<Person> findByLastname(String lastname); <1>
List<Person> findByLastname(String lastname); <1>
List<Person> findByFirstname(String firstname, Sort sort); <2>
Slice<Person> findByFirstname(String firstname, Pageable pageRequest); <2>
List<Person> findByFirstname(String firstname, QueryOptions opts); <3>
List<Person> findByFirstname(String firstname, QueryOptions opts); <3>
Person findByShippingAddress(Address address); <4>
List<Person> findByFirstname(String firstname, Sort sort); <4>
Stream<Person> findAllBy(); <5>
Person findByShippingAddress(Address address); <5>
Stream<Person> 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.

4
src/main/asciidoc/reference/cassandra.adoc
View File

@@ -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<T>` *select* `(Query query, Class<T> entityClass)` Query for a list of objects of type T from the table.
* `T` *selectOneById* `(Query query, Class<T> entityClass)` Query for a single object of type T from the table.
* `Slice<T>` *slice* `(Query query, Class<T> entityClass)` Start or continue paging by querying for a `Slice` of objects of type T from the table.
* `T` *selectOne* `(Query query, Class<T> entityClass)` Query for a single object of type T from the table.
* `Stream<T>` *stream* `(Query query, Class<T> entityClass)` Query for a stream of objects of type T from the table.
* `List<T>` *select* `(String cql, Class<T> entityClass)` Ad-hoc query for a list of objects of type T from the table providing a CQL statement.
* `T` *selectOneById* `(String cql, Class<T> entityClass)` Ad-hoc query for a single object of type T from the table providing a CQL statement.
* `Stream<T>` *stream* `(String cql, Class<T> entityClass)` Ad-hoc query for a stream of objects of type T from the table providing a CQL statement.