From 58050405a344ee041313cc9622ef37d04ca6db46 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Fri, 7 Jul 2017 16:09:35 +0200 Subject: [PATCH] DATAMONGO-1646 - Support reactive aggregation execution. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit We now support reactive aggregation execution via ReactiveMongoOperations.aggregate(…). Original Pull Request: #481 --- .../mongodb/core/ReactiveMongoOperations.java | 70 +++++++++ .../mongodb/core/ReactiveMongoTemplate.java | 84 ++++++++++ .../data/mongodb/core/aggregation/City.java | 6 + .../aggregation/ReactiveAggregationTests.java | 144 ++++++++++++++++++ 4 files changed, 304 insertions(+) create mode 100644 spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/ReactiveAggregationTests.java diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoOperations.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoOperations.java index 218c54563..b0d6b32f4 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoOperations.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoOperations.java @@ -24,6 +24,9 @@ import org.bson.Document; import org.reactivestreams.Publisher; import org.reactivestreams.Subscription; import org.springframework.data.geo.GeoResult; +import org.springframework.data.mongodb.core.aggregation.Aggregation; +import org.springframework.data.mongodb.core.aggregation.AggregationOptions; +import org.springframework.data.mongodb.core.aggregation.TypedAggregation; import org.springframework.data.mongodb.core.convert.MappingMongoConverter; import org.springframework.data.mongodb.core.convert.MongoConverter; import org.springframework.data.mongodb.core.index.ReactiveIndexOperations; @@ -370,6 +373,73 @@ public interface ReactiveMongoOperations { */ Mono findById(Object id, Class entityClass, String collectionName); + /** + * Execute an aggregation operation. + *

+ * The raw results will be mapped to the given entity class. The name of the inputCollection is derived from the + * inputType of the aggregation. + *

+ * Aggregation streaming can't be used with {@link AggregationOptions#isExplain() aggregation explain}. Enabling + * explanation mode will throw an {@link IllegalArgumentException}. + * + * @param aggregation The {@link TypedAggregation} specification holding the aggregation operations, must not be + * {@literal null}. + * @param collectionName The name of the input collection to use for the aggreation. + * @param outputType The parametrized type of the returned list, must not be {@literal null}. + * @return The results of the aggregation operation. + */ + Flux aggregate(TypedAggregation aggregation, String collectionName, Class outputType); + + /** + * Execute an aggregation operation. + *

+ * The raw results will be mapped to the given entity class and are returned as stream. The name of the + * inputCollection is derived from the inputType of the aggregation. + *

+ * Aggregation streaming can't be used with {@link AggregationOptions#isExplain() aggregation explain}. Enabling + * explanation mode will throw an {@link IllegalArgumentException}. + * + * @param aggregation The {@link TypedAggregation} specification holding the aggregation operations, must not be + * {@literal null}. + * @param outputType The parametrized type of the returned list, must not be {@literal null}. + * @return The results of the aggregation operation. + */ + Flux aggregate(TypedAggregation aggregation, Class outputType); + + /** + * Execute an aggregation operation. + *

+ * The raw results will be mapped to the given entity class. + *

+ * Aggregation streaming can't be used with {@link AggregationOptions#isExplain() aggregation explain}. Enabling + * explanation mode will throw an {@link IllegalArgumentException}. + * + * @param aggregation The {@link Aggregation} specification holding the aggregation operations, must not be + * {@literal null}. + * @param inputType the inputType where the aggregation operation will read from, must not be {@literal null} or + * empty. + * @param outputType The parametrized type of the returned list, must not be {@literal null}. + * @return The results of the aggregation operation. + */ + Flux aggregate(Aggregation aggregation, Class inputType, Class outputType); + + /** + * Execute an aggregation operation. + *

+ * The raw results will be mapped to the given entity class. + *

