DATAMONGO-2153 - Annotated aggregation support.

The repository layer offers means interact with the aggregation framework via annotated repository finder methods. Similar to the JSON based queries a pipeline can be defined via the Aggregation annotation. The definition may contain simple placeholders like `?0` as well as SpEL expression markers `?#{ ... }`.

public interface PersonRepository extends CrudReppsitory<Person, String> {

  @Aggregation("{ $group: { _id : $lastname, names : { $addToSet : $?0 } } }")
  List<PersonAggregate> groupByLastnameAnd(String property);

  @Aggregation("{ $group: { _id : $lastname, names : { $addToSet : $firstname } } }")
  List<PersonAggregate> groupByLastnameAndFirstnames(Sort sort);

  @Aggregation("{ $group: { _id : $lastname, names : { $addToSet : $?0 } } }")
  List<PersonAggregate> groupByLastnameAnd(String property, Pageable page);

  @Aggregation("{ $group : { _id : null, total : { $sum : $age } } }")
  SumValue sumAgeUsingValueWrapper();

  @Aggregation("{ $group : { _id : null, total : { $sum : $age } } }")
  Long sumAge();

  @Aggregation("{ $group : { _id : null, total : { $sum : $age } } }")
  AggregationResults<SumValue> sumAgeRaw();

  @Aggregation("{ '$project': { '_id' : '$lastname' } }")
  List<String> findAllLastnames();
}

public interface ReactivePersonRepository extends ReactiveCrudReppsitory<Person, String> {

  @Aggregation("{ $group: { _id : $lastname, names : { $addToSet : $?0 } } }")
  Flux<PersonAggregate> groupByLastnameAnd(String property);

  @Aggregation("{ $group : { _id : null, total : { $sum : $age } } }")
  Mono<Long> sumAge();

  @Aggregation("{ '$project': { '_id' : '$lastname' } }")
  Flux<String> findAllLastnames();
}

Original pull request: #743.

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/AggregationOperationContext.java

@@ -17,6 +17,7 @@ package org.springframework.data.mongodb.core.aggregation;
 
 import org.bson.Document;
 import org.springframework.data.mongodb.core.aggregation.ExposedFields.FieldReference;
+import org.springframework.lang.Nullable;
 
 /**
  * The context for an {@link AggregationOperation}.
@@ -33,7 +34,20 @@ public interface AggregationOperationContext {
 	 * @param document will never be {@literal null}.
 	 * @return must not be {@literal null}.
 	 */
