DATAMONGO-2536 - Add option to skip reading aggregation result.

Introduce dedicated AggregationPipeline to encapsulate pipeline stages.

Original pull request: #876.

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

@@ -62,6 +62,7 @@ import org.springframework.data.mongodb.core.QueryOperations.UpdateContext;
 import org.springframework.data.mongodb.core.aggregation.Aggregation;
 import org.springframework.data.mongodb.core.aggregation.AggregationOperationContext;
 import org.springframework.data.mongodb.core.aggregation.AggregationOptions;
+import org.springframework.data.mongodb.core.aggregation.AggregationOptions.ResultOptions;
 import org.springframework.data.mongodb.core.aggregation.AggregationResults;
 import org.springframework.data.mongodb.core.aggregation.TypeBasedAggregationOperationContext;
 import org.springframework.data.mongodb.core.aggregation.TypedAggregation;
@@ -2152,6 +2153,17 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 				aggregateIterable = aggregateIterable.maxTime(options.getMaxTime().toMillis(), TimeUnit.MILLISECONDS);
 			}
 
+			if(ResultOptions.SKIP.equals(options.resultOptions())) {
+
+				// toCollection only allowed for $out and $merge if those are the last stages
+				if(aggregation.getPipeline().isOutOrMerge()) {
+					aggregateIterable.toCollection();
+				} else {
+					aggregateIterable.first();
+				}
+				return new AggregationResults<>(Collections.emptyList(), new Document());
+			}
+
 			MongoIterable<O> iterable = aggregateIterable.map(val -> {
 
 				rawResult.add(val);

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

@@ -26,6 +26,7 @@ import java.util.*;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import java.util.function.Function;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
 import org.bson.BsonValue;
@@ -68,6 +69,7 @@ import org.springframework.data.mongodb.core.QueryOperations.UpdateContext;
 import org.springframework.data.mongodb.core.aggregation.Aggregation;
 import org.springframework.data.mongodb.core.aggregation.AggregationOperationContext;
 import org.springframework.data.mongodb.core.aggregation.AggregationOptions;
+import org.springframework.data.mongodb.core.aggregation.AggregationOptions.ResultOptions;
 import org.springframework.data.mongodb.core.aggregation.PrefixingDelegatingAggregationOperationContext;
 import org.springframework.data.mongodb.core.aggregation.TypeBasedAggregationOperationContext;
 import org.springframework.data.mongodb.core.aggregation.TypedAggregation;
@@ -1004,11 +1006,11 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		}
 
 		ReadDocumentCallback<O> readCallback = new ReadDocumentCallback<>(mongoConverter, outputType, collectionName);
-		return execute(collectionName, collection -> aggregateAndMap(collection, pipeline, options, readCallback,
+		return execute(collectionName, collection -> aggregateAndMap(collection, pipeline, () -> aggregation.getPipeline().isOutOrMerge(), options, readCallback,
 				aggregation instanceof TypedAggregation ? ((TypedAggregation) aggregation).getInputType() : null));
 	}
 
-	private <O> Flux<O> aggregateAndMap(MongoCollection<Document> collection, List<Document> pipeline,
+	private <O> Flux<O> aggregateAndMap(MongoCollection<Document> collection, List<Document> pipeline, Supplier<Boolean> isOutOrMerge,
 			AggregationOptions options, ReadDocumentCallback<O> readCallback, @Nullable Class<?> inputType) {
 
 		AggregatePublisher<Document> cursor = collection.aggregate(pipeline, Document.class)
@@ -1028,6 +1030,13 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 			cursor = cursor.maxTime(options.getMaxTime().toMillis(), TimeUnit.MILLISECONDS);
 		}
 
+		if (ResultOptions.SKIP.equals(options.resultOptions())) {
+			if (isOutOrMerge.get()) {
+				return Flux.from(cursor.toCollection()).map(it -> (O) it);
+			}
+			return Flux.from(cursor.first()).thenMany(Mono.empty());
+		}
+
 		return Flux.from(cursor).concatMap(readCallback::doWith);
 	}
 

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

@@ -96,7 +96,7 @@ public class Aggregation {
 	public static final AggregationOperationContext DEFAULT_CONTEXT = AggregationOperationRenderer.DEFAULT_CONTEXT;
 	public static final AggregationOptions DEFAULT_OPTIONS = newAggregationOptions().build();
 
-	protected final List<AggregationOperation> operations;
+	protected final AggregationPipeline pipeline;
 	private final AggregationOptions options;
 
 	/**
@@ -139,7 +139,7 @@ public class Aggregation {
 	public Aggregation withOptions(AggregationOptions options) {
 
 		Assert.notNull(options, "AggregationOptions must not be null.");
-		return new Aggregation(this.operations, options);
+		return new Aggregation(this.pipeline.getOperations(), options);
 	}
 
 	/**
@@ -202,26 +202,10 @@ public class Aggregation {
 		Assert.notNull(aggregationOperations, "AggregationOperations must not be null!");
 		Assert.notNull(options, "AggregationOptions must not be null!");
 
-		// check $out/$merge is the last operation if it exists
+		this.pipeline = new AggregationPipeline(aggregationOperations);
-		for (AggregationOperation aggregationOperation : aggregationOperations) {
-
-			if (aggregationOperation instanceof OutOperation && !isLast(aggregationOperation, aggregationOperations)) {
-				throw new IllegalArgumentException("The $out operator must be the last stage in the pipeline.");
-			}
-
-			if (aggregationOperation instanceof MergeOperation && !isLast(aggregationOperation, aggregationOperations)) {
-				throw new IllegalArgumentException("The $merge operator must be the last stage in the pipeline.");
-			}
-		}
-
-		this.operations = aggregationOperations;
 		this.options = options;
 	}
 
-	private boolean isLast(AggregationOperation aggregationOperation, List<AggregationOperation> aggregationOperations) {
-		return aggregationOperations.indexOf(aggregationOperation) == aggregationOperations.size() - 1;
-	}
-
 	/**
 	 * Get the {@link AggregationOptions}.
 	 *
@@ -718,7 +702,11 @@ public class Aggregation {
 	 * @since 2.1
 	 */
 	public List<Document> toPipeline(AggregationOperationContext rootContext) {
-		return AggregationOperationRenderer.toDocument(operations, rootContext);
+		return pipeline.toDocuments(rootContext);
+	}
+
+	public AggregationPipeline getPipeline() {
+		return pipeline;
 	}
 
 	/**

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

@@ -54,4 +54,8 @@ public interface AggregationOperation {
 	default List<Document> toPipelineStages(AggregationOperationContext context) {
 		return Collections.singletonList(toDocument(context));
 	}
+
+	default String operator() {
+		return toDocument(Aggregation.DEFAULT_CONTEXT).keySet().iterator().next();
+	}
 }

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

@@ -52,6 +52,7 @@ public class AggregationOptions {
 	private final Optional<Collation> collation;
 	private final Optional<String> comment;
 	private Duration maxTime = Duration.ZERO;
+	private ResultOptions resultOptions = ResultOptions.READ;
 
 	/**
 	 * Creates a new {@link AggregationOptions}.
@@ -219,6 +220,14 @@ public class AggregationOptions {
 		return maxTime;
 	}
 
+	/**
+	 * @return the {@link ResultOptions} to be used when running the {@link Aggregation}. Never {@literal null}.
+	 * @since 3.0
+	 */
+	public ResultOptions resultOptions() {
+		return resultOptions;
+	}
+
 	/**
 	 * Returns a new potentially adjusted copy for the given {@code aggregationCommandObject} with the configuration
 	 * applied.
@@ -309,6 +318,7 @@ public class AggregationOptions {
 		private @Nullable Collation collation;
 		private @Nullable String comment;
 		private @Nullable Duration maxTime;
+		private @Nullable ResultOptions resultOptions;
 
 		/**
 		 * Defines whether to off-load intensive sort-operations to disk.
@@ -399,6 +409,20 @@ public class AggregationOptions {
 			return this;
 		}
 
+		/**
+		 * Run the aggregation, but do NOT read the aggregation result from the store. <br />
+		 * If the expected result of the aggregation is rather large, eg. when using an {@literal $out} operation, this
+		 * option allows to execute the aggregation without having the cursor return the operation result.
+		 *
+		 * @return this.
+		 * @since 3.0
+		 */
+		public Builder skipOutput() {
+
+			this.resultOptions = ResultOptions.SKIP;
+			return this;
+		}
+
 		/**
 		 * Returns a new {@link AggregationOptions} instance with the given configuration.
 		 *
@@ -410,8 +434,26 @@ public class AggregationOptions {
 			if (maxTime != null) {
 				options.maxTime = maxTime;
 			}
+			if (resultOptions != null) {
+				options.resultOptions = resultOptions;
+			}
 
 			return options;
 		}
 	}
+
+	/**
+	 * @since 3.0
+	 */
+	public enum ResultOptions {
+
+		/**
+		 * Just do it!, and do not read the operation result.
+		 */
+		SKIP,
+		/**
+		 * Read the aggregation result from the cursor.
+		 */
+		READ;
+	}
 }

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

@@ -0,0 +1,111 @@
+/*
+ * Copyright 2020 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.core.aggregation;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import org.bson.Document;
+import org.springframework.util.Assert;
+
+/**
+ * The {@link AggregationPipeline} holds the collection of {@link AggregationOperation aggregation stages}.
+ *
+ * @author Christoph Strobl
+ * @since 3.1
+ */
+public class AggregationPipeline {
+
+	private final List<AggregationOperation> pipeline;
+
+	/**
+	 * Create an empty pipeline
+	 */
+	public AggregationPipeline() {
+		this(new ArrayList<>());
+	}
+
+	/**
+	 * Create a new pipeline with given {@link AggregationOperation stages}.
+	 *
+	 * @param aggregationOperations must not be {@literal null}.
+	 */
+	public AggregationPipeline(List<AggregationOperation> aggregationOperations) {
+		pipeline = new ArrayList<>(aggregationOperations);
+	}
+
+	/**
+	 * Append the given {@link AggregationOperation stage} to the pipeline.
+	 *
+	 * @param aggregationOperation must not be {@literal null}.
+	 * @return this.
+	 */
+	public AggregationPipeline add(AggregationOperation aggregationOperation) {
+
+		Assert.notNull(aggregationOperation, "AggregationOperation must not be null!");
+
+		pipeline.add(aggregationOperation);
+		return this;
+	}
+
+	/**
+	 * Get the list of {@link AggregationOperation aggregation stages}.
+	 *
+	 * @return never {@literal null}.
+	 */
+	public List<AggregationOperation> getOperations() {
+		return Collections.unmodifiableList(pipeline);
+	}
+
+	List<Document> toDocuments(AggregationOperationContext context) {
+
+		verify();
+		return AggregationOperationRenderer.toDocument(pipeline, context);
+	}
+
+	/**
+	 * @return {@literal true} if the last aggregation stage is either {@literal $out} or {@literal $merge}.
+	 */
+	public boolean isOutOrMerge() {
+
+		if (pipeline.isEmpty()) {
+			return false;
+		}
+
+		String operator = pipeline.get(pipeline.size() - 1).operator();
+		return operator.equals("$out") || operator.equals("$merge");
+	}
+
+	void verify() {
+
+		// check $out/$merge is the last operation if it exists
+		for (AggregationOperation aggregationOperation : pipeline) {
+
+			if (aggregationOperation instanceof OutOperation && !isLast(aggregationOperation)) {
+				throw new IllegalArgumentException("The $out operator must be the last stage in the pipeline.");
+			}
+
+			if (aggregationOperation instanceof MergeOperation && !isLast(aggregationOperation)) {
+				throw new IllegalArgumentException("The $merge operator must be the last stage in the pipeline.");
+			}
+		}
+	}
+
+	private boolean isLast(AggregationOperation aggregationOperation) {
+		return pipeline.indexOf(aggregationOperation) == pipeline.size() - 1;
+	}
+}

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

@@ -139,7 +139,7 @@ public class AggregationUpdate extends Aggregation implements UpdateDefinition {
 		setOperation.getFields().forEach(it -> {
 			keysTouched.add(it.getName());
 		});
-		operations.add(setOperation);
+		pipeline.add(setOperation);
 		return this;
 	}
 
@@ -155,7 +155,7 @@ public class AggregationUpdate extends Aggregation implements UpdateDefinition {
 
 		Assert.notNull(unsetOperation, "UnsetOperation must not be null!");
 
-		operations.add(unsetOperation);
+		pipeline.add(unsetOperation);
 		keysTouched.addAll(unsetOperation.removedFieldNames());
 		return this;
 	}
@@ -172,7 +172,7 @@ public class AggregationUpdate extends Aggregation implements UpdateDefinition {
 	public AggregationUpdate replaceWith(ReplaceWithOperation replaceWithOperation) {
 
 		Assert.notNull(replaceWithOperation, "ReplaceWithOperation must not be null!");
-		operations.add(replaceWithOperation);
+		pipeline.add(replaceWithOperation);
 		return this;
 	}
 

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

@@ -106,7 +106,12 @@ public class BucketAutoOperation extends BucketOperationSupport<BucketAutoOperat
 
 		options.putAll(super.toDocument(context));
 
-		return new Document("$bucketAuto", options);
+		return new Document(operator(), options);
+	}
+
+	@Override
+	public String operator() {
+		return "$bucketAuto";
 	}
 
 	/**

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

@@ -103,7 +103,12 @@ public class BucketOperation extends BucketOperationSupport<BucketOperation, Buc
 
 		options.putAll(super.toDocument(context));
 
-		return new Document("$bucket", options);
+		return new Document(operator(), options);
+	}
+
+	@Override
+	public String operator() {
+		return "$bucket";
 	}
 
 	/**

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

@@ -49,7 +49,12 @@ public class CountOperation implements FieldsExposingAggregationOperation {
 	 */
 	@Override
 	public Document toDocument(AggregationOperationContext context) {
-		return new Document("$count", fieldName);
+		return new Document(operator(), fieldName);
+	}
+
+	@Override
+	public String operator() {
+		return "$count";
 	}
 
 	/* (non-Javadoc)

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

@@ -71,6 +71,11 @@ abstract class DocumentEnhancingOperation implements InheritsFieldsAggregationOp
 	 */
 	protected abstract String mongoOperator();
 
+	@Override
+	public String operator() {
+		return mongoOperator();
+	}
+
 	/**
 	 * @return the raw value map
 	 */

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

@@ -84,7 +84,12 @@ public class FacetOperation implements FieldsExposingAggregationOperation {
 	 */
 	@Override
 	public Document toDocument(AggregationOperationContext context) {
-		return new Document("$facet", facets.toDocument(context));
+		return new Document(operator(), facets.toDocument(context));
+	}
+
+	@Override
+	public String operator() {
+		return "$facet";
 	}
 
 	/* (non-Javadoc)

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

@@ -109,7 +109,12 @@ public class GeoNearOperation implements AggregationOperation {
 			command.put("key", indexKey);
 		}
 
-		return new Document("$geoNear", command);
+		return new Document(operator(), command);
+	}
+
+	@Override
+	public String operator() {
+		return "$geoNear";
 	}
 
 	/*

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

@@ -119,7 +119,12 @@ public class GraphLookupOperation implements InheritsFieldsAggregationOperation
 			graphLookup.put("restrictSearchWithMatch", context.getMappedObject(restrictSearchWithMatch.getCriteriaObject()));
 		}
 
-		return new Document("$graphLookup", graphLookup);
+		return new Document(operator(), graphLookup);
+	}
+
+	@Override
+	public String operator() {
+		return "$graphLookup";
 	}
 
 	/*

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

@@ -429,7 +429,12 @@ public class GroupOperation implements FieldsExposingAggregationOperation {
 			operationObject.putAll(operation.toDocument(context));
 		}
 
-		return new Document("$group", operationObject);
+		return new Document(operator(), operationObject);
+	}
+
+	@Override
+	public String operator() {
+		return "$group";
 	}
 
 	interface Keyword {

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

@@ -49,6 +49,11 @@ public class LimitOperation implements AggregationOperation {
 	 */
 	@Override
 	public Document toDocument(AggregationOperationContext context) {
-		return new Document("$limit", Long.valueOf(maxElements));
+		return new Document(operator(), Long.valueOf(maxElements));
+	}
+
+	@Override
+	public String operator() {
+		return "$limit";
 	}
 }

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

@@ -83,7 +83,12 @@ public class LookupOperation implements FieldsExposingAggregationOperation, Inhe
 		lookupObject.append("foreignField", foreignField.getTarget());
 		lookupObject.append("as", as.getTarget());
 
-		return new Document("$lookup", lookupObject);
+		return new Document(operator(), lookupObject);
+	}
+
+	@Override
+	public String operator() {
+		return "$lookup";
 	}
 
 	/**

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

@@ -53,6 +53,11 @@ public class MatchOperation implements AggregationOperation {
 	 */
 	@Override
 	public Document toDocument(AggregationOperationContext context) {
-		return new Document("$match", context.getMappedObject(criteriaDefinition.getCriteriaObject()));
+		return new Document(operator(), context.getMappedObject(criteriaDefinition.getCriteriaObject()));
+	}
+
+	@Override
+	public String operator() {
+		return "$match";
 	}
 }

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

