diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/query/CassandraPageRequest.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/query/CassandraPageRequest.java index 11f89b8be..ad5334b72 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/query/CassandraPageRequest.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/query/CassandraPageRequest.java @@ -201,6 +201,20 @@ public class CassandraPageRequest extends PageRequest { return new CassandraPageRequest(getPageNumber() + 1, getPageSize(), getSort(), this.pagingState, false); } + /** + * Create a new {@link CassandraPageRequest} associated with {@link Sort} sort order. + * + * @param sort must not be {@literal null}. + * @return a new {@link CassandraPageRequest} associated with the given {@link Sort}. + * @since 2.1.13 + */ + public CassandraPageRequest withSort(Sort sort) { + + Assert.notNull(sort, "Sort must not be null"); + + return new CassandraPageRequest(this.getPageNumber(), this.getPageSize(), sort, this.pagingState, this.nextAllowed); + } + /* (non-Javadoc) * @see org.springframework.data.domain.PageRequest#previous() */ 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 630b19c4d..9e5277931 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 @@ -29,6 +29,8 @@ import org.springframework.data.cassandra.core.mapping.CassandraPersistentProper import org.springframework.data.cassandra.core.query.CassandraPageRequest; import org.springframework.data.convert.EntityInstantiators; import org.springframework.data.domain.Pageable; +import org.springframework.data.domain.Slice; +import org.springframework.data.domain.SliceImpl; import org.springframework.data.mapping.context.MappingContext; import org.springframework.data.repository.query.ResultProcessor; import org.springframework.data.repository.query.ReturnedType; @@ -96,7 +98,14 @@ interface CassandraQueryExecution { statementToUse = statementToUse.setPagingState(((CassandraPageRequest) pageable).getPagingState()); } - return operations.slice(statementToUse, type); + Slice slice = operations.slice(statementToUse, type); + + if (pageable.getSort().isUnsorted()) { + return slice; + } + + CassandraPageRequest cassandraPageRequest = (CassandraPageRequest) slice.getPageable(); + return new SliceImpl<>(slice.getContent(), cassandraPageRequest.withSort(pageable.getSort()), slice.hasNext()); } } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraQueryExecution.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraQueryExecution.java index 9dc611395..f860b6177 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraQueryExecution.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraQueryExecution.java @@ -32,6 +32,8 @@ import org.springframework.data.cassandra.core.mapping.CassandraPersistentProper import org.springframework.data.cassandra.core.query.CassandraPageRequest; import org.springframework.data.convert.EntityInstantiators; import org.springframework.data.domain.Pageable; +import org.springframework.data.domain.Slice; +import org.springframework.data.domain.SliceImpl; import org.springframework.data.mapping.context.MappingContext; import org.springframework.data.repository.query.ResultProcessor; import org.springframework.data.repository.query.ReturnedType; @@ -55,6 +57,7 @@ interface ReactiveCassandraQueryExecution { * {@link ReactiveCassandraQueryExecution} for a {@link org.springframework.data.domain.Slice}. * * @author Hleb Albau + * @author Mark Paluch * @since 2.1 */ @RequiredArgsConstructor @@ -76,8 +79,18 @@ interface ReactiveCassandraQueryExecution { if (pageable instanceof CassandraPageRequest) { statementToUse = statementToUse.setPagingState(((CassandraPageRequest) pageable).getPagingState()); } + Mono> slice = operations.slice(statementToUse, type); - return operations.slice(statementToUse, type); + if (pageable.getSort().isUnsorted()) { + return slice; + } + + return slice.map(it -> { + + CassandraPageRequest cassandraPageRequest = (CassandraPageRequest) it.getPageable(); + return new SliceImpl<>(it.getContent(), cassandraPageRequest.withSort(pageable.getSort()), it.hasNext()); + + }); } } 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 cf0020adc..5326e8dbc 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 @@ -17,11 +17,14 @@ package org.springframework.data.cassandra.repository.support; import static org.assertj.core.api.Assertions.*; +import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.Optional; +import java.util.UUID; import java.util.concurrent.CopyOnWriteArrayList; import java.util.stream.Collectors; +import java.util.stream.IntStream; import org.junit.Before; import org.junit.Test; @@ -40,9 +43,12 @@ import org.springframework.data.cassandra.core.mapping.event.BeforeSaveEvent; import org.springframework.data.cassandra.core.mapping.event.CassandraMappingEvent; import org.springframework.data.cassandra.core.query.CassandraPageRequest; import org.springframework.data.cassandra.domain.User; +import org.springframework.data.cassandra.domain.UserToken; import org.springframework.data.cassandra.repository.CassandraRepository; import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest; +import org.springframework.data.domain.Pageable; import org.springframework.data.domain.Slice; +import org.springframework.data.domain.Sort; import org.springframework.data.repository.query.ExtensionAwareQueryMethodEvaluationContextProvider; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringRunner; @@ -181,6 +187,36 @@ public class SimpleCassandraRepositoryIntegrationTests extends AbstractKeyspaceC assertThat(repository.findAll(slice.nextPageable())).hasSize(2); } + @Test // DATACASS-700 + public void findAllWithPagingAndSorting() { + + UserTokenRepostitory repository = factory.getRepository(UserTokenRepostitory.class); + repository.deleteAll(); + + UUID id = UUID.randomUUID(); + List users = IntStream.range(0, 100).mapToObj(value -> { + + UserToken token = new UserToken(); + token.setUserId(id); + token.setToken(UUID.randomUUID()); + + return token; + }).collect(Collectors.toList()); + + repository.saveAll(users); + + List result = new ArrayList<>(); + Slice slice = repository.findAllByUserId(id, CassandraPageRequest.first(10, Sort.by("token"))); + + while (!slice.isEmpty() || slice.hasNext()) { + result.addAll(slice.getContent()); + + slice = repository.findAllByUserId(id, slice.nextPageable()); + } + + assertThat(result).hasSize(100); + } + @Test // DATACASS-396 public void countShouldReturnNumberOfRecords() { @@ -328,7 +364,11 @@ public class SimpleCassandraRepositoryIntegrationTests extends AbstractKeyspaceC assertThat(loaded).isEmpty(); } - interface UserRepostitory extends CassandraRepository { } + interface UserRepostitory extends CassandraRepository {} + + interface UserTokenRepostitory extends CassandraRepository { + Slice findAllByUserId(UUID id, Pageable pageRequest); + } static class CaptureEventListener extends AbstractCassandraEventListener { diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepositoryIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepositoryIntegrationTests.java index cbec14adc..8919cf736 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepositoryIntegrationTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepositoryIntegrationTests.java @@ -15,14 +15,20 @@ */ package org.springframework.data.cassandra.repository.support; -import static org.assertj.core.api.Assertions.assertThat; - -import java.util.Arrays; +import static org.assertj.core.api.Assertions.*; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.test.StepVerifier; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.UUID; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -34,9 +40,14 @@ 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.ReactiveCassandraOperations; +import org.springframework.data.cassandra.core.query.CassandraPageRequest; import org.springframework.data.cassandra.domain.User; +import org.springframework.data.cassandra.domain.UserToken; import org.springframework.data.cassandra.repository.ReactiveCassandraRepository; import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest; +import org.springframework.data.domain.Pageable; +import org.springframework.data.domain.Slice; +import org.springframework.data.domain.Sort; import org.springframework.data.repository.query.ExtensionAwareQueryMethodEvaluationContextProvider; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringRunner; @@ -61,8 +72,7 @@ public class SimpleReactiveCassandraRepositoryIntegrationTests extends AbstractK } } - @Autowired - private ReactiveCassandraOperations operations; + @Autowired private ReactiveCassandraOperations operations; private BeanFactory beanFactory; private ClassLoader classLoader; @@ -212,6 +222,37 @@ public class SimpleReactiveCassandraRepositoryIntegrationTests extends AbstractK repository.findAllById(Flux.empty()).as(StepVerifier::create).verifyComplete(); } + @Test // DATACASS-700 + public void findAllWithPagingAndSorting() { + + UserTokenRepostitory repository = factory.getRepository(UserTokenRepostitory.class); + repository.deleteAll(); + + UUID id = UUID.randomUUID(); + List users = IntStream.range(0, 100).mapToObj(value -> { + + UserToken token = new UserToken(); + token.setUserId(id); + token.setToken(UUID.randomUUID()); + + return token; + }).collect(Collectors.toList()); + + repository.saveAll(users).then().as(StepVerifier::create).verifyComplete(); + + List result = new ArrayList<>(); + Slice slice = repository.findAllByUserId(id, CassandraPageRequest.first(10, Sort.by("token"))) + .block(Duration.ofSeconds(10)); + + while (!slice.isEmpty() || slice.hasNext()) { + + result.addAll(slice.getContent()); + slice = repository.findAllByUserId(id, slice.nextPageable()).block(Duration.ofSeconds(10)); + } + + assertThat(result).hasSize(100); + } + @Test // DATACASS-335 public void countShouldReturnNumberOfRecords() { @@ -383,6 +424,10 @@ public class SimpleReactiveCassandraRepositoryIntegrationTests extends AbstractK repository.findById(boyd.getId()).as(StepVerifier::create).expectNextCount(0).verifyComplete(); } - interface UserRepostitory extends ReactiveCassandraRepository { } + interface UserRepostitory extends ReactiveCassandraRepository {} + + interface UserTokenRepostitory extends ReactiveCassandraRepository { + Mono> findAllByUserId(UUID id, Pageable pageRequest); + } }