-	Document getMappedObject(Document document);
+	default Document getMappedObject(Document document) {
+		return getMappedObject(document, null);
+	}
+
+	/**
+	 * Returns the mapped {@link Document}, potentially converting the source considering mapping metadata for the given
+	 * type.
+	 *
+	 * @param document will never be {@literal null}.
+	 * @param type can be {@literal null}.
+	 * @return must not be {@literal null}.
+	 * @since 2.2
+	 */
+	Document getMappedObject(Document document, @Nullable Class<?> type);
 
 	/**
 	 * Returns a {@link FieldReference} for the given field or {@literal null} if the context does not expose the given

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/AggregationOperationRenderer.java

@@ -24,6 +24,7 @@ import org.springframework.data.mongodb.core.aggregation.ExposedFields.ExposedFi
 import org.springframework.data.mongodb.core.aggregation.ExposedFields.FieldReference;
 import org.springframework.data.mongodb.core.aggregation.Fields.AggregationField;
 import org.springframework.data.mongodb.core.aggregation.FieldsExposingAggregationOperation.InheritsFieldsAggregationOperation;
+import org.springframework.lang.Nullable;
 
 /**
  * Rendering support for {@link AggregationOperation} into a {@link List} of {@link org.bson.Document}.
@@ -75,15 +76,16 @@ class AggregationOperationRenderer {
 	 * Simple {@link AggregationOperationContext} that just returns {@link FieldReference}s as is.
 	 *
 	 * @author Oliver Gierke
+	 * @author Christoph Strobl
 	 */
 	private static class NoOpAggregationOperationContext implements AggregationOperationContext {
 
 		/*
 		 * (non-Javadoc)
-		 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperationContext#getMappedObject(org.bson.Document)
+		 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperationContext#getMappedObject(org.bson.Document, java.lang.Class)
 		 */
 		@Override
-		public Document getMappedObject(Document document) {
+		public Document getMappedObject(Document document, @Nullable Class<?> type) {
 			return document;
 		}
 

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/ExposedFieldsAggregationOperationContext.java

@@ -56,11 +56,11 @@ class ExposedFieldsAggregationOperationContext implements AggregationOperationCo
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperationContext#getMappedObject(org.bson.Document)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperationContext#getMappedObject(org.bson.Document, java.lang.Class)
 	 */
 	@Override
-	public Document getMappedObject(Document document) {
-		return rootContext.getMappedObject(document);
+	public Document getMappedObject(Document document, @Nullable Class<?> type) {
+		return rootContext.getMappedObject(document, type);
 	}
 
 	/*

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/NestedDelegatingExpressionAggregationOperationContext.java

@@ -45,11 +45,11 @@ class NestedDelegatingExpressionAggregationOperationContext implements Aggregati
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperationContext#getMappedObject(org.bson.Document)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperationContext#getMappedObject(org.bson.Document, java.lang.Class)
 	 */
 	@Override
-	public Document getMappedObject(Document document) {
-		return delegate.getMappedObject(document);
+	public Document getMappedObject(Document document, Class<?> type) {
+		return delegate.getMappedObject(document, type);
 	}
 
 	/*

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/PrefixingDelegatingAggregationOperationContext.java

@@ -25,6 +25,7 @@ import java.util.Set;
 
 import org.bson.Document;
 import org.springframework.data.mongodb.core.aggregation.ExposedFields.FieldReference;
+import org.springframework.lang.Nullable;
 
 /**
  * {@link AggregationOperationContext} implementation prefixing non-command keys on root level with the given prefix.
@@ -56,11 +57,11 @@ public class PrefixingDelegatingAggregationOperationContext implements Aggregati
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperationContext#getMappedObject(org.bson.Document)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperationContext#getMappedObject(org.bson.Document, java.lang.Class)
 	 */
 	@Override
-	public Document getMappedObject(Document document) {
-		return doPrefix(delegate.getMappedObject(document));
+	public Document getMappedObject(Document document, @Nullable Class<?> type) {
+		return doPrefix(delegate.getMappedObject(document, type));
 	}
 
 	/*

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/TypeBasedAggregationOperationContext.java

@@ -27,6 +27,7 @@ import org.springframework.data.mongodb.core.aggregation.ExposedFields.FieldRefe
 import org.springframework.data.mongodb.core.convert.QueryMapper;
 import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
 import org.springframework.data.mongodb.core.mapping.MongoPersistentProperty;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
 /**
@@ -70,7 +71,16 @@ public class TypeBasedAggregationOperationContext implements AggregationOperatio
 	 */
 	@Override
 	public Document getMappedObject(Document document) {
-		return mapper.getMappedObject(document, mappingContext.getPersistentEntity(type));
+		return getMappedObject(document, type);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperationContext#getMappedObject(org.bson.Document, java.lang.Class)
+	 */
+	@Override
+	public Document getMappedObject(Document document, @Nullable Class<?> type) {
+		return mapper.getMappedObject(document, type != null ? mappingContext.getPersistentEntity(type) : null);
 	}
 
 	/*

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/Aggregation.java

@@ -0,0 +1,127 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.repository;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.springframework.core.annotation.AliasFor;
+import org.springframework.data.annotation.QueryAnnotation;
+
+/**
+ * The {@link Aggregation} annotation can be used to decorate a {@link org.springframework.data.repository.Repository}
+ * query method so that it runs the {@link Aggregation#pipeline()} on invocation. <br />
+ * The pipeline stages are mapped against the {@link org.springframework.data.repository.Repository} domain type to
+ * consider {@link org.springframework.data.mongodb.core.mapping.Field field} mappings and may contain simple
+ * placeholders {@code ?0} as well as {@link org.springframework.expression.spel.standard.SpelExpression
+ * SpelExpressions}. <br />
+ * Query method {@link org.springframework.data.domain.Sort} and {@link org.springframework.data.domain.Pageable}
+ * arguments are applied at the end of the pipeline or can be defined manually as part of it.
+ *
+ * @author Christoph Strobl
+ * @since 2.2
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ ElementType.METHOD, ElementType.ANNOTATION_TYPE })
+@Documented
+@QueryAnnotation
+public @interface Aggregation {
+
+	/**
+	 * Alias for {@link #pipeline()}. Defines the aggregation pipeline to apply.
+	 *
+	 * @return an empty array by default.
+	 * @see #pipeline()
+	 */
+	@AliasFor("pipeline")
+	String[] value() default {};
+
+	/**
+	 * Defines the aggregation pipeline to apply.
+	 *
+	 * <pre class="code">
+	 *
+	 * // aggregation resulting in collection with single value
+	 * &#64;Aggregation("{ '$project': { '_id' : '$lastname' } }")
+	 * List<String> findAllLastnames();
+	 *
+	 * // aggregation with parameter replacement
+	 * &#64;Aggregation("{ '$group': { '_id' : '$lastname', names : { $addToSet : '$?0' } } }")
+	 * List<PersonAggregate> groupByLastnameAnd(String property);
+	 *
+	 * // aggregation with sort in pipeline
+	 * &#64;Aggregation(pipeline = {"{ '$group': { '_id' : '$lastname', names : { $addToSet : '$?0' } } }", "{ '$sort' : { 'lastname' : -1 } }"})
+	 * List<PersonAggregate> groupByLastnameAnd(String property);
+	 *
+	 * // Sort parameter is used for sorting results
+	 * &#64;Aggregation("{ '$group': { '_id' : '$lastname', names : { $addToSet : '$?0' } } }")
+	 * List<PersonAggregate> groupByLastnameAnd(String property, Sort sort);
+	 *
+	 * // Pageable parameter used for sort, skip and limit
+	 * &#64;Aggregation("{ '$group': { '_id' : '$lastname', names : { $addToSet : '$?0' } } }")
+	 * List<PersonAggregate> groupByLastnameAnd(String property, Pageable page);
+	 *
+	 * // Single value result aggregation.
+	 * &#64;Aggregation("{ '$group' : { '_id' : null, 'total' : { $sum: '$age' } } }")
+	 * Long sumAge();
+	 *
+	 * // Single value wrapped in container object
+	 * &#64;Aggregation("{ '$group' : { '_id' : null, 'total' : { $sum: '$age' } } })
+	 * SumAge sumAgeAndReturnAggregationResultWrapperWithConcreteType();
+	 *
+	 * // Raw aggregation result
+	 * &#64;Aggregation("{ '$group' : { '_id' : null, 'total' : { $sum: '$age' } } })
+	 * AggregationResults&lt;org.bson.Document>&gt; sumAgeAndReturnAggregationResultWrapper();
+	 * </pre>
+	 *
+	 * @return an empty array by default.
+	 */
+	@AliasFor("value")
+	String[] pipeline() default {};
+
+	/**
+	 * Defines the collation to apply when executing the aggregation.
+	 *
+	 * <pre class="code">
+	 * // Fixed value
+	 * &#64;Aggregation(pipeline = "...", collation = "en_US")
+	 * List<Entry> findAllByFixedCollation();
+	 *
+	 * // Fixed value as Document
+	 * &#64;Aggregation(pipeline = "...", collation = "{ 'locale' :  'en_US' }")
+	 * List<Entry> findAllByFixedJsonCollation();
+	 *
+	 * // Dynamic value as String
+	 * &#64;Aggregation(pipeline = "...", collation = "?0")
+	 * List<Entry> findAllByDynamicCollation(String collation);
+	 *
+	 * // Dynamic value as Document
+	 * &#64;Aggregation(pipeline = "...", collation = "{ 'locale' :  ?0 }")
+	 * List<Entry> findAllByDynamicJsonCollation(String collation);
+	 *
+	 * // SpEL expression
+	 * &#64;Aggregation(pipeline = "...", collation = "?#{[0]}")
+	 * List<Entry> findAllByDynamicSpElCollation(String collation);
+	 * </pre>
+	 *
+	 * @return an empty {@link String} by default.
+	 * @since 2.2
+	 */
+	String collation() default "";
+}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/AbstractMongoQuery.java

@@ -16,7 +16,6 @@
 package org.springframework.data.mongodb.repository.query;
 
 import org.bson.Document;
-
 import org.springframework.data.mongodb.core.ExecutableFindOperation.ExecutableFind;
 import org.springframework.data.mongodb.core.ExecutableFindOperation.FindWithQuery;
 import org.springframework.data.mongodb.core.ExecutableFindOperation.TerminatingFind;
@@ -32,6 +31,7 @@ import org.springframework.data.repository.query.QueryMethodEvaluationContextPro
 import org.springframework.data.repository.query.RepositoryQuery;
 import org.springframework.data.repository.query.ResultProcessor;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
 /**
@@ -94,22 +94,36 @@ public abstract class AbstractMongoQuery implements RepositoryQuery {
 
 		ConvertingParameterAccessor accessor = new ConvertingParameterAccessor(operations.getConverter(),
 				new MongoParametersParameterAccessor(method, parameters));
+
+		ResultProcessor processor = method.getResultProcessor().withDynamicProjection(accessor);
+		Class<?> typeToRead = processor.getReturnedType().getTypeToRead();
+
+		return processor.processResult(doExecute(method, processor, accessor, typeToRead));
+	}
+
+	/**
+	 * Execute the {@link RepositoryQuery} of the given method with the parameters provided by the
+	 * {@link ConvertingParameterAccessor accessor}
+	 *
+	 * @param method the {@link MongoQueryMethod} invoked. Never {@literal null}.
+	 * @param processor {@link ResultProcessor} for post procession. Never {@literal null}.
+	 * @param accessor for providing invocation arguments. Never {@literal null}.
+	 * @param typeToRead the desired component target type. Can be {@literal null}.
+	 */
+	protected Object doExecute(MongoQueryMethod method, ResultProcessor processor, ConvertingParameterAccessor accessor,
+			@Nullable Class<?> typeToRead) {
+
 		Query query = createQuery(accessor);
 
 		applyQueryMetaAttributesWhenPresent(query);
 		query = applyAnnotatedDefaultSortIfPresent(query);
 		query = applyAnnotatedCollationIfPresent(query, accessor);
 
-		ResultProcessor processor = method.getResultProcessor().withDynamicProjection(accessor);
-		Class<?> typeToRead = processor.getReturnedType().getTypeToRead();
-
 		FindWithQuery<?> find = typeToRead == null //
 				? executableFind //
 				: executableFind.as(typeToRead);
 
-		MongoQueryExecution execution = getExecution(accessor, find);
-
-		return processor.processResult(execution.execute(query));
+		return getExecution(accessor, find).execute(query);
 	}
 
 	private MongoQueryExecution getExecution(ConvertingParameterAccessor accessor, FindWithQuery<?> operation) {

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/AbstractReactiveMongoQuery.java

@@ -20,7 +20,6 @@ import reactor.core.publisher.Mono;
 
 import org.bson.Document;
 import org.reactivestreams.Publisher;
-
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.data.convert.EntityInstantiators;
 import org.springframework.data.mongodb.core.MongoOperations;
@@ -38,6 +37,7 @@ import org.springframework.data.repository.query.QueryMethodEvaluationContextPro
 import org.springframework.data.repository.query.RepositoryQuery;
 import org.springframework.data.repository.query.ResultProcessor;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
 /**
@@ -119,24 +119,39 @@ public abstract class AbstractReactiveMongoQuery implements RepositoryQuery {
 
 		ConvertingParameterAccessor convertingParamterAccessor = new ConvertingParameterAccessor(operations.getConverter(),
 				parameterAccessor);
-		Query query = createQuery(convertingParamterAccessor);
-
-		applyQueryMetaAttributesWhenPresent(query);
-		query = applyAnnotatedDefaultSortIfPresent(query);
-		query = applyAnnotatedCollationIfPresent(query, convertingParamterAccessor);
 
 		ResultProcessor processor = method.getResultProcessor().withDynamicProjection(convertingParamterAccessor);
 		Class<?> typeToRead = processor.getReturnedType().getTypeToRead();
 
+		return doExecute(method, processor, convertingParamterAccessor, typeToRead);
+	}
+
+	/**
+	 * Execute the {@link RepositoryQuery} of the given method with the parameters provided by the
+	 * {@link ConvertingParameterAccessor accessor}
+	 *
+	 * @param method the {@link ReactiveMongoQueryMethod} invoked. Never {@literal null}.
+	 * @param processor {@link ResultProcessor} for post procession. Never {@literal null}.
+	 * @param accessor for providing invocation arguments. Never {@literal null}.
+	 * @param typeToRead the desired component target type. Can be {@literal null}.
+	 */
+	protected Object doExecute(ReactiveMongoQueryMethod method, ResultProcessor processor,
+			ConvertingParameterAccessor accessor, @Nullable Class<?> typeToRead) {
+
+		Query query = createQuery(accessor);
+
+		applyQueryMetaAttributesWhenPresent(query);
+		query = applyAnnotatedDefaultSortIfPresent(query);
+		query = applyAnnotatedCollationIfPresent(query, accessor);
+
 		FindWithQuery<?> find = typeToRead == null //
 				? findOperationWithProjection //
 				: findOperationWithProjection.as(typeToRead);
 
 		String collection = method.getEntityInformation().getCollectionName();
 
-		ReactiveMongoQueryExecution execution = getExecution(convertingParamterAccessor,
+		ReactiveMongoQueryExecution execution = getExecution(accessor,
 				new ResultProcessingConverter(processor, operations, instantiators), find);
-
 		return execution.execute(query, processor.getReturnedType().getDomainType(), collection);
 	}
 

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

@@ -0,0 +1,208 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.repository.query;
+
+import lombok.experimental.UtilityClass;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.bson.Document;
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort.Order;
+import org.springframework.data.mongodb.core.aggregation.Aggregation;
+import org.springframework.data.mongodb.core.aggregation.AggregationOperation;
+import org.springframework.data.mongodb.core.aggregation.AggregationOperationContext;
+import org.springframework.data.mongodb.core.aggregation.AggregationOptions;
+import org.springframework.data.mongodb.core.convert.MongoConverter;
+import org.springframework.data.mongodb.core.query.Collation;
+import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.data.mongodb.util.json.ParameterBindingContext;
+import org.springframework.data.mongodb.util.json.ParameterBindingDocumentCodec;
+import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.lang.Nullable;
+import org.springframework.util.ClassUtils;
+
+/**
+ * Internal utility class to help avoid duplicate code required in both the reactive and the sync {@link Aggregation}
+ * support offered by repositories.
+ *
+ * @author Christoph Strobl
+ * @since 2.2
+ */
+@UtilityClass
+class AggregationUtils {
+
+	private static final ParameterBindingDocumentCodec CODEC = new ParameterBindingDocumentCodec();
+
+	/**
+	 * Apply a collation extracted from the given {@literal collationExpression} to the given
+	 * {@link org.springframework.data.mongodb.core.aggregation.AggregationOptions.Builder}. Potentially replace parameter
+	 * placeholders with values from the {@link ConvertingParameterAccessor accessor}.
+	 *
+	 * @param builder must not be {@literal null}.
+	 * @param collationExpression must not be {@literal null}.
+	 * @param accessor must not be {@literal null}.
+	 * @return the {@link Query} having proper {@link Collation}.
+	 * @see AggregationOptions#getCollation()
+	 * @see CollationUtils#computeCollation(String, ConvertingParameterAccessor, MongoParameters, SpelExpressionParser,
+	 *      QueryMethodEvaluationContextProvider)
+	 * @since 2.2
+	 */
+	static AggregationOptions.Builder applyCollation(AggregationOptions.Builder builder,
+			@Nullable String collationExpression, ConvertingParameterAccessor accessor, MongoParameters parameters,
+			SpelExpressionParser expressionParser, QueryMethodEvaluationContextProvider evaluationContextProvider) {
+
+		Collation collation = CollationUtils.computeCollation(collationExpression, accessor, parameters, expressionParser,
+				evaluationContextProvider);
+		return collation == null ? builder : builder.collation(collation);
+	}
+
+	/**
+	 * Compute the {@link AggregationOperation aggregation} pipeline for the given {@link MongoQueryMethod}. The raw
+	 * {@link org.springframework.data.mongodb.repository.Aggregation#pipeline()} is parsed with a
+	 * {@link ParameterBindingDocumentCodec} to obtain the MongoDB native {@link Document} representation returned by
+	 * {@link AggregationOperation#toDocument(AggregationOperationContext)} that is mapped against the domain type
+	 * properties.
+	 * 
+	 * @param method
+	 * @param accessor
+	 * @param expressionParser
+	 * @param evaluationContextProvider
+	 * @return
+	 */
+	static List<AggregationOperation> computePipeline(MongoQueryMethod method, ConvertingParameterAccessor accessor,
+			SpelExpressionParser expressionParser, QueryMethodEvaluationContextProvider evaluationContextProvider) {
+
+		ParameterBindingContext bindingContext = new ParameterBindingContext((accessor::getBindableValue), expressionParser,
+				evaluationContextProvider.getEvaluationContext(method.getParameters(), accessor.getValues()));
+
+		List<AggregationOperation> target = new ArrayList<>(method.getAnnotatedAggregation().length);
+		for (String source : method.getAnnotatedAggregation()) {
+			target.add(ctx -> ctx.getMappedObject(CODEC.decode(source, bindingContext), method.getRepositoryDomainType()));
+		}
+		return target;
+	}
+
+	/**
+	 * Append {@code $sort} aggregation stage if {@link ConvertingParameterAccessor#getSort()} is present.
+	 *
+	 * @param aggregationPipeline
+	 * @param accessor
+	 * @param targetType
+	 */
+	static void appendSortIfPresent(List<AggregationOperation> aggregationPipeline, ConvertingParameterAccessor accessor,
+			Class<?> targetType) {
+
+		if (accessor.getSort().isUnsorted()) {
+			return;
+		}
+
+		aggregationPipeline.add(ctx -> {
+
+			Document sort = new Document();
+			for (Order order : accessor.getSort()) {
+				sort.append(order.getProperty(), order.isAscending() ? 1 : -1);
+			}
+
+			return ctx.getMappedObject(new Document("$sort", sort), targetType);
+		});
+	}
+
+	/**
+	 * Append {@code $skip} and {@code $limit} aggregation stage if {@link ConvertingParameterAccessor#getSort()} is
+	 * present.
+	 *
+	 * @param aggregationPipeline
+	 * @param accessor
+	 */
+	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()));
+	}
+
+	/**
+	 * Extract a single entry from the given {@link Document}. <br />
+	 * <ol>
+	 * <li><strong>empty source:</strong> {@literal null}</li>
+	 * <li><strong>single entry</strong> convert that one</li>
+	 * <li><strong>single entry when ignoring {@literal _id} field</strong> convert that one</li>
+	 * <li><strong>multiple entries</strong> first value assignable to the target type</li>
+	 * <li><strong>no match</strong> IllegalArgumentException</li>
+	 * </ol>
+	 *
+	 * @param <T>
+	 * @param source
+	 * @param targetType
+	 * @param converter
+	 * @return can be {@literal null} if source {@link Document#isEmpty() is empty}.
+	 * @throws IllegalArgumentException when none of the above rules is met.
+	 */
+	@Nullable
+	static <T> T extractSimpleTypeResult(Document source, Class<T> targetType, MongoConverter converter) {
+
+		if (source.isEmpty()) {
+			return null;
+		}
+
+		if (source.size() == 1) {
+			return getPotentiallyConvertedSimpleTypeValue(converter, source.values().iterator().next(), targetType);
+		}
+
+		Document tmp = new Document(source);
+		tmp.remove("_id");
+
+		if (tmp.size() == 1) {
+			return getPotentiallyConvertedSimpleTypeValue(converter, tmp.values().iterator().next(), targetType);
+		}
+
+		for (Map.Entry<String, Object> entry : tmp.entrySet()) {
+			if (entry != null && ClassUtils.isAssignable(targetType, entry.getValue().getClass())) {
+				return (T) entry.getValue();
+			}
+		}
+
+		throw new IllegalArgumentException(
+				String.format("o_O no entry of type %s found in %s.", targetType.getSimpleName(), source.toJson()));
+	}
+
+	@Nullable
+	private static <T> T getPotentiallyConvertedSimpleTypeValue(MongoConverter converter, Object value,
+			Class<T> targetType) {
+
+		if (value == null) {
+			return (T) value;
+		}
+
+		if (!converter.getConversionService().canConvert(value.getClass(), targetType)) {
+			return (T) value;
+		}
+
+		return converter.getConversionService().convert(value, targetType);
+	}
+}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/CollationUtils.java

@@ -0,0 +1,110 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.repository.query;
+
+import lombok.experimental.UtilityClass;
+
+import java.util.Locale;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.bson.Document;
+import org.springframework.data.mongodb.core.query.Collation;
+import org.springframework.data.mongodb.util.json.ParameterBindingContext;
+import org.springframework.data.mongodb.util.json.ParameterBindingDocumentCodec;
+import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.lang.Nullable;
+import org.springframework.util.NumberUtils;
+import org.springframework.util.ObjectUtils;
+import org.springframework.util.StringUtils;
+
+/**
+ * Internal utility class to help avoid duplicate code required in both the reactive and the sync {@link Collation}
+ * support offered by repositories.
+ *
+ * @author Christoph Strobl
+ * @since 2.2
+ */
+@UtilityClass
+class CollationUtils {
+
+	private static final ParameterBindingDocumentCodec CODEC = new ParameterBindingDocumentCodec();
+	private static final Pattern PARAMETER_BINDING_PATTERN = Pattern.compile("\\?(\\d+)");
+
+	/**
+	 * Compute the {@link Collation} by inspecting the {@link ConvertingParameterAccessor#getCollation() parameter
+	 * accessor} or parsing a potentially given {@literal collationExpression}.
+	 *
+	 * @param collationExpression
+	 * @param accessor
+	 * @param parameters
+	 * @param expressionParser
+	 * @param evaluationContextProvider
+	 * @return can be {@literal null} if neither {@link ConvertingParameterAccessor#getCollation()} nor
+	 *         {@literal collationExpression} are present.
+	 */
+	@Nullable
+	static Collation computeCollation(@Nullable String collationExpression, ConvertingParameterAccessor accessor,
+			MongoParameters parameters, SpelExpressionParser expressionParser,
+			QueryMethodEvaluationContextProvider evaluationContextProvider) {
+
+		if (accessor.getCollation() != null) {
+			return accessor.getCollation();
+		}
+
+		if (!StringUtils.hasText(collationExpression)) {
+			return null;
+		}
+
+		if (StringUtils.trimLeadingWhitespace(collationExpression).startsWith("{")) {
+
+			ParameterBindingContext bindingContext = new ParameterBindingContext((accessor::getBindableValue),
+					expressionParser, evaluationContextProvider.getEvaluationContext(parameters, accessor.getValues()));
+
+			return Collation.from(CODEC.decode(collationExpression, bindingContext));
+		}
+
+		Matcher matcher = PARAMETER_BINDING_PATTERN.matcher(collationExpression);
+		if (!matcher.find()) {
+			return Collation.parse(collationExpression);
+		}
+
+		String placeholder = matcher.group();
+		Object placeholderValue = accessor.getBindableValue(computeParameterIndex(placeholder));
+
+		if (collationExpression.startsWith("?")) {
+
+			if (placeholderValue instanceof String) {
+				return Collation.parse(placeholderValue.toString());
+			}
+			if (placeholderValue instanceof Locale) {
+				return Collation.of((Locale) placeholderValue);
+			}
+			if (placeholderValue instanceof Document) {
+				return Collation.from((Document) placeholderValue);
+			}
+			throw new IllegalArgumentException(String.format("Collation must be a String, Locale or Document but was %s",
+					ObjectUtils.nullSafeClassName(placeholderValue)));
+		}
+
+		return Collation.parse(collationExpression.replace(placeholder, placeholderValue.toString()));
+	}
+
+	private static int computeParameterIndex(String parameter) {
+		return NumberUtils.parseNumber(parameter.replace("?", ""), Integer.class);
+	}
+}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/MongoQueryMethod.java

@@ -30,6 +30,7 @@ import org.springframework.data.geo.GeoResults;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
 import org.springframework.data.mongodb.core.mapping.MongoPersistentProperty;
+import org.springframework.data.mongodb.repository.Aggregation;
 import org.springframework.data.mongodb.repository.Meta;
 import org.springframework.data.mongodb.repository.Query;
 import org.springframework.data.mongodb.repository.Tailable;
@@ -167,6 +168,17 @@ public class MongoQueryMethod extends QueryMethod {
 		return this.metadata;
 	}
 
+	/**
+	 * Get the declared {@link org.springframework.data.repository.Repository} domain type.
+	 *
+	 * @return the domain type declared at repository level.
+	 * @see QueryMethod#getDomainClass()
+	 * @since 2.2
+	 */
+	Class<?> getRepositoryDomainType() {
+		return getDomainClass();
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.repository.query.QueryMethod#getParameters()
@@ -323,27 +335,65 @@ public class MongoQueryMethod extends QueryMethod {
 	}
 
 	/**
-	 * Check if the query method is decorated with an non empty {@link Query#collation()}.
+	 * Check if the query method is decorated with an non empty {@link Query#collation()} or or
+	 * {@link Aggregation#collation()}.
 	 *
-	 * @return true if method annotated with {@link Query} having an non empty collation attribute.
+	 * @return true if method annotated with {@link Query} or {@link Aggregation} having an non empty collation attribute.
 	 * @since 2.2
 	 */
 	public boolean hasAnnotatedCollation() {
-		return lookupQueryAnnotation().map(it -> !it.collation().isEmpty()).orElse(false);
+
+		return lookupQueryAnnotation().map(it -> !it.collation().isEmpty())
+				.orElseGet(() -> lookupAggregationAnnotation().map(it -> !it.collation().isEmpty()).orElse(false));
 	}
 
 	/**
-	 * Get the collation value extracted from the {@link Query} annotation.
+	 * Get the collation value extracted from the {@link Query} or {@link Aggregation} annotation.
 	 *
-	 * @return the {@link Query#collation()} value.
-	 * @throws IllegalStateException if method not annotated with {@link Query}. Make sure to check
+	 * @return the {@link Query#collation()} or or {@link Aggregation#collation()} value.
+	 * @throws IllegalStateException if method not annotated with {@link Query} or {@link Aggregation}. Make sure to check
 	 *           {@link #hasAnnotatedQuery()} first.
 	 * @since 2.2
 	 */
 	public String getAnnotatedCollation() {
 
-		return lookupQueryAnnotation().map(Query::collation).orElseThrow(() -> new IllegalStateException(
-				"Expected to find @Query annotation but did not. Make sure to check hasAnnotatedCollation() before."));
+		return lookupQueryAnnotation().map(Query::collation)
+				.orElseGet(() -> lookupAggregationAnnotation().map(Aggregation::collation) //
+						.orElseThrow(() -> new IllegalStateException(
+								"Expected to find @Query annotation but did not. Make sure to check hasAnnotatedCollation() before.")));
+	}
+
+	/**
+	 * Returns whether the method has an annotated query.
+	 *
+	 * @return true if {@link Aggregation} is present.
+	 * @since 2.2
+	 */
+	public boolean hasAnnotatedAggregation() {
+		return findAnnotatedAggregation().isPresent();
+	}
+
+	/**
+	 * Returns the query string declared in a {@link Query} annotation or {@literal null} if neither the annotation found
+	 * nor the attribute was specified.
+	 *
+	 * @return
+	 * @since 2.2
+	 */
+	@Nullable
+	public String[] getAnnotatedAggregation() {
+		return findAnnotatedAggregation().orElse(null);
+	}
+
+	private Optional<String[]> findAnnotatedAggregation() {
+
+		return lookupAggregationAnnotation() //
+				.map(Aggregation::pipeline) //
+				.filter(it -> !ObjectUtils.isEmpty(it));
+	}
+
+	Optional<Aggregation> lookupAggregationAnnotation() {
+		return doFindAnnotation(Aggregation.class);
 	}
 
 	@SuppressWarnings("unchecked")

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/QueryUtils.java

@@ -15,24 +15,14 @@
  */
 package org.springframework.data.mongodb.repository.query;
 
-import java.util.Locale;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-
 import org.aopalliance.intercept.MethodInterceptor;
 import org.bson.Document;
-
 import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.data.mongodb.core.query.Collation;
 import org.springframework.data.mongodb.core.query.Query;
-import org.springframework.data.mongodb.util.json.ParameterBindingContext;
-import org.springframework.data.mongodb.util.json.ParameterBindingDocumentCodec;
 import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.lang.Nullable;
-import org.springframework.util.NumberUtils;
-import org.springframework.util.ObjectUtils;
-import org.springframework.util.StringUtils;
 
 /**
  * Internal utility class to help avoid duplicate code required in both the reactive and the sync {@link Query} support
@@ -45,10 +35,6 @@ import org.springframework.util.StringUtils;
  */
 class QueryUtils {
 
-	private static final ParameterBindingDocumentCodec CODEC = new ParameterBindingDocumentCodec();
-
-	private static final Pattern PARAMETER_BINDING_PATTERN = Pattern.compile("\\?(\\d+)");
-
 	/**
 	 * Decorate {@link Query} and add a default sort expression to the given {@link Query}. Attributes of the given
 	 * {@code sort} may be overwritten by the sort explicitly defined by the {@link Query} itself.
@@ -93,49 +79,8 @@ class QueryUtils {
 			MongoParameters parameters, SpelExpressionParser expressionParser,
 			QueryMethodEvaluationContextProvider evaluationContextProvider) {
 
-		if (accessor.getCollation() != null) {
-			return query.collation(accessor.getCollation());
-		}
-
-		if (collationExpression == null) {
-			return query;
-		}
-
-		if (StringUtils.trimLeadingWhitespace(collationExpression).startsWith("{")) {
-
-			ParameterBindingContext bindingContext = new ParameterBindingContext((accessor::getBindableValue),
-					expressionParser, evaluationContextProvider.getEvaluationContext(parameters, accessor.getValues()));
-
-			return query.collation(Collation.from(CODEC.decode(collationExpression, bindingContext)));
-		}
-
-		Matcher matcher = PARAMETER_BINDING_PATTERN.matcher(collationExpression);
-		if (!matcher.find()) {
-			return query.collation(Collation.parse(collationExpression));
-		}
-
-		String placeholder = matcher.group();
-		Object placeholderValue = accessor.getBindableValue(computeParameterIndex(placeholder));
-
-		if (collationExpression.startsWith("?")) {
-
-			if (placeholderValue instanceof String) {
-				return query.collation(Collation.parse(placeholderValue.toString()));
-			}
-			if (placeholderValue instanceof Locale) {
-				return query.collation(Collation.of((Locale) placeholderValue));
-			}
-			if (placeholderValue instanceof Document) {
-				return query.collation(Collation.from((Document) placeholderValue));
-			}
-			throw new IllegalArgumentException(String.format("Collation must be a String, Locale or Document but was %s",
-					ObjectUtils.nullSafeClassName(placeholderValue)));
-		}
-
-		return query.collation(Collation.parse(collationExpression.replace(placeholder, placeholderValue.toString())));
-	}
-
-	private static int computeParameterIndex(String parameter) {
-		return NumberUtils.parseNumber(parameter.replace("?", ""), Integer.class);
+		Collation collation = CollationUtils.computeCollation(collationExpression, accessor, parameters, expressionParser,
+				evaluationContextProvider);
+		return collation == null ? query : query.collation(collation);
 	}
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/ReactiveStringBasedAggregation.java

@@ -0,0 +1,163 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.repository.query;
+
+import reactor.core.publisher.Flux;
+
+import java.util.List;
+
+import org.bson.Document;
+import org.springframework.data.mongodb.core.ReactiveMongoOperations;
+import org.springframework.data.mongodb.core.aggregation.Aggregation;
+import org.springframework.data.mongodb.core.aggregation.AggregationOperation;
+import org.springframework.data.mongodb.core.aggregation.AggregationOptions;
+import org.springframework.data.mongodb.core.aggregation.TypedAggregation;
+import org.springframework.data.mongodb.core.convert.MongoConverter;
+import org.springframework.data.mongodb.core.mapping.MongoSimpleTypes;
+import org.springframework.data.mongodb.core.query.BasicQuery;
+import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
+import org.springframework.data.repository.query.ResultProcessor;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.util.ClassUtils;
+
+/**
+ * A reactive {@link org.springframework.data.repository.query.RepositoryQuery} to use a plain JSON String to create an
+ * {@link AggregationOperation aggregation} pipeline to actually execute.
+ *
+ * @author Christoph Strobl
+ * @since 2.2
+ */
+public class ReactiveStringBasedAggregation extends AbstractReactiveMongoQuery {
+
+	private final SpelExpressionParser expressionParser;
+	private final QueryMethodEvaluationContextProvider evaluationContextProvider;
+	private final ReactiveMongoOperations reactiveMongoOperations;
+	private final MongoConverter mongoConverter;
+
+	/**
+	 * @param method must not be {@literal null}.
+	 * @param reactiveMongoOperations must not be {@literal null}.
+	 * @param expressionParser must not be {@literal null}.
+	 * @param evaluationContextProvider must not be {@literal null}.
+	 */
+	public ReactiveStringBasedAggregation(ReactiveMongoQueryMethod method,
+			ReactiveMongoOperations reactiveMongoOperations, SpelExpressionParser expressionParser,
+			QueryMethodEvaluationContextProvider evaluationContextProvider) {
+
+		super(method, reactiveMongoOperations, expressionParser, evaluationContextProvider);
+
+		this.reactiveMongoOperations = reactiveMongoOperations;
+		this.mongoConverter = reactiveMongoOperations.getConverter();
+		this.expressionParser = expressionParser;
+		this.evaluationContextProvider = evaluationContextProvider;
+	}
+
+	/*
+	 * (non-Javascript)
+	 * @see org.springframework.data.mongodb.repository.query.AbstractReactiveMongoQuery#doExecute(org.springframework.data.mongodb.repository.query.ReactiveMongoQueryMethod, org.springframework.data.repository.query.ResultProcessor, org.springframework.data.mongodb.repository.query.ConvertingParameterAccessor, java.lang.Class)
+	 */
+	@Override
+	protected Object doExecute(ReactiveMongoQueryMethod method, ResultProcessor processor,
+			ConvertingParameterAccessor accessor, Class<?> typeToRead) {
+
+		Class<?> sourceType = method.getRepositoryDomainType();
+		Class<?> targetType = typeToRead;
+
+		List<AggregationOperation> pipeline = computePipeline(method, accessor);
+		AggregationUtils.appendSortIfPresent(pipeline, accessor, typeToRead);
+		AggregationUtils.appendLimitAndOffsetIfPresent(pipeline, accessor);
+
+		boolean isSimpleReturnType = isSimpleReturnType(typeToRead);
+		boolean isRawReturnType = ClassUtils.isAssignable(org.bson.Document.class, typeToRead);
+
+		if (isSimpleReturnType || isRawReturnType) {
+			targetType = Document.class;
+		}
+
+		AggregationOptions options = computeOptions(method, accessor);
+		TypedAggregation<?> aggregation = new TypedAggregation<>(sourceType, pipeline, options);
+
+		Flux<?> flux = reactiveMongoOperations.aggregate(aggregation, targetType);
+
+		if (isSimpleReturnType && !isRawReturnType) {
+			return flux.map(it -> AggregationUtils.extractSimpleTypeResult((Document) it, typeToRead, mongoConverter));
+		}
+
+		return flux;
+	}
+
+	private boolean isSimpleReturnType(Class<?> targetType) {
+		return MongoSimpleTypes.HOLDER.isSimpleType(targetType);
+	}
+
+	List<AggregationOperation> computePipeline(MongoQueryMethod method, ConvertingParameterAccessor accessor) {
+		return AggregationUtils.computePipeline(getQueryMethod(), accessor, expressionParser, evaluationContextProvider);
+	}
+
+	private AggregationOptions computeOptions(MongoQueryMethod method, ConvertingParameterAccessor accessor) {
+
+		return AggregationUtils
+				.applyCollation(Aggregation.newAggregationOptions(), method.getAnnotatedCollation(), accessor,
+						method.getParameters(), expressionParser, evaluationContextProvider) //
+				.build();
+	}
+
+	/*
+	 * (non-Javascript)
+	 * @see org.springframework.data.mongodb.repository.query.AbstractReactiveMongoQuery#createQuery(org.springframework.data.mongodb.repository.query.ConvertingParameterAccessor)
+	 */
+	@Override
+	protected Query createQuery(ConvertingParameterAccessor accessor) {
+		return new BasicQuery("{}");
+	}
+
+	/*
+	 * (non-Javascript)
+	 * @see org.springframework.data.mongodb.repository.query.AbstractReactiveMongoQuery#isCountQuery()
+	 */
+	@Override
+	protected boolean isCountQuery() {
+		return false;
+	}
+
+	/*
+	 * (non-Javascript)
+	 * @see org.springframework.data.mongodb.repository.query.AbstractReactiveMongoQuery#isExistsQuery()
+	 */
+	@Override
+	protected boolean isExistsQuery() {
+		return false;
+	}
+
+	/*
+	 * (non-Javascript)
+	 * @see org.springframework.data.mongodb.repository.query.AbstractReactiveMongoQuery#isDeleteQuery()
+	 */
+	@Override
+	protected boolean isDeleteQuery() {
+		return false;
+	}
+
+	/*
+	 * (non-Javascript)
+	 * @see org.springframework.data.mongodb.repository.query.AbstractReactiveMongoQuery#isLimiting()
+	 */
+	@Override
+	protected boolean isLimiting() {
+		return false;
+	}
+}

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

@@ -0,0 +1,180 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.repository.query;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+import org.bson.Document;
+import org.springframework.data.mongodb.core.MongoOperations;
+import org.springframework.data.mongodb.core.aggregation.Aggregation;
+import org.springframework.data.mongodb.core.aggregation.AggregationOperation;
+import org.springframework.data.mongodb.core.aggregation.AggregationOptions;
+import org.springframework.data.mongodb.core.aggregation.AggregationResults;
+import org.springframework.data.mongodb.core.aggregation.TypedAggregation;
+import org.springframework.data.mongodb.core.convert.MongoConverter;
+import org.springframework.data.mongodb.core.mapping.MongoSimpleTypes;
+import org.springframework.data.mongodb.core.query.BasicQuery;
+import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
+import org.springframework.data.repository.query.ResultProcessor;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.util.ClassUtils;
+
+/**
+ * @author Christoph Strobl
+ * @since 2.2
+ */
+public class StringBasedAggregation extends AbstractMongoQuery {
+
+	private final MongoOperations mongoOperations;
+	private final MongoConverter mongoConverter;
+	private final SpelExpressionParser expressionParser;
+	private final QueryMethodEvaluationContextProvider evaluationContextProvider;
+
+	/**
+	 * Creates a new {@link StringBasedAggregation} from the given {@link MongoQueryMethod} and {@link MongoOperations}.
+	 *
+	 * @param method must not be {@literal null}.
+	 * @param mongoOperations must not be {@literal null}.
+	 * @param expressionParser
+	 * @param evaluationContextProvider
+	 */
+	public StringBasedAggregation(MongoQueryMethod method, MongoOperations mongoOperations,
+			SpelExpressionParser expressionParser, QueryMethodEvaluationContextProvider evaluationContextProvider) {
+		super(method, mongoOperations, expressionParser, evaluationContextProvider);
+
+		this.mongoOperations = mongoOperations;
+		this.mongoConverter = mongoOperations.getConverter();
+		this.expressionParser = expressionParser;
+		this.evaluationContextProvider = evaluationContextProvider;
+	}
+
+	/*
+	 * (non-Javascript)
+	 * @see org.springframework.data.mongodb.repository.query.AbstractReactiveMongoQuery#doExecute(org.springframework.data.mongodb.repository.query.MongoQueryMethod, org.springframework.data.repository.query.ResultProcessor, org.springframework.data.mongodb.repository.query.ConvertingParameterAccessor, java.lang.Class)
+	 */
+	@Override
+	protected Object doExecute(MongoQueryMethod method, ResultProcessor resultProcessor,
+			ConvertingParameterAccessor accessor, Class<?> typeToRead) {
+
+		Class<?> sourceType = method.getRepositoryDomainType();
+		Class<?> targetType = typeToRead;
+
+		List<AggregationOperation> pipeline = computePipeline(method, accessor);
+		AggregationUtils.appendSortIfPresent(pipeline, accessor, typeToRead);
+		AggregationUtils.appendLimitAndOffsetIfPresent(pipeline, accessor);
+
+		boolean isSimpleReturnType = isSimpleReturnType(typeToRead);
+		boolean isRawAggregationResult = ClassUtils.isAssignable(AggregationResults.class, typeToRead);
+
+		if (isSimpleReturnType) {
+			targetType = Document.class;
+		} else if (isRawAggregationResult) {
+			targetType = method.getReturnType().getActualType().getComponentType().getType();
+		}
+
+		AggregationOptions options = computeOptions(method, accessor);
+		TypedAggregation<?> aggregation = new TypedAggregation<>(sourceType, pipeline, options);
+
+		AggregationResults<?> result = mongoOperations.aggregate(aggregation, targetType);
+
+		if (isRawAggregationResult) {
+			return result;
+		}
+
+		if (method.isCollectionQuery()) {
+
+			if (isSimpleReturnType) {
+
+				return result.getMappedResults().stream()
+						.map(it -> AggregationUtils.extractSimpleTypeResult((Document) it, typeToRead, mongoConverter))
+						.collect(Collectors.toList());
+			}
+
+			return result.getMappedResults();
+		}
+
+		if (isSimpleReturnType) {
+
+			return AggregationUtils.extractSimpleTypeResult((Document) result.getUniqueMappedResult(), typeToRead,
+					mongoConverter);
+		}
+
+		return result.getUniqueMappedResult();
+	}
+
+	private boolean isSimpleReturnType(Class<?> targetType) {
+		return MongoSimpleTypes.HOLDER.isSimpleType(targetType);
+	}
+
+	List<AggregationOperation> computePipeline(MongoQueryMethod method, ConvertingParameterAccessor accessor) {
+		return AggregationUtils.computePipeline(method, accessor, expressionParser, evaluationContextProvider);
+	}
+
+	private AggregationOptions computeOptions(MongoQueryMethod method, ConvertingParameterAccessor accessor) {
+
+		return AggregationUtils
+				.applyCollation(Aggregation.newAggregationOptions(), method.getAnnotatedCollation(), accessor,
+						method.getParameters(), expressionParser, evaluationContextProvider) //
+				.build();
+	}
+
+	/*
+	 * (non-Javascript)
+	 * @see org.springframework.data.mongodb.repository.query.AbstractMongoQuery#createQuery(org.springframework.data.mongodb.repository.query.ConvertingParameterAccessor)
+	 */
+	@Override
+	protected Query createQuery(ConvertingParameterAccessor accessor) {
+		return new BasicQuery("{}");
+	}
+
+	/*
+	 * (non-Javascript)
+	 * @see org.springframework.data.mongodb.repository.query.AbstractMongoQuery#isCountQuery()
+	 */
+	@Override
+	protected boolean isCountQuery() {
+		return false;
+	}
+
+	/*
+	 * (non-Javascript)
+	 * @see org.springframework.data.mongodb.repository.query.AbstractMongoQuery#isExistsQuery()
+	 */
+	@Override
+	protected boolean isExistsQuery() {
+		return false;
+	}
+
+	/*
+	 * (non-Javascript)
+	 * @see org.springframework.data.mongodb.repository.query.AbstractMongoQuery#isDeleteQuery()
+	 */
+	@Override
+	protected boolean isDeleteQuery() {
+		return false;
+	}
+
+	/*
+	 * (non-Javascript)
+	 * @see org.springframework.data.mongodb.repository.query.AbstractMongoQuery#isLimiting()
+	 */
+	@Override
+	protected boolean isLimiting() {
+		return false;
+	}
+}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/support/MongoRepositoryFactory.java

@@ -30,6 +30,7 @@ import org.springframework.data.mongodb.repository.MongoRepository;
 import org.springframework.data.mongodb.repository.query.MongoEntityInformation;
 import org.springframework.data.mongodb.repository.query.MongoQueryMethod;
 import org.springframework.data.mongodb.repository.query.PartTreeMongoQuery;
+import org.springframework.data.mongodb.repository.query.StringBasedAggregation;
 import org.springframework.data.mongodb.repository.query.StringBasedMongoQuery;
 import org.springframework.data.projection.ProjectionFactory;
 import org.springframework.data.querydsl.QuerydslPredicateExecutor;
@@ -187,6 +188,8 @@ public class MongoRepositoryFactory extends RepositoryFactorySupport {
 				String namedQuery = namedQueries.getQuery(namedQueryName);
 				return new StringBasedMongoQuery(namedQuery, queryMethod, operations, EXPRESSION_PARSER,
 						evaluationContextProvider);
+			} else if (queryMethod.hasAnnotatedAggregation()) {
+				return new StringBasedAggregation(queryMethod, operations, EXPRESSION_PARSER, evaluationContextProvider);
 			} else if (queryMethod.hasAnnotatedQuery()) {
 				return new StringBasedMongoQuery(queryMethod, operations, EXPRESSION_PARSER, evaluationContextProvider);
 			} else {

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/support/ReactiveMongoRepositoryFactory.java

@@ -32,6 +32,7 @@ import org.springframework.data.mongodb.repository.query.MongoEntityInformation;
 import org.springframework.data.mongodb.repository.query.PartTreeMongoQuery;
 import org.springframework.data.mongodb.repository.query.ReactiveMongoQueryMethod;
 import org.springframework.data.mongodb.repository.query.ReactivePartTreeMongoQuery;
+import org.springframework.data.mongodb.repository.query.ReactiveStringBasedAggregation;
 import org.springframework.data.mongodb.repository.query.ReactiveStringBasedMongoQuery;
 import org.springframework.data.projection.ProjectionFactory;
 import org.springframework.data.querydsl.ReactiveQuerydslPredicateExecutor;
@@ -154,6 +155,7 @@ public class ReactiveMongoRepositoryFactory extends ReactiveRepositoryFactorySup
 	 * {@link QueryLookupStrategy} to create {@link PartTreeMongoQuery} instances.
 	 *
 	 * @author Mark Paluch
+	 * @author Christoph Strobl
 	 */
 	@RequiredArgsConstructor(access = AccessLevel.PACKAGE)
 	private static class MongoQueryLookupStrategy implements QueryLookupStrategy {
@@ -177,6 +179,9 @@ public class ReactiveMongoRepositoryFactory extends ReactiveRepositoryFactorySup
 				String namedQuery = namedQueries.getQuery(namedQueryName);
 				return new ReactiveStringBasedMongoQuery(namedQuery, queryMethod, operations, EXPRESSION_PARSER,
 						evaluationContextProvider);
+			} else if (queryMethod.hasAnnotatedAggregation()) {
+				return new ReactiveStringBasedAggregation(queryMethod, operations, EXPRESSION_PARSER,
+						evaluationContextProvider);
 			} else if (queryMethod.hasAnnotatedQuery()) {
 				return new ReactiveStringBasedMongoQuery(queryMethod, operations, EXPRESSION_PARSER, evaluationContextProvider);
 			} else {

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/util/json/ParameterBindingJsonReader.java

@@ -36,13 +36,13 @@ import org.bson.types.Decimal128;
 import org.bson.types.MaxKey;
 import org.bson.types.MinKey;
 import org.bson.types.ObjectId;
-
 import org.springframework.data.spel.EvaluationContextProvider;
 import org.springframework.expression.EvaluationContext;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.lang.Nullable;
 import org.springframework.util.ClassUtils;
 import org.springframework.util.NumberUtils;
+import org.springframework.util.ObjectUtils;
 
 /**
  * Reads a JSON and evaluates placehoders and SpEL expressions. Modified version of <a href=
@@ -387,12 +387,29 @@ public class ParameterBindingJsonReader extends AbstractBsonReader {
 		}
 
 		String computedValue = tokenValue;
+
+		boolean matched = false;
 		while (matcher.find()) {
 
+			matched = true;
 			String group = matcher.group();
 			int index = computeParameterIndex(group);
-			computedValue = computedValue.replace(group, getBindableValueForIndex(index).toString());
+			computedValue = computedValue.replace(group, ObjectUtils.nullSafeToString(getBindableValueForIndex(index)));
+		}
+
+		if (!matched) {
+
+			Matcher regexMatcher = EXPRESSION_BINDING_PATTERN.matcher(tokenValue);
+
+			while (regexMatcher.find()) {
+
+				String binding = regexMatcher.group();
+				String expression = binding.substring(3, binding.length() - 1);
+
+				computedValue = computedValue.replace(binding, ObjectUtils.nullSafeToString(evaluateExpression(expression)));
+			}
 		}
+
 		bindableValue.setValue(computedValue);
 		bindableValue.setType(BsonType.STRING);
 
@@ -1332,7 +1349,7 @@ public class ParameterBindingJsonReader extends AbstractBsonReader {
 		if (patternToken.getType() == JsonTokenType.STRING || patternToken.getType() == JsonTokenType.UNQUOTED_STRING) {
 			return bindableValueFor(patternToken).getValue().toString();
 		}
-		
+
 		throw new JsonParseException("JSON reader expected a string but found '%s'.", patternToken.getValue());
 
 		// Spring Data Customization END

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

@@ -24,6 +24,7 @@ import static org.springframework.data.geo.Metrics.*;
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Optional;
@@ -34,6 +35,7 @@ import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 
+import org.bson.Document;
 import org.hamcrest.Matchers;
 import org.junit.Before;
 import org.junit.Rule;
@@ -60,6 +62,7 @@ import org.springframework.data.geo.Metrics;
 import org.springframework.data.geo.Point;
 import org.springframework.data.geo.Polygon;
 import org.springframework.data.mongodb.core.MongoOperations;
+import org.springframework.data.mongodb.core.aggregation.AggregationResults;
 import org.springframework.data.mongodb.core.geo.GeoJsonPoint;
 import org.springframework.data.mongodb.core.query.BasicQuery;
 import org.springframework.data.mongodb.core.query.Query;
@@ -1275,6 +1278,7 @@ public abstract class AbstractPersonRepositoryIntegrationTests {
 
 	@Test // DATAMONGO-2149, DATAMONGO-2154, DATAMONGO-2199
 	public void annotatedQueryShouldAllowPositionalParameterInFieldsProjectionWithDbRef() {
+
 		List<User> userList = IntStream.range(0, 10).mapToObj(it -> {
 
 			User user = new User();
@@ -1294,4 +1298,71 @@ public abstract class AbstractPersonRepositoryIntegrationTests {
 		assertThat(target).isNotNull();
 		assertThat(target.getFans()).hasSize(1);
 	}
+
+	@Test // DATAMONGO-2153
+	public void findListOfSingleValue() {
+
+		assertThat(repository.findAllLastnames()) //
+				.contains("Lessard") //
+				.contains("Keys") //
+				.contains("Tinsley") //
+				.contains("Beauford") //
+				.contains("Moore") //
+				.contains("Matthews"); //
+	}
+
+	@Test // DATAMONGO-2153
+	public void annotatedAggregationWithPlaceholderValue() {
+
+		assertThat(repository.groupByLastnameAnd("firstname"))
+				.contains(new PersonAggregate("Lessard", Collections.singletonList("Stefan"))) //
+				.contains(new PersonAggregate("Keys", Collections.singletonList("Alicia"))) //
+				.contains(new PersonAggregate("Tinsley", Collections.singletonList("Boyd"))) //
+				.contains(new PersonAggregate("Beauford", Collections.singletonList("Carter"))) //
+				.contains(new PersonAggregate("Moore", Collections.singletonList("Leroi"))) //
+				.contains(new PersonAggregate("Matthews", Arrays.asList("Dave", "Oliver August")));
+	}
+
+	@Test // DATAMONGO-2153
+	public void annotatedAggregationWithSort() {
+
+		assertThat(repository.groupByLastnameAnd("firstname", Sort.by("lastname"))) //
+				.containsSequence( //
+						new PersonAggregate("Beauford", Collections.singletonList("Carter")), //
+						new PersonAggregate("Keys", Collections.singletonList("Alicia")), //
+						new PersonAggregate("Lessard", Collections.singletonList("Stefan")), //
+						new PersonAggregate("Matthews", Arrays.asList("Dave", "Oliver August")), //
+						new PersonAggregate("Moore", Collections.singletonList("Leroi")), //
+						new PersonAggregate("Tinsley", Collections.singletonList("Boyd")));
+	}
+
+	@Test // DATAMONGO-2153
+	public void annotatedAggregationWithPageable() {
+
+		assertThat(repository.groupByLastnameAnd("firstname", PageRequest.of(1, 2, Sort.by("lastname")))) //
+				.containsExactly( //
+						new PersonAggregate("Lessard", Collections.singletonList("Stefan")), //
+						new PersonAggregate("Matthews", Arrays.asList("Dave", "Oliver August")));
+	}
+
+	@Test // DATAMONGO-2153
+	public void annotatedAggregationWithSingleSimpleResult() {
+		assertThat(repository.sumAge()).isInstanceOf(Long.class).isEqualTo(245L);
+	}
+
+	@Test // DATAMONGO-2153
+	public void annotatedAggregationWithAggregationResultAsReturnType() {
+
+		assertThat(repository.sumAgeAndReturnAggregationResultWrapper()) //
+				.isInstanceOf(AggregationResults.class) //
+				.containsExactly(new Document("_id", null).append("total", 245));
+	}
+
+	@Test // DATAMONGO-2153
+	public void annotatedAggregationWithAggregationResultAsReturnTypeAndProjection() {
+
+		assertThat(repository.sumAgeAndReturnAggregationResultWrapperWithConcreteType()) //
+				.isInstanceOf(AggregationResults.class) //
+				.containsExactly(new SumAge(245L));
+	}
 }

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/PersonAggregate.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.repository;
+
+import lombok.Value;
+
+import java.util.List;
+
+import org.springframework.data.annotation.Id;
+
+/**
+ * @author Christoph Strobl
+ */
+@Value
+public class PersonAggregate {
+
+	@Id private String lastname;
+	private List<String> names;
+}

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

@@ -35,6 +35,8 @@ import org.springframework.data.geo.GeoPage;
 import org.springframework.data.geo.GeoResults;
 import org.springframework.data.geo.Point;
 import org.springframework.data.geo.Polygon;
+import org.springframework.data.mongodb.core.aggregation.AggregationResults;
+import org.springframework.data.mongodb.core.mapping.Document;
 import org.springframework.data.mongodb.repository.Person.Sex;
 import org.springframework.data.querydsl.QuerydslPredicateExecutor;
 import org.springframework.data.repository.query.Param;
@@ -363,6 +365,27 @@ public interface PersonRepository extends MongoRepository<Person, String>, Query
 	@Query(value = "{ 'shippingAddresses' : { '$elemMatch' : { 'city' : { '$eq' : 'lnz' } } } }", fields = "{ 'shippingAddresses.$': ?0 }")
 	Person findWithArrayPositionInProjection(int position);
 
-	 @Query(value = "{ 'fans' : { '$elemMatch' : { '$ref' : 'user' } } }", fields = "{ 'fans.$': ?0 }")
-	 Person findWithArrayPositionInProjectionWithDbRef(int position);
+	@Query(value = "{ 'fans' : { '$elemMatch' : { '$ref' : 'user' } } }", fields = "{ 'fans.$': ?0 }")
+	Person findWithArrayPositionInProjectionWithDbRef(int position);
+
+	@Aggregation("{ '$project': { '_id' : '$lastname' } }")
+	List<String> findAllLastnames();
+
+	@Aggregation("{ '$group': { '_id' : '$lastname', names : { $addToSet : '$?0' } } }")
+	List<PersonAggregate> groupByLastnameAnd(String property);
+
+	@Aggregation("{ '$group': { '_id' : '$lastname', names : { $addToSet : '$?0' } } }")
+	List<PersonAggregate> groupByLastnameAnd(String property, Sort sort);
+
+	@Aggregation("{ '$group': { '_id' : '$lastname', names : { $addToSet : '$?0' } } }")
+	List<PersonAggregate> groupByLastnameAnd(String property, Pageable page);
+
+	@Aggregation(pipeline = "{ '$group' : { '_id' : null, 'total' : { $sum: '$age' } } }")
+	Long sumAge();
+
+	@Aggregation(pipeline = "{ '$group' : { '_id' : null, 'total' : { $sum: '$age' } } }")
+	AggregationResults<org.bson.Document> sumAgeAndReturnAggregationResultWrapper();
+
+	@Aggregation(pipeline = "{ '$group' : { '_id' : null, 'total' : { $sum: '$age' } } }")
+	AggregationResults<SumAge> sumAgeAndReturnAggregationResultWrapperWithConcreteType();
 }

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/ReactiveMongoRepositoryTests.java

@@ -27,6 +27,7 @@ import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.TimeUnit;
@@ -156,7 +157,7 @@ public class ReactiveMongoRepositoryTests {
 
 		alicia = new Person("Alicia", "Keys", 30, Sex.FEMALE);
 
-		StepVerifier.create(repository.saveAll(Arrays.asList(oliver, dave, carter, boyd, stefan, leroi, alicia))) //
+		StepVerifier.create(repository.saveAll(Arrays.asList(oliver, carter, boyd, stefan, leroi, alicia, dave))) //
 				.expectNextCount(7) //
 				.verifyComplete();
 	}
@@ -429,6 +430,101 @@ public class ReactiveMongoRepositoryTests {
 				}).verifyComplete();
 	}
 
+	@Test // DATAMONGO-2153
+	public void findListOfSingleValue() {
+
+		repository.findAllLastnames() //
+				.collectList() //
+				.as(StepVerifier::create) //
+				.assertNext(actual -> {
+					assertThat(actual) //
+							.contains("Lessard") //
+							.contains("Keys") //
+							.contains("Tinsley") //
+							.contains("Beauford") //
+							.contains("Moore") //
+							.contains("Matthews");
+				}).verifyComplete();
+	}
+
+	@Test // DATAMONGO-2153
+	public void annotatedAggregationWithPlaceholderValue() {
+
+		repository.groupByLastnameAnd("firstname") //
+				.collectList() //
+				.as(StepVerifier::create) //
+				.assertNext(actual -> {
+					assertThat(actual) //
+							.contains(new PersonAggregate("Lessard", Collections.singletonList("Stefan"))) //
+							.contains(new PersonAggregate("Keys", Collections.singletonList("Alicia"))) //
+							.contains(new PersonAggregate("Tinsley", Collections.singletonList("Boyd"))) //
+							.contains(new PersonAggregate("Beauford", Collections.singletonList("Carter"))) //
+							.contains(new PersonAggregate("Moore", Collections.singletonList("Leroi"))) //
+							.contains(new PersonAggregate("Matthews", Arrays.asList("Dave", "Oliver August")));
+				}).verifyComplete();
+	}
+
+	@Test // DATAMONGO-2153
+	public void annotatedAggregationWithSort() {
+
+		repository.groupByLastnameAnd("firstname", Sort.by("lastname")) //
+				.collectList() //
+				.as(StepVerifier::create) //
+				.assertNext(actual -> {
+					assertThat(actual) //
+							.containsSequence( //
+									new PersonAggregate("Beauford", Collections.singletonList("Carter")), //
+									new PersonAggregate("Keys", Collections.singletonList("Alicia")), //
+									new PersonAggregate("Lessard", Collections.singletonList("Stefan")), //
+									new PersonAggregate("Matthews", Arrays.asList("Dave", "Oliver August")), //
+									new PersonAggregate("Moore", Collections.singletonList("Leroi")), //
+									new PersonAggregate("Tinsley", Collections.singletonList("Boyd")));
+				}) //
+				.verifyComplete();
+	}
+
+	@Test // DATAMONGO-2153
+	public void annotatedAggregationWithPageable() {
+
+		repository.groupByLastnameAnd("firstname", PageRequest.of(1, 2, Sort.by("lastname"))) //
+				.collectList() //
+				.as(StepVerifier::create) //
+				.assertNext(actual -> {
+					assertThat(actual) //
+							.containsExactly( //
+									new PersonAggregate("Lessard", Collections.singletonList("Stefan")), //
+									new PersonAggregate("Matthews", Arrays.asList("Dave", "Oliver August")));
+				}) //
+				.verifyComplete();
+	}
+
+	@Test // DATAMONGO-2153
+	public void annotatedAggregationWithSingleSimpleResult() {
+
+		repository.sumAge() //
+				.as(StepVerifier::create) //
+				.expectNext(245L) //
+				.verifyComplete();
+	}
+
+	@Test // DATAMONGO-2153
+	public void annotatedAggregationWithAggregationResultAsReturnType() {
+
+		repository.sumAgeAndReturnRawResult() //
+				.as(StepVerifier::create) //
+				.expectNext(new org.bson.Document("_id", null).append("total", 245)) //
+				.verifyComplete();
+	}
+
+	@Test // DATAMONGO-2153
+	public void annotatedAggregationWithAggregationResultAsReturnTypeAndProjection() {
+
+		repository.sumAgeAndReturnSumWrapper() //
+				.as(StepVerifier::create) //
+				.expectNext(new SumAge(245L)) //
+				.verifyComplete();
+	}
+
 	interface ReactivePersonRepository
 			extends ReactiveMongoRepository<Person, String>, ReactiveQuerydslPredicateExecutor<Person> {
 
@@ -468,6 +564,27 @@ public class ReactiveMongoRepositoryTests {
 
 		@Query(sort = "{ age : -1 }")
 		Flux<Person> findByAgeGreaterThan(int age, Sort sort);
+
+		@Aggregation("{ '$project': { '_id' : '$lastname' } }")
+		Flux<String> findAllLastnames();
+
+		@Aggregation("{ '$group': { '_id' : '$lastname', names : { $addToSet : '$?0' } } }")
+		Flux<PersonAggregate> groupByLastnameAnd(String property);
+
+		@Aggregation("{ '$group': { '_id' : '$lastname', names : { $addToSet : '$?0' } } }")
+		Flux<PersonAggregate> groupByLastnameAnd(String property, Sort sort);
+
+		@Aggregation("{ '$group': { '_id' : '$lastname', names : { $addToSet : '$?0' } } }")
+		Flux<PersonAggregate> groupByLastnameAnd(String property, Pageable page);
+
+		@Aggregation(pipeline = "{ '$group' : { '_id' : null, 'total' : { $sum: '$age' } } }")
+		Mono<Long> sumAge();
+
+		@Aggregation(pipeline = "{ '$group' : { '_id' : null, 'total' : { $sum: '$age' } } }")
+		Mono<org.bson.Document> sumAgeAndReturnRawResult();
+
+		@Aggregation(pipeline = "{ '$group' : { '_id' : null, 'total' : { $sum: '$age' } } }")
+		Mono<SumAge> sumAgeAndReturnSumWrapper();
 	}
 
 	interface ReactiveContactRepository extends ReactiveMongoRepository<Contact, String> {}

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/SumAge.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.repository;
+
+import lombok.Value;
+
+/**
+ * @author Christoph Strobl
+ */
+@Value
+public class SumAge {
+
+	private Long total;
+}

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/query/MongoQueryMethodUnitTests.java

@@ -22,6 +22,7 @@ import java.lang.reflect.Method;
 import java.util.Collection;
 import java.util.List;
 
+import org.assertj.core.api.Assertions;
 import org.junit.Before;
 import org.junit.Test;
 import org.springframework.data.domain.Pageable;
@@ -33,6 +34,7 @@ import org.springframework.data.geo.Point;
 import org.springframework.data.mongodb.core.User;
 import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
 import org.springframework.data.mongodb.repository.Address;
+import org.springframework.data.mongodb.repository.Aggregation;
 import org.springframework.data.mongodb.repository.Contact;
 import org.springframework.data.mongodb.repository.Meta;
 import org.springframework.data.mongodb.repository.Person;
@@ -217,7 +219,8 @@ public class MongoQueryMethodUnitTests {
 
 		assertThat(method.hasQueryMetaAttributes(), is(true));
 		assertThat(method.getQueryMetaAttributes().getFlags(),
-				containsInAnyOrder(org.springframework.data.mongodb.core.query.Meta.CursorOption.NO_TIMEOUT, org.springframework.data.mongodb.core.query.Meta.CursorOption.SLAVE_OK));
+				containsInAnyOrder(org.springframework.data.mongodb.core.query.Meta.CursorOption.NO_TIMEOUT,
+						org.springframework.data.mongodb.core.query.Meta.CursorOption.SLAVE_OK));
 	}
 
 	@Test // DATAMONGO-1266
@@ -228,6 +231,24 @@ public class MongoQueryMethodUnitTests {
 		assertThat(method.getEntityInformation().getJavaType(), is(typeCompatibleWith(User.class)));
 	}
 
+	@Test // DATAMONGO-2153
+	public void findsAnnotatedAggregation() throws Exception {
+
+		MongoQueryMethod method = queryMethod(PersonRepository.class, "findByAggregation");
+
+		Assertions.assertThat(method.hasAnnotatedAggregation()).isTrue();
+		Assertions.assertThat(method.getAnnotatedAggregation()).hasSize(1);
+	}
+
+	@Test // DATAMONGO-2153
+	public void detectsCollationForAggregation() throws Exception {
+
+		MongoQueryMethod method = queryMethod(PersonRepository.class, "findByAggregationWithCollation");
+
+		Assertions.assertThat(method.hasAnnotatedCollation()).isTrue();
+		Assertions.assertThat(method.getAnnotatedCollation()).isEqualTo("de_AT");
+	}
+
 	private MongoQueryMethod queryMethod(Class<?> repository, String name, Class<?>... parameters) throws Exception {
 
 		Method method = repository.getMethod(name, parameters);
@@ -275,11 +296,19 @@ public class MongoQueryMethodUnitTests {
 		@Meta(flags = { org.springframework.data.mongodb.core.query.Meta.CursorOption.NO_TIMEOUT })
 		List<User> metaWithNoCursorTimeout();
 
-		@Meta(flags = { org.springframework.data.mongodb.core.query.Meta.CursorOption.NO_TIMEOUT, org.springframework.data.mongodb.core.query.Meta.CursorOption.SLAVE_OK })
+		@Meta(flags = { org.springframework.data.mongodb.core.query.Meta.CursorOption.NO_TIMEOUT,
+				org.springframework.data.mongodb.core.query.Meta.CursorOption.SLAVE_OK })
 		List<User> metaWithMultipleFlags();
 
 		// DATAMONGO-1266
 		void deleteByUserName(String userName);
+
+		@Aggregation("{'$group': { _id: '$templateId', maxVersion : { $max : '$version'} } }")
+		List<User> findByAggregation();
+
+		@Aggregation(pipeline = "{'$group': { _id: '$templateId', maxVersion : { $max : '$version'} } }",
+				collation = "de_AT")
+		List<User> findByAggregationWithCollation();
 	}
 
 	interface SampleRepository extends Repository<Contact, Long> {

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/query/ReactiveMongoQueryMethodUnitTests.java

@@ -18,9 +18,13 @@ package org.springframework.data.mongodb.repository.query;
 import static org.hamcrest.Matchers.*;
 import static org.junit.Assert.*;
 
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
 import java.lang.reflect.Method;
 import java.util.List;
 
+import org.assertj.core.api.Assertions;
 import org.junit.Before;
 import org.junit.Test;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
@@ -33,6 +37,7 @@ import org.springframework.data.geo.Point;
 import org.springframework.data.mongodb.core.User;
 import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
 import org.springframework.data.mongodb.repository.Address;
+import org.springframework.data.mongodb.repository.Aggregation;
 import org.springframework.data.mongodb.repository.Contact;
 import org.springframework.data.mongodb.repository.Meta;
 import org.springframework.data.mongodb.repository.Person;
@@ -41,9 +46,6 @@ import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
 import org.springframework.data.repository.Repository;
 import org.springframework.data.repository.core.support.DefaultRepositoryMetadata;
 
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-
 /**
  * Unit test for {@link ReactiveMongoQueryMethod}.
  *
@@ -149,11 +151,29 @@ public class ReactiveMongoQueryMethodUnitTests {
 	@Test // DATAMONGO-1444
 	public void fallsBackToRepositoryDomainTypeIfMethodDoesNotReturnADomainType() throws Exception {
 
-		MongoQueryMethod method = queryMethod(PersonRepository.class, "deleteByUserName", String.class);
+		ReactiveMongoQueryMethod method = queryMethod(PersonRepository.class, "deleteByUserName", String.class);
 
 		assertThat(method.getEntityInformation().getJavaType(), is(typeCompatibleWith(User.class)));
 	}
 
+	@Test // DATAMONGO-2153
+	public void findsAnnotatedAggregation() throws Exception {
+
+		ReactiveMongoQueryMethod method = queryMethod(PersonRepository.class, "findByAggregation");
+
+		Assertions.assertThat(method.hasAnnotatedAggregation()).isTrue();
+		Assertions.assertThat(method.getAnnotatedAggregation()).hasSize(1);
+	}
+
+	@Test // DATAMONGO-2153
+	public void detectsCollationForAggregation() throws Exception {
+
+		ReactiveMongoQueryMethod method = queryMethod(PersonRepository.class, "findByAggregationWithCollation");
+
+		Assertions.assertThat(method.hasAnnotatedCollation()).isTrue();
+		Assertions.assertThat(method.getAnnotatedCollation()).isEqualTo("de_AT");
+	}
+
 	private ReactiveMongoQueryMethod queryMethod(Class<?> repository, String name, Class<?>... parameters)
 			throws Exception {
 
@@ -188,6 +208,13 @@ public class ReactiveMongoQueryMethodUnitTests {
 		Flux<User> metaWithMaxExecutionTime();
 
 		void deleteByUserName(String userName);
+
+		@Aggregation("{'$group': { _id: '$templateId', maxVersion : { $max : '$version'} } }")
+		Flux<User> findByAggregation();
+
+		@Aggregation(pipeline = "{'$group': { _id: '$templateId', maxVersion : { $max : '$version'} } }",
+				collation = "de_AT")
+		Flux<User> findByAggregationWithCollation();
 	}
 
 	interface SampleRepository extends Repository<Contact, Long> {

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/query/ReactiveStringBasedAggregationUnitTests.java

@@ -0,0 +1,229 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.repository.query;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+
+import lombok.Value;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.lang.reflect.Method;
+import java.util.Arrays;
+import java.util.List;
+
+import org.bson.Document;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.domain.Sort.Direction;
+import org.springframework.data.mongodb.core.ReactiveMongoOperations;
+import org.springframework.data.mongodb.core.aggregation.AggregationOperationContext;
+import org.springframework.data.mongodb.core.aggregation.TypeBasedAggregationOperationContext;
+import org.springframework.data.mongodb.core.aggregation.TypedAggregation;
+import org.springframework.data.mongodb.core.convert.DbRefResolver;
+import org.springframework.data.mongodb.core.convert.MappingMongoConverter;
+import org.springframework.data.mongodb.core.convert.MongoConverter;
+import org.springframework.data.mongodb.core.convert.QueryMapper;
+import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
+import org.springframework.data.mongodb.core.query.Collation;
+import org.springframework.data.mongodb.repository.Aggregation;
+import org.springframework.data.mongodb.repository.Person;
+import org.springframework.data.projection.ProjectionFactory;
+import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
+import org.springframework.data.repository.core.support.DefaultRepositoryMetadata;
+import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
+import org.springframework.data.repository.reactive.ReactiveCrudRepository;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.lang.Nullable;
+
+/**
+ * @author Christoph Strobl
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class ReactiveStringBasedAggregationUnitTests {
+
+	SpelExpressionParser PARSER = new SpelExpressionParser();
+
+	@Mock ReactiveMongoOperations operations;
+	@Mock DbRefResolver dbRefResolver;
+	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' } } }";
+	private static final String GROUP_BY_LASTNAME_STRING_WITH_PARAMETER_PLACEHOLDER = "{ '$group': { '_id' : '$lastname', names : { '$addToSet' : '$?0' } } }";
+	private static final String GROUP_BY_LASTNAME_STRING_WITH_SPEL_PARAMETER_PLACEHOLDER = "{ '$group': { '_id' : '$lastname', 'names' : { '$addToSet' : '$?#{[0]}' } } }";
+
+	private static final Document SORT = Document.parse(RAW_SORT_STRING);
+	private static final Document GROUP_BY_LASTNAME = Document.parse(RAW_GROUP_BY_LASTNAME_STRING);
+
+	@Before
+	public void setUp() {
+
+		converter = new MappingMongoConverter(dbRefResolver, new MongoMappingContext());
+		when(operations.getConverter()).thenReturn(converter);
+		when(operations.aggregate(any(TypedAggregation.class), any())).thenReturn(Flux.empty());
+	}
+
+	@Test // DATAMONGO-2153
+	public void plainStringAggregation() {
+
+		AggregationInvocation invocation = executeAggregation("plainStringAggregation");
+
+		assertThat(inputTypeOf(invocation)).isEqualTo(Person.class);
+		assertThat(targetTypeOf(invocation)).isEqualTo(PersonAggregate.class);
+		assertThat(pipelineOf(invocation)).containsExactly(GROUP_BY_LASTNAME, SORT);
+	}
+
+	@Test // DATAMONGO-2153
+	public void plainStringAggregationWithSortParameter() {
+
+		AggregationInvocation invocation = executeAggregation("plainStringAggregation",
+				Sort.by(Direction.DESC, "lastname"));
+
+		assertThat(inputTypeOf(invocation)).isEqualTo(Person.class);
+		assertThat(targetTypeOf(invocation)).isEqualTo(PersonAggregate.class);
+		assertThat(pipelineOf(invocation)).containsExactly(GROUP_BY_LASTNAME, SORT);
+	}
+
+	@Test // DATAMONGO-2153
+	public void replaceParameter() {
+
+		AggregationInvocation invocation = executeAggregation("parameterReplacementAggregation", "firstname");
+
+		assertThat(inputTypeOf(invocation)).isEqualTo(Person.class);
+		assertThat(targetTypeOf(invocation)).isEqualTo(PersonAggregate.class);
+		assertThat(pipelineOf(invocation)).containsExactly(GROUP_BY_LASTNAME);
+	}
+
+	@Test // DATAMONGO-2153
+	public void replaceSpElParameter() {
+
+		AggregationInvocation invocation = executeAggregation("spelParameterReplacementAggregation", "firstname");
+
+		assertThat(inputTypeOf(invocation)).isEqualTo(Person.class);
+		assertThat(targetTypeOf(invocation)).isEqualTo(PersonAggregate.class);
+		assertThat(pipelineOf(invocation)).containsExactly(GROUP_BY_LASTNAME);
+	}
+
+	@Test // DATAMONGO-2153
+	public void aggregateWithCollation() {
+
+		AggregationInvocation invocation = executeAggregation("aggregateWithCollation");
+
+		assertThat(collationOf(invocation)).isEqualTo(Collation.of("de_AT"));
+	}
+
+	@Test // DATAMONGO-2153
+	public void aggregateWithCollationParameter() {
+
+		AggregationInvocation invocation = executeAggregation("aggregateWithCollation", Collation.of("en_US"));
+
+		assertThat(collationOf(invocation)).isEqualTo(Collation.of("en_US"));
+	}
+
+	private AggregationInvocation executeAggregation(String name, Object... args) {
+
+		Class<?>[] argTypes = Arrays.stream(args).map(Object::getClass).toArray(size -> new Class<?>[size]);
+		ReactiveStringBasedAggregation aggregation = createAggregationForMethod(name, argTypes);
+
+		ArgumentCaptor<TypedAggregation> aggregationCaptor = ArgumentCaptor.forClass(TypedAggregation.class);
+		ArgumentCaptor<Class> targetTypeCaptor = ArgumentCaptor.forClass(Class.class);
+
+		Object result = aggregation.execute(args);
+
+		verify(operations).aggregate(aggregationCaptor.capture(), targetTypeCaptor.capture());
+
+		return new AggregationInvocation(aggregationCaptor.getValue(), targetTypeCaptor.getValue(), result);
+	}
+
+	private ReactiveStringBasedAggregation createAggregationForMethod(String name, Class<?>... parameters) {
+
+		try {
+
+			Method method = SampleRepository.class.getMethod(name, parameters);
+			ProjectionFactory factory = new SpelAwareProxyProjectionFactory();
+			ReactiveMongoQueryMethod queryMethod = new ReactiveMongoQueryMethod(method,
+					new DefaultRepositoryMetadata(SampleRepository.class), factory, converter.getMappingContext());
+			return new ReactiveStringBasedAggregation(queryMethod, operations, PARSER,
+					QueryMethodEvaluationContextProvider.DEFAULT);
+
+		} catch (Exception e) {
+			throw new IllegalArgumentException(e.getMessage(), e);
+		}
+	}
+
+	private List<Document> pipelineOf(AggregationInvocation invocation) {
+
+		AggregationOperationContext context = new TypeBasedAggregationOperationContext(
+				invocation.aggregation.getInputType(), converter.getMappingContext(), new QueryMapper(converter));
+
+		return invocation.aggregation.toPipeline(context);
+	}
+
+	private Class<?> inputTypeOf(AggregationInvocation invocation) {
+		return invocation.aggregation.getInputType();
+	}
+
+	@Nullable
+	private Collation collationOf(AggregationInvocation invocation) {
+		return invocation.aggregation.getOptions() != null ? invocation.aggregation.getOptions().getCollation().orElse(null)
+				: null;
+	}
+
+	private Class<?> targetTypeOf(AggregationInvocation invocation) {
+		return invocation.getTargetType();
+	}
+
+	private interface SampleRepository extends ReactiveCrudRepository<Person, Long> {
+
+		@Aggregation({ RAW_GROUP_BY_LASTNAME_STRING, RAW_SORT_STRING })
+		Mono<PersonAggregate> plainStringAggregation();
+
+		@Aggregation(RAW_GROUP_BY_LASTNAME_STRING)
+		Mono<PersonAggregate> plainStringAggregation(Sort sort);
+
+		@Aggregation(GROUP_BY_LASTNAME_STRING_WITH_PARAMETER_PLACEHOLDER)
+		Mono<PersonAggregate> parameterReplacementAggregation(String attribute);
+
+		@Aggregation(GROUP_BY_LASTNAME_STRING_WITH_SPEL_PARAMETER_PLACEHOLDER)
+		Mono<PersonAggregate> spelParameterReplacementAggregation(String arg0);
+
+		@Aggregation(pipeline = RAW_GROUP_BY_LASTNAME_STRING, collation = "de_AT")
+		Mono<PersonAggregate> aggregateWithCollation();
+
+		@Aggregation(pipeline = RAW_GROUP_BY_LASTNAME_STRING, collation = "de_AT")
+		Mono<PersonAggregate> aggregateWithCollation(Collation collation);
+	}
+
+	static class PersonAggregate {
+
+	}
+
+	@Value
+	static class AggregationInvocation {
+
+		final TypedAggregation<?> aggregation;
+		final Class<?> targetType;
+		final Object result;
+	}
+}

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

@@ -0,0 +1,273 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.repository.query;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+
+import lombok.Value;
+
+import java.lang.reflect.Method;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.bson.Document;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.domain.Sort.Direction;
+import org.springframework.data.mongodb.core.MongoOperations;
+import org.springframework.data.mongodb.core.aggregation.AggregationOperationContext;
+import org.springframework.data.mongodb.core.aggregation.AggregationResults;
+import org.springframework.data.mongodb.core.aggregation.TypeBasedAggregationOperationContext;
+import org.springframework.data.mongodb.core.aggregation.TypedAggregation;
+import org.springframework.data.mongodb.core.convert.DbRefResolver;
+import org.springframework.data.mongodb.core.convert.MappingMongoConverter;
+import org.springframework.data.mongodb.core.convert.MongoConverter;
+import org.springframework.data.mongodb.core.convert.QueryMapper;
+import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
+import org.springframework.data.mongodb.core.query.Collation;
+import org.springframework.data.mongodb.repository.Aggregation;
+import org.springframework.data.mongodb.repository.Person;
+import org.springframework.data.projection.ProjectionFactory;
+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.expression.spel.standard.SpelExpressionParser;
+import org.springframework.lang.Nullable;
+
+/**
+ * @author Christoph Strobl
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class StringBasedAggregationUnitTests {
+
+	SpelExpressionParser PARSER = new SpelExpressionParser();
+
+	@Mock MongoOperations operations;
+	@Mock DbRefResolver dbRefResolver;
+	@Mock AggregationResults aggregationResults;
+	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' } } }";
+	private static final String GROUP_BY_LASTNAME_STRING_WITH_PARAMETER_PLACEHOLDER = "{ '$group': { '_id' : '$lastname', names : { '$addToSet' : '$?0' } } }";
+	private static final String GROUP_BY_LASTNAME_STRING_WITH_SPEL_PARAMETER_PLACEHOLDER = "{ '$group': { '_id' : '$lastname', 'names' : { '$addToSet' : '$?#{[0]}' } } }";
+
+	private static final Document SORT = Document.parse(RAW_SORT_STRING);
+	private static final Document GROUP_BY_LASTNAME = Document.parse(RAW_GROUP_BY_LASTNAME_STRING);
+
+	@Before
+	public void setUp() {
+
+		converter = new MappingMongoConverter(dbRefResolver, new MongoMappingContext());
+		when(operations.getConverter()).thenReturn(converter);
+		when(operations.aggregate(any(TypedAggregation.class), any())).thenReturn(aggregationResults);
+	}
+
+	@Test // DATAMONGO-2153
+	public void plainStringAggregation() {
+
+		AggregationInvocation invocation = executeAggregation("plainStringAggregation");
+
+		assertThat(inputTypeOf(invocation)).isEqualTo(Person.class);
+		assertThat(targetTypeOf(invocation)).isEqualTo(PersonAggregate.class);
+		assertThat(pipelineOf(invocation)).containsExactly(GROUP_BY_LASTNAME, SORT);
+	}
+
+	@Test // DATAMONGO-2153
+	public void returnSingleObject() {
+
+		PersonAggregate expected = new PersonAggregate();
+		when(aggregationResults.getUniqueMappedResult()).thenReturn(Collections.singletonList(expected));
+
+		assertThat(executeAggregation("returnSingleEntity").result).isEqualTo(expected);
+	}
+
+	@Test // DATAMONGO-2153
+	public void returnSingleObjectThrowsError() {
+
+		when(aggregationResults.getUniqueMappedResult()).thenThrow(new IllegalArgumentException("o_O"));
+
+		assertThatExceptionOfType(IllegalArgumentException.class)
+				.isThrownBy(() -> executeAggregation("returnSingleEntity"));
+	}
+
+	@Test // DATAMONGO-2153
+	public void returnCollection() {
+
+		List<PersonAggregate> expected = Collections.singletonList(new PersonAggregate());
+		when(aggregationResults.getMappedResults()).thenReturn(expected);
+
+		assertThat(executeAggregation("returnCollection").result).isEqualTo(expected);
+	}
+
+	@Test // DATAMONGO-2153
+	public void returnRawResultType() {
+		assertThat(executeAggregation("returnRawResultType").result).isEqualTo(aggregationResults);
+	}
+
+	@Test // DATAMONGO-2153
+	public void plainStringAggregationWithSortParameter() {
+
+		AggregationInvocation invocation = executeAggregation("plainStringAggregation",
+				Sort.by(Direction.DESC, "lastname"));
+
+		assertThat(inputTypeOf(invocation)).isEqualTo(Person.class);
+		assertThat(targetTypeOf(invocation)).isEqualTo(PersonAggregate.class);
+		assertThat(pipelineOf(invocation)).containsExactly(GROUP_BY_LASTNAME, SORT);
+	}
+
+	@Test // DATAMONGO-2153
+	public void replaceParameter() {
+
+		AggregationInvocation invocation = executeAggregation("parameterReplacementAggregation", "firstname");
+
+		assertThat(inputTypeOf(invocation)).isEqualTo(Person.class);
+		assertThat(targetTypeOf(invocation)).isEqualTo(PersonAggregate.class);
+		assertThat(pipelineOf(invocation)).containsExactly(GROUP_BY_LASTNAME);
+	}
+
+	@Test // DATAMONGO-2153
+	public void replaceSpElParameter() {
+
+		AggregationInvocation invocation = executeAggregation("spelParameterReplacementAggregation", "firstname");
+
+		assertThat(inputTypeOf(invocation)).isEqualTo(Person.class);
+		assertThat(targetTypeOf(invocation)).isEqualTo(PersonAggregate.class);
+		assertThat(pipelineOf(invocation)).containsExactly(GROUP_BY_LASTNAME);
+	}
+
+	@Test // DATAMONGO-2153
+	public void aggregateWithCollation() {
+
+		AggregationInvocation invocation = executeAggregation("aggregateWithCollation");
+
+		assertThat(collationOf(invocation)).isEqualTo(Collation.of("de_AT"));
+	}
+
+	@Test // DATAMONGO-2153
+	public void aggregateWithCollationParameter() {
+
+		AggregationInvocation invocation = executeAggregation("aggregateWithCollation", Collation.of("en_US"));
+
+		assertThat(collationOf(invocation)).isEqualTo(Collation.of("en_US"));
+	}
+
+	private AggregationInvocation executeAggregation(String name, Object... args) {
+
+		Class<?>[] argTypes = Arrays.stream(args).map(Object::getClass).toArray(size -> new Class<?>[size]);
+		StringBasedAggregation aggregation = createAggregationForMethod(name, argTypes);
+
+		ArgumentCaptor<TypedAggregation> aggregationCaptor = ArgumentCaptor.forClass(TypedAggregation.class);
+		ArgumentCaptor<Class> targetTypeCaptor = ArgumentCaptor.forClass(Class.class);
+
+		Object result = aggregation.execute(args);
+
+		verify(operations).aggregate(aggregationCaptor.capture(), targetTypeCaptor.capture());
+
+		return new AggregationInvocation(aggregationCaptor.getValue(), targetTypeCaptor.getValue(), result);
+	}
+
+	private StringBasedAggregation createAggregationForMethod(String name, Class<?>... parameters) {
+
+		try {
+
+			Method method = SampleRepository.class.getMethod(name, parameters);
+			ProjectionFactory factory = new SpelAwareProxyProjectionFactory();
+			MongoQueryMethod queryMethod = new MongoQueryMethod(method, new DefaultRepositoryMetadata(SampleRepository.class),
+					factory, converter.getMappingContext());
+			return new StringBasedAggregation(queryMethod, operations, PARSER, QueryMethodEvaluationContextProvider.DEFAULT);
+
+		} catch (Exception e) {
+			throw new IllegalArgumentException(e.getMessage(), e);
+		}
+	}
+
+	private List<Document> pipelineOf(AggregationInvocation invocation) {
+
+		AggregationOperationContext context = new TypeBasedAggregationOperationContext(
+				invocation.aggregation.getInputType(), converter.getMappingContext(), new QueryMapper(converter));
+
+		return invocation.aggregation.toPipeline(context);
+	}
+
+	private Class<?> inputTypeOf(AggregationInvocation invocation) {
+		return invocation.aggregation.getInputType();
+	}
+
+	@Nullable
+	private Collation collationOf(AggregationInvocation invocation) {
+		return invocation.aggregation.getOptions() != null ? invocation.aggregation.getOptions().getCollation().orElse(null)
+				: null;
+	}
+
+	private Class<?> targetTypeOf(AggregationInvocation invocation) {
+		return invocation.getTargetType();
+	}
+
+	private interface SampleRepository extends Repository<Person, Long> {
+
+		@Aggregation({ RAW_GROUP_BY_LASTNAME_STRING, RAW_SORT_STRING })
+		PersonAggregate plainStringAggregation();
+
+		@Aggregation(RAW_GROUP_BY_LASTNAME_STRING)
+		PersonAggregate plainStringAggregation(Sort sort);
+
+		@Aggregation(RAW_GROUP_BY_LASTNAME_STRING)
+		PersonAggregate returnSingleEntity();
+
+		@Aggregation(RAW_GROUP_BY_LASTNAME_STRING)
+		List<PersonAggregate> returnCollection();
+
+		@Aggregation(RAW_GROUP_BY_LASTNAME_STRING)
+		AggregationResults<PersonAggregate> returnRawResultType();
+
+		@Aggregation(RAW_GROUP_BY_LASTNAME_STRING)
+		AggregationResults<PersonAggregate> returnRawResults();
+
+		@Aggregation(GROUP_BY_LASTNAME_STRING_WITH_PARAMETER_PLACEHOLDER)
+		PersonAggregate parameterReplacementAggregation(String attribute);
+
+		@Aggregation(GROUP_BY_LASTNAME_STRING_WITH_SPEL_PARAMETER_PLACEHOLDER)
+		PersonAggregate spelParameterReplacementAggregation(String arg0);
+
+		@Aggregation(pipeline = RAW_GROUP_BY_LASTNAME_STRING, collation = "de_AT")
+		PersonAggregate aggregateWithCollation();
+
+		@Aggregation(pipeline = RAW_GROUP_BY_LASTNAME_STRING, collation = "de_AT")
+		PersonAggregate aggregateWithCollation(Collation collation);
+	}
+
+	static class PersonAggregate {
+
+	}
+
+	@Value
+	static class AggregationInvocation {
+
+		final TypedAggregation<?> aggregation;
+		final Class<?> targetType;
+		final Object result;
+	}
+}

src/main/asciidoc/new-features.adoc

@@ -17,6 +17,7 @@
 * <<mongo.jsonSchema.generated, JSON Schema generation>> from domain types.
 * SpEL support in for expressions in `@Indexed`.
 * Annotation-based Collation support through `@Document` and `@Query`.
+* <<mongodb.repositories.queries.aggregation, Aggregation framework>> support via repository query methods.
 * Declarative reactive transactions using <<mongo.transactions.reactive-tx-manager, @Transactional>>.
 
 [[new-features.2-1-0]]

src/main/asciidoc/reference/mongo-repositories-aggregation.adoc

@@ -0,0 +1,90 @@
+[[mongodb.repositories.queries.aggregation]]
+=== Aggregation Repository Methods
+
+The repository layer offers means interact with <<mongo.aggregation, the aggregation framework>> via annotated repository
+finder methods. Similar to the <<mongodb.repositories.queries.json-based, JSON based queries>> a pipeline can be defined
+via the `org.springframework.data.mongodb.repository.Aggregation` annotation. The definition may contain simple placeholders
+like `?0` as well as https://docs.spring.io/spring/docs/{springVersion}/spring-framework-reference/core.html#expressions[SpEL expressions]
+`?#{ ... }`.
+
+.Aggregating Repository Method
+====
+[source,java]
+----
+public interface PersonRepository extends CrudReppsitory<Person, String> {
+
+  @Aggregation("{ $group: { _id : $lastname, names : { $addToSet : $?0 } } }")
+  List<PersonAggregate> groupByLastnameAnd(String property); <1>
+
+  @Aggregation("{ $group: { _id : $lastname, names : { $addToSet : $firstname } } }")
+  List<PersonAggregate> groupByLastnameAndFirstnames(Sort sort); <2>
+
+  @Aggregation("{ $group: { _id : $lastname, names : { $addToSet : $?0 } } }")
+  List<PersonAggregate> groupByLastnameAnd(String property, Pageable page); <3>
+
+  @Aggregation("{ $group : { _id : null, total : { $sum : $age } } }")
+  SumValue sumAgeUsingValueWrapper(); <4>
+
+  @Aggregation("{ $group : { _id : null, total : { $sum : $age } } }")
+  Long sumAge(); <5>
+
+  @Aggregation("{ $group : { _id : null, total : { $sum : $age } } }")
+  AggregationResults<SumValue> sumAgeRaw(); <6>
+
+  @Aggregation("{ '$project': { '_id' : '$lastname' } }")
+  List<String> findAllLastnames(); <7>
+}
+----
+[source,java]
+----
+public class PersonAggregate {
+
+  private @Id String lastname; <2>
+  private List<String> names;
+
+  public PersonAggregate(String lastname, List<String> names) {
+     // ...
+  }
+
+  // Getter / Setter omitted
+}
+
+public class SumValue {
+
+  private final Long total; <4> <6>
+
+  public SumValue(Long total) {
+    // ...
+  }
+
+  // Getter omitted
+}
+----
+<1> Replace `?0` with the given value for `property`.
+<2> If `Sort` argument is present, `$sort` is added at the pipelines tail 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> `$skip`, `$limit` and `$sort` can be passed on via a `Pageable` argument. Same as in 2., the operators are applied at
+the pipelines tail.
+<4> Map the result of an aggregation returning a single `Document` to an instance of a desired `SumValue` target type.
+<5> 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 one might consider `AggregationResult` as the methods return type as shown in 4. or 6.
+<6> Obtain the raw `AggregationResults` mapped to the generic target wrapper type `SumValue` or `org.bson.Document`.
+<7> Like in (5) a single value can be directly obtained from mutliple result ``Document``s.
+====
+
+TIP: `@Aggregation` can also be used with <<mongo.reactive.repositories, Reactive Repositories>>.
+
+[NOTE]
+====
+Obtaining simple type single results inspects the returned `Document` and checks for the following
+
+. Only one entry in the document, return it.
+. Two entries, one is the `_id` value. Return the other.
+. Return for the first value assignable to the return type.
+. Throw an execption if none of the above applied.
+====
+
+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.

src/main/asciidoc/reference/mongo-repositories.adoc

@@ -579,6 +579,8 @@ List<FullTextDocument> result = repository.findByTitleOrderByScoreDesc("mongodb"
 
 include::../{spring-data-commons-docs}/repository-projections.adoc[leveloffset=+2]
 
+include::./mongo-repositories-aggregation.adoc[]
+
 [[mongodb.repositories.misc.cdi-integration]]
 == CDI Integration