From 34ad06d6a770e4d6d2bcd0aa564b9fc3e508e455 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Wed, 13 Nov 2019 11:30:44 +0100 Subject: [PATCH] DATACASS-700 - Retain Sort in paging through repository queries. SlicedExecution in repository query methods associates the next Pageable with Sort to propagate the pagination configuration across multiple paging requests. Previously, Sort was not considered causing a corrupt paging state. --- .../core/query/CassandraPageRequest.java | 14 +++++ .../query/CassandraQueryExecution.java | 11 +++- .../ReactiveCassandraQueryExecution.java | 15 ++++- ...leCassandraRepositoryIntegrationTests.java | 42 +++++++++++++- ...veCassandraRepositoryIntegrationTests.java | 57 +++++++++++++++++-- 5 files changed, 130 insertions(+), 9 deletions(-) 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); + } }