@@ -100,7 +100,7 @@ public class MergeOperation implements FieldsExposingAggregationOperation, Inher
 	public Document toDocument(AggregationOperationContext context) {
 
 		if (isJustCollection()) {
-			return new Document("$merge", into.collection);
+			return new Document(operator(), into.collection);
 		}
 
 		Document $merge = new Document();
@@ -122,7 +122,12 @@ public class MergeOperation implements FieldsExposingAggregationOperation, Inher
 			$merge.putAll(whenNotMatched.toDocument(context));
 		}
 
-		return new Document("$merge", $merge);
+		return new Document(operator(), $merge);
+	}
+
+	@Override
+	public String operator() {
+		return "$merge";
 	}
 
 	/*

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

@@ -211,7 +211,12 @@ public class OutOperation implements AggregationOperation {
 			$out.append("uniqueKey", uniqueKey);
 		}
 
-		return new Document("$out", $out);
+		return new Document(operator(), $out);
+	}
+
+	@Override
+	public String operator() {
+		return "$out";
 	}
 
 	private boolean requiresMongoDb42Format() {

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

@@ -261,7 +261,12 @@ public class ProjectionOperation implements FieldsExposingAggregationOperation {
 			fieldObject.putAll(projection.toDocument(context));
 		}
 
-		return new Document("$project", fieldObject);
+		return new Document(operator(), fieldObject);
+	}
+
+	@Override
+	public String operator() {
+		return "$project";
 	}
 
 	/**

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

@@ -74,7 +74,12 @@ public class RedactOperation implements AggregationOperation {
 	 */
 	@Override
 	public Document toDocument(AggregationOperationContext context) {
-		return new Document("$redact", condition.toDocument(context));
+		return new Document(operator(), condition.toDocument(context));
+	}
+
+	@Override
+	public String operator() {
+		return "$redact";
 	}
 
 	/**

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

@@ -86,6 +86,11 @@ public class ReplaceRootOperation implements FieldsExposingAggregationOperation
 		return new Document("$replaceRoot", new Document("newRoot", getReplacement().toDocumentExpression(context)));
 	}
 
+	@Override
+	public String operator() {
+		return "$replaceRoot";
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.mongodb.core.aggregation.FieldsExposingAggregationOperation#getFields()
 	 */

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

