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

DATACASS-700 - Retain Sort in paging through repository queries.

Browse Source 
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.
This commit is contained in:
Mark Paluch
2019-11-13 11:30:44 +01:00
parent daddb04d38
commit 34ad06d6a7
5 changed files with 130 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

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

15
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraQueryExecution.java
View File

@@ -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<? extends Slice<?>> 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());
});
}
}

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

@@ -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<UserToken> 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<UserToken> result = new ArrayList<>();
Slice<UserToken> 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<User, String> { }
interface UserRepostitory extends CassandraRepository<User, String> {}
interface UserTokenRepostitory extends CassandraRepository<UserToken, UUID> {
Slice<UserToken> findAllByUserId(UUID id, Pageable pageRequest);
}
static class CaptureEventListener extends AbstractCassandraEventListener<User> {

57
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepositoryIntegrationTests.java
View File

@@ -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<UserToken> 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<UserToken> result = new ArrayList<>();
Slice<UserToken> 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<User, String> { }
interface UserRepostitory extends ReactiveCassandraRepository<User, String> {}
interface UserTokenRepostitory extends ReactiveCassandraRepository<UserToken, UUID> {
Mono<Slice<UserToken>> findAllByUserId(UUID id, Pageable pageRequest);
}
}