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

Polishing.

Browse Source 
Let appendLimitAndOffsetIfPresent accept unary operators for adjusting limit/offset values instead of appendModifiedLimitAndOffsetIfPresent. Apply simple type extraction for Slice. Add support for aggregation result streaming.

Extend tests, add author tags, update docs.

See #3543.
Original pull request: #3645.
This commit is contained in:
Mark Paluch
2021-05-11 11:47:36 +02:00
parent 9a48e32565
commit 90d03d92d8
6 changed files with 186 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/AggregationUtils.java
View File

@@ -18,6 +18,8 @@ package org.springframework.data.mongodb.repository.query;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.function.IntUnaryOperator;
import java.util.function.LongUnaryOperator;
import org.bson.Document;
import org.springframework.data.domain.Pageable;
@@ -42,6 +44,7 @@ import org.springframework.util.StringUtils;
*
* @author Christoph Strobl
* @author Mark Paluch
* @author Divya Srivastava
* @since 2.2
*/
abstract class AggregationUtils {
@@ -133,28 +136,22 @@ abstract class AggregationUtils {
*/
static void appendLimitAndOffsetIfPresent(List<AggregationOperation> aggregationPipeline,
ConvertingParameterAccessor accessor) {
Pageable pageable = accessor.getPageable();
if (pageable.isUnpaged()) {
return;
}
if (pageable.getOffset() > 0) {
aggregationPipeline.add(Aggregation.skip(pageable.getOffset()));
}
aggregationPipeline.add(Aggregation.limit(pageable.getPageSize()));
appendLimitAndOffsetIfPresent(aggregationPipeline, accessor, LongUnaryOperator.identity(),
IntUnaryOperator.identity());
}
/**
* Append {@code $skip} and {@code $limit} aggregation stage if {@link ConvertingParameterAccessor#getSort()} is
* present.
*
* @param aggregationPipeline
* @param accessor
* @param offsetOperator
* @param limitOperator
* @since 3.3
*/
static void appendModifiedLimitAndOffsetIfPresent(List<AggregationOperation> aggregationPipeline,
ConvertingParameterAccessor accessor) {
static void appendLimitAndOffsetIfPresent(List<AggregationOperation> aggregationPipeline,
ConvertingParameterAccessor accessor, LongUnaryOperator offsetOperator, IntUnaryOperator limitOperator) {
Pageable pageable = accessor.getPageable();
if (pageable.isUnpaged()) {
@@ -162,10 +159,10 @@ abstract class AggregationUtils {
}
if (pageable.getOffset() > 0) {
aggregationPipeline.add(Aggregation.skip(pageable.getOffset()));
aggregationPipeline.add(Aggregation.skip(offsetOperator.applyAsLong(pageable.getOffset())));
}
aggregationPipeline.add(Aggregation.limit(pageable.getPageSize()+1));
aggregationPipeline.add(Aggregation.limit(limitOperator.applyAsInt(pageable.getPageSize())));
}
/**

74
spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/StringBasedAggregation.java
View File

@@ -17,9 +17,11 @@ package org.springframework.data.mongodb.repository.query;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;
import java.util.function.LongUnaryOperator;
import java.util.stream.Stream;
import org.bson.Document;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.SliceImpl;
import org.springframework.data.mapping.model.SpELExpressionEvaluator;
@@ -42,7 +44,12 @@ import org.springframework.expression.ExpressionParser;
import org.springframework.util.ClassUtils;
/**
* {@link AbstractMongoQuery} implementation to run string-based aggregations using
* {@link org.springframework.data.mongodb.repository.Aggregation}.
*
* @author Christoph Strobl
* @author Divya Srivastava
* @author Mark Paluch
* @since 2.2
*/
public class StringBasedAggregation extends AbstractMongoQuery {
@@ -64,6 +71,12 @@ public class StringBasedAggregation extends AbstractMongoQuery {
ExpressionParser expressionParser, QueryMethodEvaluationContextProvider evaluationContextProvider) {
super(method, mongoOperations, expressionParser, evaluationContextProvider);
if (method.isPageQuery()) {
throw new InvalidMongoDbApiUsageException(String.format(
"Repository aggregation method '%s' does not support '%s' return type. Please use 'Slice' or 'List' instead.",
method.getName(), method.getReturnType().getType().getSimpleName()));
}
this.mongoOperations = mongoOperations;
this.mongoConverter = mongoOperations.getConverter();
this.expressionParser = expressionParser;
@@ -83,10 +96,11 @@ public class StringBasedAggregation extends AbstractMongoQuery {
List<AggregationOperation> pipeline = computePipeline(method, accessor);
AggregationUtils.appendSortIfPresent(pipeline, accessor, typeToRead);
if (method.isSliceQuery()) {
AggregationUtils.appendModifiedLimitAndOffsetIfPresent(pipeline, accessor);
}else{
AggregationUtils.appendLimitAndOffsetIfPresent(pipeline, accessor, LongUnaryOperator.identity(),
limit -> limit + 1);
} else {
AggregationUtils.appendLimitAndOffsetIfPresent(pipeline, accessor);
}
@@ -96,40 +110,45 @@ public class StringBasedAggregation extends AbstractMongoQuery {
if (isSimpleReturnType) {
targetType = Document.class;
} else if (isRawAggregationResult) {
// 🙈
targetType = method.getReturnType().getRequiredActualType().getRequiredComponentType().getType();
}
AggregationOptions options = computeOptions(method, accessor);
TypedAggregation<?> aggregation = new TypedAggregation<>(sourceType, pipeline, options);
AggregationResults<?> result = mongoOperations.aggregate(aggregation, targetType);
if (method.isStreamQuery()) {
Stream<?> stream = mongoOperations.aggregateStream(aggregation, targetType).stream();
if (isSimpleReturnType) {
return stream.map(it -> AggregationUtils.extractSimpleTypeResult((Document) it, typeToRead, mongoConverter));
}
return stream;
}
AggregationResults<Object> result = (AggregationResults<Object>) mongoOperations.aggregate(aggregation, targetType);
if (isRawAggregationResult) {
return result;
}
List<Object> results = result.getMappedResults();
if (method.isCollectionQuery()) {
if (isSimpleReturnType) {
return result.getMappedResults().stream()
.map(it -> AggregationUtils.extractSimpleTypeResult((Document) it, typeToRead, mongoConverter))
.collect(Collectors.toList());
}
return result.getMappedResults();
return isSimpleReturnType ? convertResults(typeToRead, results) : results;
}
List mappedResults = result.getMappedResults();
if(method.isSliceQuery()) {
if (method.isSliceQuery()) {
Pageable pageable = accessor.getPageable();
int pageSize = pageable.getPageSize();
boolean hasNext = mappedResults.size() > pageSize;
return new SliceImpl<Object>(hasNext ? mappedResults.subList(0, pageSize) : mappedResults, pageable, hasNext);
List<Object> resultsToUse = isSimpleReturnType ? convertResults(typeToRead, results) : results;
boolean hasNext = resultsToUse.size() > pageSize;
return new SliceImpl<>(hasNext ? resultsToUse.subList(0, pageSize) : resultsToUse, pageable, hasNext);
}
Object uniqueResult = result.getUniqueMappedResult();
return isSimpleReturnType
@@ -137,6 +156,17 @@ public class StringBasedAggregation extends AbstractMongoQuery {
: uniqueResult;
}
private List<Object> convertResults(Class<?> typeToRead, List<Object> mappedResults) {
List<Object> list = new ArrayList<>(mappedResults.size());
for (Object it : mappedResults) {
Object extractSimpleTypeResult = AggregationUtils.extractSimpleTypeResult((Document) it, typeToRead,
mongoConverter);
list.add(extractSimpleTypeResult);
}
return list;
}
private boolean isSimpleReturnType(Class<?> targetType) {
return MongoSimpleTypes.HOLDER.isSimpleType(targetType);
}

26
spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/AbstractPersonRepositoryIntegrationTests.java
View File

@@ -43,6 +43,7 @@ import org.springframework.dao.IncorrectResultSizeDataAccessException;
import org.springframework.data.domain.Example;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Range;
import org.springframework.data.domain.Slice;
import org.springframework.data.domain.Sort;
@@ -1269,13 +1270,16 @@ public abstract class AbstractPersonRepositoryIntegrationTests {
@Test // DATAMONGO-2153
void findListOfSingleValue() {
assertThat(repository.findAllLastnames()) //
.contains("Lessard") //
.contains("Keys") //
.contains("Tinsley") //
.contains("Beauford") //
.contains("Moore") //
.contains("Matthews"); //
assertThat(repository.findAllLastnames()).contains("Lessard", "Keys", "Tinsley", "Beauford", "Moore", "Matthews");
}
@Test // GH-3543
void findStreamOfSingleValue() {
try (Stream<String> lastnames = repository.findAllLastnamesAsStream()) {
assertThat(lastnames) //
.contains("Lessard", "Keys", "Tinsley", "Beauford", "Moore", "Matthews");
}
}
@Test // DATAMONGO-2153
@@ -1290,6 +1294,14 @@ public abstract class AbstractPersonRepositoryIntegrationTests {
.contains(new PersonAggregate("Matthews", Arrays.asList("Dave", "Oliver August")));
}
@Test // GH-3543
void annotatedAggregationWithPlaceholderAsSlice() {
Slice<PersonAggregate> slice = repository.groupByLastnameAndAsSlice("firstname", Pageable.ofSize(5));
assertThat(slice).hasSize(5);
assertThat(slice.hasNext()).isTrue();
}
@Test // DATAMONGO-2153
void annotatedAggregationWithSort() {

6
spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/PersonRepository.java
View File

@@ -379,9 +379,15 @@ public interface PersonRepository extends MongoRepository<Person, String>, Query
@Aggregation("{ '$project': { '_id' : '$lastname' } }")
List<String> findAllLastnames();
@Aggregation("{ '$project': { '_id' : '$lastname' } }")
Stream<String> findAllLastnamesAsStream();
@Aggregation("{ '$group': { '_id' : '$lastname', names : { $addToSet : '$?0' } } }")
List<PersonAggregate> groupByLastnameAnd(String property);
@Aggregation("{ '$group': { '_id' : '$lastname', names : { $addToSet : '$?0' } } }")
Slice<PersonAggregate> groupByLastnameAndAsSlice(String property, Pageable pageable);
@Aggregation("{ '$group': { '_id' : '$lastname', names : { $addToSet : '$?0' } } }")
List<PersonAggregate> groupByLastnameAnd(String property, Sort sort);

100
spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/query/StringBasedAggregationUnitTests.java
View File

@@ -26,6 +26,7 @@ import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.stream.Stream;
import org.bson.Document;
import org.junit.jupiter.api.BeforeEach;
@@ -36,11 +37,12 @@ import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.quality.Strictness;
import org.springframework.data.domain.Slice;
import org.springframework.data.domain.SliceImpl;
import org.springframework.data.domain.Page;
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.data.domain.Sort;
import org.springframework.data.domain.Sort.Direction;
import org.springframework.data.mongodb.InvalidMongoDbApiUsageException;
@@ -64,6 +66,7 @@ import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
import org.springframework.data.repository.Repository;
import org.springframework.data.repository.core.support.DefaultRepositoryMetadata;
import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
import org.springframework.data.util.CloseableIterator;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.lang.Nullable;
import org.springframework.util.ClassUtils;
@@ -75,17 +78,18 @@ import com.mongodb.MongoClientSettings;
*
* @author Christoph Strobl
* @author Mark Paluch
* @author Divya Srivastava
*/
@ExtendWith(MockitoExtension.class)
@MockitoSettings(strictness = Strictness.LENIENT)
public class StringBasedAggregationUnitTests {
SpelExpressionParser PARSER = new SpelExpressionParser();
private SpelExpressionParser PARSER = new SpelExpressionParser();
@Mock MongoOperations operations;
@Mock DbRefResolver dbRefResolver;
@Mock AggregationResults aggregationResults;
MongoConverter converter;
private MongoConverter converter;
private static final String RAW_SORT_STRING = "{ '$sort' : { 'lastname' : -1 } }";
private static final String RAW_GROUP_BY_LASTNAME_STRING = "{ '$group': { '_id' : '$lastname', 'names' : { '$addToSet' : '$firstname' } } }";
@@ -96,7 +100,7 @@ public class StringBasedAggregationUnitTests {
private static final Document GROUP_BY_LASTNAME = Document.parse(RAW_GROUP_BY_LASTNAME_STRING);
@BeforeEach
public void setUp() {
void setUp() {
converter = new MappingMongoConverter(dbRefResolver, new MongoMappingContext());
when(operations.getConverter()).thenReturn(converter);
@@ -105,7 +109,7 @@ public class StringBasedAggregationUnitTests {
}
@Test // DATAMONGO-2153
public void plainStringAggregation() {
void plainStringAggregation() {
AggregationInvocation invocation = executeAggregation("plainStringAggregation");
@@ -115,7 +119,7 @@ public class StringBasedAggregationUnitTests {
}
@Test // DATAMONGO-2153, DATAMONGO-2449
public void plainStringAggregationConsidersMeta() {
void plainStringAggregationConsidersMeta() {
AggregationInvocation invocation = executeAggregation("plainStringAggregation");
AggregationOptions options = invocation.aggregation.getOptions();
@@ -127,7 +131,7 @@ public class StringBasedAggregationUnitTests {
}
@Test // DATAMONGO-2153, DATAMONGO-2449
public void returnSingleObject() {
void returnSingleObject() {
PersonAggregate expected = new PersonAggregate();
when(aggregationResults.getUniqueMappedResult()).thenReturn(Collections.singletonList(expected));
@@ -144,7 +148,7 @@ public class StringBasedAggregationUnitTests {
}
@Test // DATAMONGO-2153
public void returnSingleObjectThrowsError() {
void returnSingleObjectThrowsError() {
when(aggregationResults.getUniqueMappedResult()).thenThrow(new IllegalArgumentException("o_O"));
@@ -153,7 +157,7 @@ public class StringBasedAggregationUnitTests {
}
@Test // DATAMONGO-2153
public void returnCollection() {
void returnCollection() {
List<PersonAggregate> expected = Collections.singletonList(new PersonAggregate());
when(aggregationResults.getMappedResults()).thenReturn(expected);
@@ -162,7 +166,7 @@ public class StringBasedAggregationUnitTests {
}
@Test // GH-3623
public void returnNullWhenSingleResultIsNotPresent() {
void returnNullWhenSingleResultIsNotPresent() {
when(aggregationResults.getMappedResults()).thenReturn(Collections.emptyList());
@@ -170,12 +174,12 @@ public class StringBasedAggregationUnitTests {
}
@Test // DATAMONGO-2153
public void returnRawResultType() {
void returnRawResultType() {
assertThat(executeAggregation("returnRawResultType").result).isEqualTo(aggregationResults);
}
@Test // DATAMONGO-2153
public void plainStringAggregationWithSortParameter() {
void plainStringAggregationWithSortParameter() {
AggregationInvocation invocation = executeAggregation("plainStringAggregation",
Sort.by(Direction.DESC, "lastname"));
@@ -186,7 +190,7 @@ public class StringBasedAggregationUnitTests {
}
@Test // DATAMONGO-2153
public void replaceParameter() {
void replaceParameter() {
AggregationInvocation invocation = executeAggregation("parameterReplacementAggregation", "firstname");
@@ -196,7 +200,7 @@ public class StringBasedAggregationUnitTests {
}
@Test // DATAMONGO-2153
public void replaceSpElParameter() {
void replaceSpElParameter() {
AggregationInvocation invocation = executeAggregation("spelParameterReplacementAggregation", "firstname");
@@ -206,7 +210,7 @@ public class StringBasedAggregationUnitTests {
}
@Test // DATAMONGO-2153
public void aggregateWithCollation() {
void aggregateWithCollation() {
AggregationInvocation invocation = executeAggregation("aggregateWithCollation");
@@ -214,18 +218,48 @@ public class StringBasedAggregationUnitTests {
}
@Test // DATAMONGO-2153
public void aggregateWithCollationParameter() {
void aggregateWithCollationParameter() {
AggregationInvocation invocation = executeAggregation("aggregateWithCollation", Collation.of("en_US"));
assertThat(collationOf(invocation)).isEqualTo(Collation.of("en_US"));
}
@Test // DATAMONGO-2506
public void aggregationWithSliceReturnType() {
@Test // GH-3543
void aggregationWithSliceReturnType() {
StringBasedAggregation sba = createAggregationForMethod("aggregationWithSliceReturnType", Pageable.class);
Object result = sba.execute(new Object[] { PageRequest.of(0, 1) });
assertThat(result.getClass()).isEqualTo(SliceImpl.class);
assertThat(result).isInstanceOf(Slice.class);
}
@Test // GH-3543
void aggregationWithStreamReturnType() {
when(operations.aggregateStream(any(TypedAggregation.class), any())).thenReturn(new CloseableIterator<Object>() {
@Override
public void close() {
}
@Override
public boolean hasNext() {
return false;
}
@Override
public Object next() {
return null;
}
});
StringBasedAggregation sba = createAggregationForMethod("aggregationWithStreamReturnType", Pageable.class);
Object result = sba.execute(new Object[] { PageRequest.of(0, 1) });
assertThat(result).isInstanceOf(Stream.class);
}
@Test // DATAMONGO-2557
@@ -235,6 +269,21 @@ public class StringBasedAggregationUnitTests {
verify(operations).execute(any());
}
@Test // DATAMONGO-2506
void aggregateRaisesErrorOnInvalidReturnType() {
Method method = ClassUtils.getMethod(UnsupportedRepository.class, "pageIsUnsupported", Pageable.class);
ProjectionFactory factory = new SpelAwareProxyProjectionFactory();
MongoQueryMethod queryMethod = new MongoQueryMethod(method, new DefaultRepositoryMetadata(SampleRepository.class),
factory, converter.getMappingContext());
assertThatExceptionOfType(InvalidMongoDbApiUsageException.class) //
.isThrownBy(() -> new StringBasedAggregation(queryMethod, operations, PARSER,
QueryMethodEvaluationContextProvider.DEFAULT)) //
.withMessageContaining("pageIsUnsupported") //
.withMessageContaining("Page");
}
private AggregationInvocation executeAggregation(String name, Object... args) {
Class<?>[] argTypes = Arrays.stream(args).map(Object::getClass).toArray(Class[]::new);
@@ -320,16 +369,25 @@ public class StringBasedAggregationUnitTests {
@Aggregation(RAW_GROUP_BY_LASTNAME_STRING)
Slice<Person> aggregationWithSliceReturnType(Pageable page);
@Aggregation(RAW_GROUP_BY_LASTNAME_STRING)
Stream<Person> aggregationWithStreamReturnType(Pageable page);
@Aggregation(RAW_GROUP_BY_LASTNAME_STRING)
String simpleReturnType();
}
private interface UnsupportedRepository extends Repository<Person, Long> {
@Aggregation(RAW_GROUP_BY_LASTNAME_STRING)
Page<Person> pageIsUnsupported(Pageable page);
}
static class PersonAggregate {
}
@Value
static class AggregationInvocation {
private static class AggregationInvocation {
TypedAggregation<?> aggregation;
Class<?> targetType;

30
src/main/asciidoc/reference/mongo-repositories-aggregation.adoc
View File

@@ -21,19 +21,22 @@ public interface PersonRepository extends CrudReppsitory<Person, String> {
List<PersonAggregate> groupByLastnameAnd(String property); <3>
@Aggregation("{ $group: { _id : $lastname, names : { $addToSet : ?0 } } }")
List<PersonAggregate> groupByLastnameAnd(String property, Pageable page); <4>
Slice<PersonAggregate> groupByLastnameAnd(String property, Pageable page); <4>
@Aggregation("{ $group: { _id : $lastname, names : { $addToSet : $firstname } } }")
Stream<PersonAggregate> groupByLastnameAndFirstnamesAsStream(); <5>
@Aggregation("{ $group : { _id : null, total : { $sum : $age } } }")
SumValue sumAgeUsingValueWrapper(); <5>
SumValue sumAgeUsingValueWrapper(); <6>
@Aggregation("{ $group : { _id : null, total : { $sum : $age } } }")
Long sumAge(); <6>
Long sumAge(); <7>
@Aggregation("{ $group : { _id : null, total : { $sum : $age } } }")
AggregationResults<SumValue> sumAgeRaw(); <7>
AggregationResults<SumValue> sumAgeRaw(); <8>
@Aggregation("{ '$project': { '_id' : '$lastname' } }")
List<String> findAllLastnames(); <8>
List<String> findAllLastnames(); <9>
}
----
[source,java]
@@ -52,7 +55,7 @@ public class PersonAggregate {
public class SumValue {
private final Long total; <5> <7>
private final Long total; <6> <8>
public SumValue(Long total) {
// ...
@@ -65,12 +68,13 @@ public class SumValue {
<2> If `Sort` argument is present, `$sort` is appended after the declared pipeline stages so that it only affects the order of the final results after having passed all other aggregation stages.
Therefore, the `Sort` properties are mapped against the methods return type `PersonAggregate` which turns `Sort.by("lastname")` into `{ $sort : { '_id', 1 } }` because `PersonAggregate.lastname` is annotated with `@Id`.
<3> Replaces `?0` with the given value for `property` for a dynamic aggregation pipeline.
<4> `$skip`, `$limit` and `$sort` can be passed on via a `Pageable` argument. Same as in <2>, the operators are appended to the pipeline definition.
<5> Map the result of an aggregation returning a single `Document` to an instance of a desired `SumValue` target type.
<6> Aggregations resulting in single document holding just an accumulation result like eg. `$sum` can be extracted directly from the result `Document`.
<4> `$skip`, `$limit` and `$sort` can be passed on via a `Pageable` argument. Same as in <2>, the operators are appended to the pipeline definition. Methods accepting `Pageable` can return `Slice` for easier pagination.
<5> Aggregation methods can return `Stream` to consume results directly from an underlying cursor. Make sure to close the stream after consuming it to release the server-side cursor by either calling `close()` or through `try-with-resources`.
<6> Map the result of an aggregation returning a single `Document` to an instance of a desired `SumValue` target type.
<7> Aggregations resulting in single document holding just an accumulation result like eg. `$sum` can be extracted directly from the result `Document`.
To gain more control, you might consider `AggregationResult` as method return type as shown in <7>.
<7> Obtain the raw `AggregationResults` mapped to the generic target wrapper type `SumValue` or `org.bson.Document`.
<8> Like in <6>, a single value can be directly obtained from multiple result ``Document``s.
<8> Obtain the raw `AggregationResults` mapped to the generic target wrapper type `SumValue` or `org.bson.Document`.
<9> Like in <6>, a single value can be directly obtained from multiple result ``Document``s.
====
In some scenarios, aggregations might require additional options, such as a maximum run time, additional log comments, or the permission to temporarily write data to disk.
@@ -115,5 +119,5 @@ Simple-type single-result inspects the returned `Document` and checks for the fo
. Throw an exception if none of the above is applicable.
====
WARNING: The `Page` return type is not supported for repository methods using `@Aggregation`. However you can use a
`Pageable` argument to add `$skip`, `$limit` and `$sort` to the pipeline.
WARNING: The `Page` return type is not supported for repository methods using `@Aggregation`. However, you can use a
`Pageable` argument to add `$skip`, `$limit` and `$sort` to the pipeline and let the method return `Slice`.