@@ -48,6 +48,11 @@ public class SampleOperation implements AggregationOperation {
 	 */
 	@Override
 	public Document toDocument(AggregationOperationContext context) {
-		return new Document("$sample", new Document("size", this.sampleSize));
+		return new Document(operator(), new Document("size", this.sampleSize));
+	}
+
+	@Override
+	public String operator() {
+		return "$sample";
 	}
 }

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

@@ -51,6 +51,11 @@ public class SkipOperation implements AggregationOperation {
 	 */
 	@Override
 	public Document toDocument(AggregationOperationContext context) {
-		return new Document("$skip", skipCount);
+		return new Document(operator(), skipCount);
+	}
+
+	@Override
+	public String operator() {
+		return "$skip";
 	}
 }

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

@@ -74,7 +74,12 @@ public class SortByCountOperation implements AggregationOperation {
 	@Override
 	public Document toDocument(AggregationOperationContext context) {
 
-		return new Document("$sortByCount", groupByExpression == null ? context.getReference(groupByField).toString()
+		return new Document(operator(), groupByExpression == null ? context.getReference(groupByField).toString()
 				: groupByExpression.toDocument(context));
 	}
+
+	@Override
+	public String operator() {
+		return "$sortByCount";
+	}
 }

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

@@ -74,6 +74,11 @@ public class SortOperation implements AggregationOperation {
 			object.put(reference.getRaw(), order.isAscending() ? 1 : -1);
 		}
 
-		return new Document("$sort", object);
+		return new Document(operator(), object);
+	}
+
+	@Override
+	public String operator() {
+		return "$sort";
 	}
 }

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