+ * Aggregation streaming can't be used with {@link AggregationOptions#isExplain() aggregation explain}. Enabling + * explanation mode will throw an {@link IllegalArgumentException}. + * + * @param aggregation The {@link Aggregation} specification holding the aggregation operations, must not be + * {@literal null}. + * @param collectionName the collection where the aggregation operation will read from, must not be {@literal null} or + * empty. + * @param outputType The parametrized type of the returned list, must not be {@literal null}. + * @return The results of the aggregation operation. + */ + Flux aggregate(Aggregation aggregation, String collectionName, Class outputType); + /** * Returns {@link Flux} of {@link GeoResult} for all entities matching the given {@link NearQuery}. Will consider * entity mapping information to determine the collection the query is ran against. Note, that MongoDB limits the diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java index 280241b1b..8dec3d83e 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java @@ -65,6 +65,11 @@ import org.springframework.data.mapping.context.MappingContext; import org.springframework.data.mapping.model.ConvertingPropertyAccessor; import org.springframework.data.mongodb.MongoDbFactory; import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory; +import org.springframework.data.mongodb.core.aggregation.Aggregation; +import org.springframework.data.mongodb.core.aggregation.AggregationOperationContext; +import org.springframework.data.mongodb.core.aggregation.AggregationOptions; +import org.springframework.data.mongodb.core.aggregation.TypeBasedAggregationOperationContext; +import org.springframework.data.mongodb.core.aggregation.TypedAggregation; import org.springframework.data.mongodb.core.convert.DbRefProxyHandler; import org.springframework.data.mongodb.core.convert.DbRefResolver; import org.springframework.data.mongodb.core.convert.DbRefResolverCallback; @@ -119,6 +124,7 @@ import com.mongodb.client.model.ReturnDocument; import com.mongodb.client.model.UpdateOptions; import com.mongodb.client.result.DeleteResult; import com.mongodb.client.result.UpdateResult; +import com.mongodb.reactivestreams.client.AggregatePublisher; import com.mongodb.reactivestreams.client.FindPublisher; import com.mongodb.reactivestreams.client.MongoClient; import com.mongodb.reactivestreams.client.MongoCollection; @@ -628,6 +634,84 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati return doFindOne(collectionName, new Document(idKey, id), null, entityClass, null); } + /* (non-Javadoc) + * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#aggregate(org.springframework.data.mongodb.core.aggregation.TypedAggregation, java.lang.String, java.lang.Class) + */ + @Override + public Flux aggregate(TypedAggregation aggregation, String inputCollectionName, Class outputType) { + + Assert.notNull(aggregation, "Aggregation pipeline must not be null!"); + + AggregationOperationContext context = new TypeBasedAggregationOperationContext(aggregation.getInputType(), + mappingContext, queryMapper); + return aggregate(aggregation, inputCollectionName, outputType, context); + } + + /* (non-Javadoc) + * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#aggregate(org.springframework.data.mongodb.core.aggregation.TypedAggregation, java.lang.Class) + */ + @Override + public Flux aggregate(TypedAggregation aggregation, Class outputType) { + return aggregate(aggregation, determineCollectionName(aggregation.getInputType()), outputType); + } + + /* (non-Javadoc) + * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#aggregate(org.springframework.data.mongodb.core.aggregation.Aggregation, java.lang.Class, java.lang.Class) + */ + @Override + public Flux aggregate(Aggregation aggregation, Class inputType, Class outputType) { + + return aggregate(aggregation, determineCollectionName(inputType), outputType, + new TypeBasedAggregationOperationContext(inputType, mappingContext, queryMapper)); + } + + /* (non-Javadoc) + * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#aggregate(org.springframework.data.mongodb.core.aggregation.Aggregation, java.lang.String, java.lang.Class) + */ + @Override + public Flux aggregate(Aggregation aggregation, String collectionName, Class outputType) { + return aggregate(aggregation, collectionName, outputType, null); + } + + protected Flux aggregate(Aggregation aggregation, String collectionName, Class outputType, + AggregationOperationContext context) { + + Assert.hasText(collectionName, "Collection name must not be null or empty!"); + Assert.notNull(aggregation, "Aggregation pipeline must not be null!"); + Assert.notNull(outputType, "Output type must not be null!"); + + AggregationOperationContext rootContext = context == null ? Aggregation.DEFAULT_CONTEXT : context; + + Document command = aggregation.toDocument(collectionName, rootContext); + + Boolean explain = command.get("explain", Boolean.class); + if (explain != null && explain) { + throw new IllegalArgumentException("Can't use explain option with streaming!"); + } + + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("Streaming aggregation: {}", serializeToJsonSafely(command)); + } + + ReadDocumentCallback readCallback = new ReadDocumentCallback<>(mongoConverter, outputType, collectionName); + + return execute(collectionName, collection -> { + + List pipeline = (List) command.get("pipeline"); + + AggregationOptions options = AggregationOptions.fromDocument(command); + + AggregatePublisher cursor = collection.aggregate(pipeline).allowDiskUse(options.isAllowDiskUse()) + .useCursor(true); + + if (options.getCollation().isPresent()) { + cursor = cursor.collation(options.getCollation().map(Collation::toMongoCollation).get()); + } + + return Flux.from(cursor).map(readCallback::doWith); + }); + } + /* * (non-Javadoc) * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#geoNear(org.springframework.data.mongodb.core.query.NearQuery, java.lang.Class) diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/City.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/City.java index dafd837c0..76829cd45 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/City.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/City.java @@ -1,5 +1,11 @@ package org.springframework.data.mongodb.core.aggregation; +import lombok.AllArgsConstructor; +import lombok.NoArgsConstructor; + +@lombok.Data +@AllArgsConstructor +@NoArgsConstructor class City { String name; diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/ReactiveAggregationTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/ReactiveAggregationTests.java new file mode 100644 index 000000000..b94e9937a --- /dev/null +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/ReactiveAggregationTests.java @@ -0,0 +1,144 @@ +/* + * 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.mongodb.core.aggregation; + +import static org.assertj.core.api.AssertionsForInterfaceTypes.*; +import static org.springframework.data.mongodb.core.aggregation.Aggregation.*; +import static org.springframework.data.mongodb.core.query.Criteria.*; + +import reactor.test.StepVerifier; + +import java.util.Arrays; + +import org.bson.Document; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.data.mongodb.core.ReactiveMongoTemplate; +import org.springframework.data.mongodb.core.query.Query; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * Integration test for aggregation via {@link org.springframework.data.mongodb.core.ReactiveMongoTemplate}. + * + * @author Mark Paluch + */ +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration("classpath:reactive-infrastructure.xml") +public class ReactiveAggregationTests { + + private static final String INPUT_COLLECTION = "aggregation_test_collection"; + private static final String OUTPUT_COLLECTION = "aggregation_test_out"; + + @Autowired ReactiveMongoTemplate reactiveMongoTemplate; + + @Before + public void setUp() { + cleanDb(); + } + + @After + public void cleanUp() { + cleanDb(); + } + + private void cleanDb() { + StepVerifier + .create(reactiveMongoTemplate.dropCollection(INPUT_COLLECTION) // + .then(reactiveMongoTemplate.dropCollection(OUTPUT_COLLECTION)) // + .then(reactiveMongoTemplate.dropCollection(Product.class)) // + .then(reactiveMongoTemplate.dropCollection(City.class))) // + .verifyComplete(); + } + + @Test(expected = IllegalArgumentException.class) // DATAMONGO-1646 + public void shouldHandleMissingInputCollection() { + reactiveMongoTemplate.aggregate(newAggregation(), (String) null, TagCount.class); + } + + @Test(expected = IllegalArgumentException.class) // DATAMONGO-1646 + public void shouldHandleMissingAggregationPipeline() { + reactiveMongoTemplate.aggregate(null, INPUT_COLLECTION, TagCount.class); + } + + @Test(expected = IllegalArgumentException.class) // DATAMONGO-1646 + public void shouldHandleMissingEntityClass() { + reactiveMongoTemplate.aggregate(newAggregation(), INPUT_COLLECTION, null); + } + + @Test // DATAMONGO-1646 + public void expressionsInProjectionExampleShowcase() { + + Product product = new Product("P1", "A", 1.99, 3, 0.05, 0.19); + StepVerifier.create(reactiveMongoTemplate.insert(product)).expectNextCount(1).verifyComplete(); + + double shippingCosts = 1.2; + + TypedAggregation agg = newAggregation(Product.class, // + project("name", "netPrice") // + .andExpression("netPrice * 10", shippingCosts).as("salesPrice") // + ); + + StepVerifier.create(reactiveMongoTemplate.aggregate(agg, Document.class)).consumeNextWith(actual -> { + + assertThat(actual).containsEntry("_id", product.id); + assertThat(actual).containsEntry("name", product.name); + assertThat(actual).containsEntry("salesPrice", product.netPrice * 10); + }).verifyComplete(); + } + + @Test // DATAMONGO-1646 + public void shouldProjectMultipleDocuments() { + + City dresden = new City("Dresden", 100); + City linz = new City("Linz", 101); + City braunschweig = new City("Braunschweig", 102); + City weinheim = new City("Weinheim", 103); + + StepVerifier.create(reactiveMongoTemplate.insertAll(Arrays.asList(dresden, linz, braunschweig, weinheim))) + .expectNextCount(4).verifyComplete(); + + Aggregation agg = newAggregation( // + match(where("population").lt(103))); + + StepVerifier.create(reactiveMongoTemplate.aggregate(agg, "city", City.class).collectList()) + .consumeNextWith(actual -> { + assertThat(actual).hasSize(3).contains(dresden, linz, braunschweig); + }).verifyComplete(); + } + + @Test // DATAMONGO-1646 + public void shouldAggregateToOutCollection() { + + City dresden = new City("Dresden", 100); + City linz = new City("Linz", 101); + City braunschweig = new City("Braunschweig", 102); + City weinheim = new City("Weinheim", 103); + + StepVerifier.create(reactiveMongoTemplate.insertAll(Arrays.asList(dresden, linz, braunschweig, weinheim))) + .expectNextCount(4).verifyComplete(); + + Aggregation agg = newAggregation( // + out(OUTPUT_COLLECTION)); + + StepVerifier.create(reactiveMongoTemplate.aggregate(agg, "city", City.class)).expectNextCount(4).verifyComplete(); + StepVerifier.create(reactiveMongoTemplate.find(new Query(), City.class, OUTPUT_COLLECTION)).expectNextCount(4) + .verifyComplete(); + } +}