@@ -81,6 +81,6 @@ public class TypedAggregation<I> extends Aggregation {
 	public TypedAggregation<I> withOptions(AggregationOptions options) {
 
 		Assert.notNull(options, "AggregationOptions must not be null.");
-		return new TypedAggregation<I>(inputType, operations, options);
+		return new TypedAggregation<I>(inputType, pipeline.getOperations(), options);
 	}
 }

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

@@ -117,13 +117,18 @@ public class UnsetOperation implements InheritsFieldsAggregationOperation {
 	public Document toDocument(AggregationOperationContext context) {
 
 		if (fields.size() == 1) {
-			return new Document("$unset", computeFieldName(fields.iterator().next(), context));
+			return new Document(operator(), computeFieldName(fields.iterator().next(), context));
 		}
 
-		return new Document("$unset",
+		return new Document(operator(),
 				fields.stream().map(it -> computeFieldName(it, context)).collect(Collectors.toList()));
 	}
 
+	@Override
+	public String operator() {
+		return "$unset";
+	}
+
 	private Object computeFieldName(Object field, AggregationOperationContext context) {
 
 		if (field instanceof Field) {

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

@@ -94,7 +94,7 @@ public class UnwindOperation
 		String path = context.getReference(field).toString();
 
 		if (!preserveNullAndEmptyArrays && arrayIndex == null) {
-			return new Document("$unwind", path);
+			return new Document(operator(), path);
 		}
 
 		Document unwindArgs = new Document();
@@ -104,7 +104,12 @@ public class UnwindOperation
 		}
 		unwindArgs.put("preserveNullAndEmptyArrays", preserveNullAndEmptyArrays);
 
-		return new Document("$unwind", unwindArgs);
+		return new Document(operator(), unwindArgs);
+	}
+
+	@Override
+	public String operator() {
+		return "$unwind";
 	}
 
 	/*

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/AggregationTests.java

@@ -1625,12 +1625,12 @@ public class AggregationTests {
 		mongoTemplate.save(new Person("Leoniv", "Yakubov", 55, Person.Sex.MALE));
 	}
 
-	@Test // DATAMONGO-1418
+	@Test // DATAMONGO-1418, DATAMONGO-2536
 	public void outShouldOutBeTheLastOperation() {
 		assertThatIllegalArgumentException().isThrownBy(() -> newAggregation(match(new Criteria()), //
 				group("field1").count().as("totalCount"), //
 				out("collection1"), //
-				skip(100L)));
+				skip(100L)).toPipeline(DEFAULT_CONTEXT));
 	}
 
 	@Test // DATAMONGO-1325
@@ -1907,6 +1907,27 @@ public class AggregationTests {
 		assertThat(result.getMappedResults()).containsOnly(source);
 	}
 
+	@Test // DATAMONGO-2536
+	public void skipOutputDoesNotReadBackAggregationResults() {
+
+		createTagDocuments();
+
+		Aggregation agg = newAggregation( //
+				project("tags"), //
+				unwind("tags"), //
+				group("tags") //
+						.count().as("n"), //
+				project("n") //
+						.and("tag").previousOperation(), //
+				sort(DESC, "n") //
+		).withOptions(AggregationOptions.builder().skipOutput().build());
+
+		AggregationResults<TagCount> results = mongoTemplate.aggregate(agg, INPUT_COLLECTION, TagCount.class);
+
+		assertThat(results.getMappedResults()).isEmpty();
+		assertThat(results.getRawResults()).isEmpty();
+	}
+
 	private void createUsersWithReferencedPersons() {
 
 		mongoTemplate.dropCollection(User.class);

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/ReactiveAggregationTests.java

@@ -164,4 +164,22 @@ public class ReactiveAggregationTests {
 		reactiveMongoTemplate.aggregate(aggregation, "newyork", Document.class).as(StepVerifier::create).expectNextCount(4)
 				.verifyComplete();
 	}
+
+	@Test // DATAMONGO-2356
+	public void skipOutputDoesNotReadBackAggregationResults() {
+
+		Product product = new Product("P1", "A", 1.99, 3, 0.05, 0.19);
+		reactiveMongoTemplate.insert(product).as(StepVerifier::create).expectNextCount(1).verifyComplete();
+
+		double shippingCosts = 1.2;
+
+		TypedAggregation<Product> agg = newAggregation(Product.class, //
+				project("name", "netPrice") //
+						.andExpression("netPrice * 10", shippingCosts).as("salesPrice") //
+		).withOptions(AggregationOptions.builder().skipOutput().build());
+
+		reactiveMongoTemplate.aggregate(agg, Document.class).as(StepVerifier::create).verifyComplete();
+	}
+
+
 }