DATAMONGO-1552 - Add $bucket aggregation stage.

Original Pull Request: #426

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

@@ -202,7 +202,7 @@ public class Aggregation {
 	}
 
 	/**
-	 * Creates a new {@link ProjectionOperation} includeing the given {@link Fields}.
+	 * Creates a new {@link ProjectionOperation} including the given {@link Fields}.
 	 *
 	 * @param fields must not be {@literal null}.
 	 * @return
@@ -418,6 +418,26 @@ public class Aggregation {
 		return new OutOperation(outCollectionName);
 	}
 
+	/**
+	 * Creates a new {@link BucketOperation} using given {@literal groupByField}.
+	 *
+	 * @param groupByField must not be {@literal null} or empty.
+	 * @return
+	 */
+	public static BucketOperation bucket(String groupByField) {
+		return new BucketOperation(field(groupByField));
+	}
+
+	/**
+	 * Creates a new {@link BucketOperation} using given {@link AggregationExpression group-by expression}.
+	 *
+	 * @param groupByExpression must not be {@literal null}.
+	 * @return
+	 */
+	public static BucketOperation bucket(AggregationExpression groupByExpression) {
+		return new BucketOperation(groupByExpression);
+	}
+
 	/**
 	 * Creates a new {@link LookupOperation}.
 	 *

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

@@ -0,0 +1,226 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.mongodb.core.aggregation;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.springframework.data.mongodb.core.aggregation.BucketOperation.BucketOperationOutputBuilder;
+import org.springframework.util.Assert;
+
+import com.mongodb.BasicDBObject;
+import com.mongodb.DBObject;
+
+/**
+ * Encapsulates the aggregation framework {@code $bucket}-operation.
+ * <p>
+ * Bucket stage is typically used with {@link Aggregation} and {@code $facet}. Categorizes incoming documents into
+ * groups, called buckets, based on a specified expression and bucket boundaries.
+ * <p>
+ * We recommend to use the static factory method {@link Aggregation#bucket(String)} instead of creating instances of
+ * this class directly.
+ *
+ * @see http://docs.mongodb.org/manual/reference/aggregation/bucket/
+ * @see BucketOperationSupport
+ * @author Mark Paluch
+ * @since 1.10
+ */
+public class BucketOperation extends BucketOperationSupport<BucketOperation, BucketOperationOutputBuilder>
+		implements FieldsExposingAggregationOperation {
+
+	private final List<Object> boundaries;
+	private final Object defaultBucket;
+
+	/**
+	 * Creates a new {@link BucketOperation} given a {@link Field group-by field}.
+	 *
+	 * @param groupByField must not be {@literal null}.
+	 */
+	public BucketOperation(Field groupByField) {
+
+		super(groupByField);
+
+		this.boundaries = Collections.emptyList();
+		this.defaultBucket = null;
+	}
+
+	/**
+	 * Creates a new {@link BucketOperation} given a {@link AggregationExpression group-by expression}.
+	 *
+	 * @param groupByExpression must not be {@literal null}.
+	 */
+	public BucketOperation(AggregationExpression groupByExpression) {
+
+		super(groupByExpression);
+
+		this.boundaries = Collections.emptyList();
+		this.defaultBucket = null;
+	}
+
+	private BucketOperation(BucketOperation bucketOperation, Outputs outputs) {
+
+		super(bucketOperation, outputs);
+
+		this.boundaries = bucketOperation.boundaries;
+		this.defaultBucket = bucketOperation.defaultBucket;
+	}
+
+	private BucketOperation(BucketOperation bucketOperation, List<Object> boundaries, Object defaultBucket) {
+
+		super(bucketOperation);
+
+		this.boundaries = new ArrayList<Object>(boundaries);
+		this.defaultBucket = defaultBucket;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.BucketOperationSupport#toDBObject(org.springframework.data.mongodb.core.aggregation.AggregationOperationContext)
+	 */
+	@Override
+	public DBObject toDBObject(AggregationOperationContext context) {
+
+		DBObject options = new BasicDBObject();
+
+		options.put("boundaries", context.getMappedObject(new BasicDBObject("$set", boundaries)).get("$set"));
+
+		if (defaultBucket != null) {
+			options.put("default", context.getMappedObject(new BasicDBObject("$set", defaultBucket)).get("$set"));
+		}
+
+		options.putAll(super.toDBObject(context));
+
+		return new BasicDBObject("$bucket", options);
+	}
+
+	/**
+	 * Configures a default bucket {@literal literal} and return a new {@link BucketOperation}.
+	 * 
+	 * @param literal must not be {@literal null}.
+	 * @return
+	 */
+	public BucketOperation withDefaultBucket(Object literal) {
+
+		Assert.notNull(literal, "Default bucket literal must not be null!");
+		return new BucketOperation(this, boundaries, literal);
+	}
+
+	/**
+	 * Configures {@literal boundaries} and return a new {@link BucketOperation}. Existing {@literal boundaries} are
+	 * preserved and the new {@literal boundaries} are appended.
+	 * 
+	 * @param boundaries must not be {@literal null}.
+	 * @return
+	 */
+	public BucketOperation withBoundaries(Object... boundaries) {
+
+		Assert.notNull(boundaries, "Boundaries must not be null!");
+
+		List<Object> newBoundaries = new ArrayList<Object>(this.boundaries.size() + boundaries.length);
+		newBoundaries.addAll(this.boundaries);
+		newBoundaries.addAll(Arrays.asList(boundaries));
+
+		return new BucketOperation(this, newBoundaries, defaultBucket);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.BucketOperationSupport#newBucketOperation(org.springframework.data.mongodb.core.aggregation.BucketOperationSupport.Outputs)
+	 */
+	@Override
+	protected BucketOperation newBucketOperation(Outputs outputs) {
+		return new BucketOperation(this, outputs);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.BucketOperationSupport#andOutputExpression(java.lang.String, java.lang.Object[])
+	 */
+	@Override
+	public ExpressionBucketOperationBuilder andOutputExpression(String expression, Object... params) {
+		return new ExpressionBucketOperationBuilder(expression, this, params);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.BucketOperationSupport#andOutput(org.springframework.data.mongodb.core.aggregation.AggregationExpression)
+	 */
+	@Override
+	public BucketOperationOutputBuilder andOutput(AggregationExpression expression) {
+		return new BucketOperationOutputBuilder(expression, this);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.BucketOperationSupport#andOutput(java.lang.String)
+	 */
+	@Override
+	public BucketOperationOutputBuilder andOutput(String fieldName) {
+		return new BucketOperationOutputBuilder(Fields.field(fieldName), this);
+	}
+
+	/**
+	 * {@link OutputBuilder} implementation for {@link BucketOperation}.
+	 */
+	public static class BucketOperationOutputBuilder
+			extends BucketOperationSupport.OutputBuilder<BucketOperationOutputBuilder, BucketOperation> {
+
+		/**
+		 * Creates a new {@link BucketOperationOutputBuilder} fot the given value and {@link BucketOperation}.
+		 *
+		 * @param value must not be {@literal null}.
+		 * @param operation must not be {@literal null}.
+		 */
+		protected BucketOperationOutputBuilder(Object value, BucketOperation operation) {
+			super(value, operation);
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.aggregation.BucketOperationSupport.OutputBuilder#apply(org.springframework.data.mongodb.core.aggregation.BucketOperationSupport.OperationOutput)
+		 */
+		@Override
+		protected BucketOperationOutputBuilder apply(OperationOutput operationOutput) {
+			return new BucketOperationOutputBuilder(operationOutput, this.operation);
+		}
+	}
+
+	/**
+	 * {@link ExpressionBucketOperationBuilderSupport} implementation for {@link BucketOperation} using SpEL expression
+	 * based {@link Output}.
+	 * 
+	 * @author Mark Paluch
+	 */
+	public static class ExpressionBucketOperationBuilder
+			extends ExpressionBucketOperationBuilderSupport<BucketOperationOutputBuilder, BucketOperation> {
+
+		/**
+		 * Creates a new {@link ExpressionBucketOperationBuilderSupport} for the given value, {@link BucketOperation}
+		 * and parameters.
+		 *
+		 * @param expression must not be {@literal null}.
+		 * @param operation must not be {@literal null}.
+		 * @param parameters
+		 */
+		protected ExpressionBucketOperationBuilder(String expression, BucketOperation operation, Object[] parameters) {
+			super(expression, operation, parameters);
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.aggregation.BucketOperationSupport.OutputBuilder#apply(org.springframework.data.mongodb.core.aggregation.BucketOperationSupport.OperationOutput)
+		 */
+		@Override
+		protected BucketOperationOutputBuilder apply(OperationOutput operationOutput) {
+			return new BucketOperationOutputBuilder(operationOutput, this.operation);
+		}
+	}
+}

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

@@ -0,0 +1,697 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.mongodb.core.aggregation;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+import org.springframework.data.mongodb.core.aggregation.BucketOperationSupport.OutputBuilder;
+import org.springframework.data.mongodb.core.aggregation.ExposedFields.ExposedField;
+import org.springframework.data.mongodb.core.aggregation.ProjectionOperation.ProjectionOperationBuilder;
+import org.springframework.expression.spel.ast.Projection;
+import org.springframework.util.Assert;
+
+import com.mongodb.BasicDBObject;
+import com.mongodb.DBObject;
+
+/**
+ * Base class for bucket operations that support output expressions the aggregation framework.
+ * <p>
+ * Bucket stages collect documents into buckets and can contribute output fields.
+ * <p>
+ * Implementing classes are required to provide an {@link OutputBuilder}.
+ *
+ * @see http://docs.mongodb.org/manual/reference/aggregation/bucket/
+ * @author Mark Paluch
+ * @since 1.10
+ */
+public abstract class BucketOperationSupport<T extends BucketOperationSupport<T, B>, B extends OutputBuilder<B, T>>
+		implements FieldsExposingAggregationOperation {
+
+	private final Field groupByField;
+	private final AggregationExpression groupByExpression;
+	private final Outputs outputs;
+
+	/**
+	 * Creates a new {@link BucketOperationSupport} given a {@link Field group-by field}.
+	 * 
+	 * @param groupByField must not be {@literal null}.
+	 */
+	protected BucketOperationSupport(Field groupByField) {
+
+		Assert.notNull(groupByField, "Group by field must not be null!");
+
+		this.groupByField = groupByField;
+		this.groupByExpression = null;
+		this.outputs = Outputs.EMPTY;
+
+	}
+
+	/**
+	 * Creates a new {@link BucketOperationSupport} given a {@link AggregationExpression group-by expression}.
+	 *
+	 * @param groupByExpression must not be {@literal null}.
+	 */
+	protected BucketOperationSupport(AggregationExpression groupByExpression) {
+
+		Assert.notNull(groupByExpression, "Group by AggregationExpression must not be null!");
+
+		this.groupByExpression = groupByExpression;
+		this.groupByField = null;
+		this.outputs = Outputs.EMPTY;
+	}
+
+	/**
+	 * Creates a copy of {@link BucketOperationSupport}.
+	 * 
+	 * @param operationSupport must not be {@literal null}.
+	 */
+	protected BucketOperationSupport(BucketOperationSupport<?, ?> operationSupport) {
+		this(operationSupport, operationSupport.outputs);
+	}
+
+	/**
+	 * Creates a copy of {@link BucketOperationSupport} and applies the new {@link Outputs}.
+	 * 
+	 * @param operationSupport must not be {@literal null}.
+	 * @param outputs must not be {@literal null}.
+	 */
+	protected BucketOperationSupport(BucketOperationSupport<?, ?> operationSupport, Outputs outputs) {
+
+		Assert.notNull(operationSupport, "BucketOperationSupport must not be null!");
+		Assert.notNull(outputs, "Outputs must not be null!");
+
+		this.groupByField = operationSupport.groupByField;
+		this.groupByExpression = operationSupport.groupByExpression;
+		this.outputs = outputs;
+	}
+
+	/**
+	 * Creates a new {@link ExpressionBucketOperationBuilderSupport} given a SpEL {@literal expression} and optional
+	 * {@literal params} to add an output field to the resulting bucket documents.
+	 * 
+	 * @param expression the SpEL expression, must not be {@literal null} or empty.
+	 * @param params must not be {@literal null}
+	 * @return
+	 */
+	public abstract ExpressionBucketOperationBuilderSupport<B, T> andOutputExpression(String expression,
+			Object... params);
+
+	/**
+	 * Creates a new {@link BucketOperationSupport} given an {@link AggregationExpression} to add an output field to the
+	 * resulting bucket documents.
+	 *
+	 * @param expression the SpEL expression, must not be {@literal null} or empty.
+	 * @return
+	 */
+	public abstract B andOutput(AggregationExpression expression);
+
+	/**
+	 * Creates a new {@link BucketOperationSupport} given {@literal fieldName} to add an output field to the resulting
+	 * bucket documents. {@link BucketOperationSupport} exposes accumulation operations that can be applied to
+	 * {@literal fieldName}.
+	 *
+	 * @param fieldName must not be {@literal null} or empty.
+	 * @return
+	 */
+	public abstract B andOutput(String fieldName);
+
+	/**
+	 * Creates a new {@link BucketOperationSupport} given to add a count field to the resulting bucket documents.
+	 *
+	 * @return
+	 */
+	public B andOutputCount() {
+		return andOutput(new AggregationExpression() {
+			@Override
+			public DBObject toDbObject(AggregationOperationContext context) {
+				return new BasicDBObject("$sum", 1);
+			}
+		});
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#toDBObject(org.springframework.data.mongodb.core.aggregation.AggregationOperationContext)
+	 */
+	@Override
+	public DBObject toDBObject(AggregationOperationContext context) {
+
+		DBObject dbObject = new BasicDBObject();
+
+		dbObject.put("groupBy", groupByExpression == null ? context.getReference(groupByField).toString()
+				: groupByExpression.toDbObject(context));
+
+		if (!outputs.isEmpty()) {
+			dbObject.put("output", outputs.toDbObject(context));
+		}
+
+		return dbObject;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.FieldsExposingAggregationOperation#getFields()
+	 */
+	@Override
+	public ExposedFields getFields() {
+		return outputs.asExposedFields();
+	}
+
+	/**
+	 * Implementation hook to create a new bucket operation.
+	 *
+	 * @param outputs the outputs
+	 * @return the new bucket operation.
+	 */
+	protected abstract T newBucketOperation(Outputs outputs);
+
+	protected T andOutput(Output output) {
+		return newBucketOperation(outputs.and(output));
+	}
+
+	/**
+	 * Builder for SpEL expression-based {@link Output}.
+	 * 
+	 * @author Mark Paluch
+	 */
+	public abstract static class ExpressionBucketOperationBuilderSupport<B extends OutputBuilder<B, T>, T extends BucketOperationSupport<T, B>>
+			extends OutputBuilder<B, T> {
+
+		/**
+		 * Creates a new {@link ExpressionBucketOperationBuilderSupport} for the given value, {@link BucketOperationSupport}
+		 * and parameters.
+		 *
+		 * @param expression must not be {@literal null}.
+		 * @param operation must not be {@literal null}.
+		 * @param parameters
+		 */
+		protected ExpressionBucketOperationBuilderSupport(String expression, T operation, Object[] parameters) {
+			super(new SpelExpressionOutput(expression, parameters), operation);
+		}
+	}
+
+	/**
+	 * Base class for {@link Output} builders that result in a {@link BucketOperationSupport} providing the built
+	 * {@link Output}.
+	 *
+	 * @author Mark Paluch
+	 */
+	public abstract static class OutputBuilder<B extends OutputBuilder<B, T>, T extends BucketOperationSupport<T, B>> {
+
+		protected final Object value;
+		protected final T operation;
+
+		/**
+		 * Creates a new {@link OutputBuilder} for the given value and {@link BucketOperationSupport}.
+		 *
+		 * @param value must not be {@literal null}.
+		 * @param operation must not be {@literal null}.
+		 */
+		public OutputBuilder(Object value, T operation) {
+
+			Assert.notNull(value, "Value must not be null or empty!");
+			Assert.notNull(operation, "ProjectionOperation must not be null!");
+
+			this.value = value;
+			this.operation = operation;
+		}
+
+		/**
+		 * Generates a builder for a {@code $sum}-expression.
+		 * <p>
+		 * Count expressions are emulated via {@code $sum: 1}.
+		 * <p>
+		 * 
+		 * @return
+		 */
+		public B count() {
+			return sum(1);
+		}
+
+		/**
+		 * Generates a builder for a {@code $sum}-expression for the current value.
+		 * 
+		 * @return
+		 */
+		public B sum() {
+			return apply(Accumulators.SUM);
+		}
+
+		/**
+		 * Generates a builder for a {@code $sum}-expression for the given {@literal value}.
+		 *
+		 * @param value
+		 * @return
+		 */
+		public B sum(Number value) {
+			return apply(new OperationOutput(Accumulators.SUM.toString(), Collections.singleton(value)));
+		}
+
+		/**
+		 * Generates a builder for an {@code $last}-expression for the current value..
+		 * 
+		 * @return
+		 */
+		public B last() {
+			return apply(Accumulators.LAST);
+		}
+
+		/**
+		 * Generates a builder for a {@code $first}-expression the current value.
+		 * 
+		 * @return
+		 */
+		public B first() {
+			return apply(Accumulators.FIRST);
+		}
+
+		/**
+		 * Generates a builder for an {@code $avg}-expression for the current value.
+		 * 
+		 * @param reference
+		 * @return
+		 */
+		public B avg() {
+			return apply(Accumulators.AVG);
+		}
+
+		/**
+		 * Generates a builder for an {@code $min}-expression for the current value.
+		 * 
+		 * @return
+		 */
+		public B min() {
+			return apply(Accumulators.MIN);
+		}
+
+		/**
+		 * Generates a builder for an {@code $max}-expression for the current value.
+		 * 
+		 * @return
+		 */
+		public B max() {
+			return apply(Accumulators.MAX);
+		}
+
+		/**
+		 * Generates a builder for an {@code $push}-expression for the current value.
+		 *
+		 * @return
+		 */
+		public B push() {
+			return apply(Accumulators.PUSH);
+		}
+
+		/**
+		 * Generates a builder for an {@code $addToSet}-expression for the current value.
+		 *
+		 * @return
+		 */
+		public B addToSet() {
+			return apply(Accumulators.ADDTOSET);
+		}
+
+		/**
+		 * Apply an operator to the current value.
+		 *
+		 * @param operation the operation name, must not be {@literal null} or empty.
+		 * @param values must not be {@literal null}.
+		 * @return
+		 */
+		public B apply(String operation, Object... values) {
+
+			Assert.hasText(operation, "Operation must not be empty or null!");
+			Assert.notNull(value, "Values must not be null!");
+
+			List<Object> objects = new ArrayList<Object>(values.length + 1);
+			objects.add(value);
+			objects.addAll(Arrays.asList(values));
+			return apply(new OperationOutput(operation, objects));
+		}
+
+		/**
+		 * Apply an {@link OperationOutput} to this output.
+		 * 
+		 * @param operationOutput must not be {@literal null}.
+		 * @return
+		 */
+		protected abstract B apply(OperationOutput operationOutput);
+
+		private B apply(Accumulators operation) {
+			return this.apply(operation.toString());
+		}
+
+		/**
+		 * Returns the finally to be applied {@link BucketOperation} with the given alias.
+		 *
+		 * @param alias will never be {@literal null} or empty.
+		 * @return
+		 */
+		public T as(String alias) {
+
+			if (value instanceof OperationOutput) {
+				return this.operation.andOutput(((OperationOutput) this.value).withAlias(alias));
+			}
+
+			if (value instanceof Field) {
+				throw new IllegalStateException("Cannot add a field as top-level output. Use accumulator expressions.");
+			}
+
+			return this.operation
+					.andOutput(new AggregationExpressionOutput(Fields.field(alias), (AggregationExpression) value));
+		}
+	}
+
+	private enum Accumulators {
+
+		SUM("$sum"), AVG("$avg"), FIRST("$first"), LAST("$last"), MAX("$max"), MIN("$min"), PUSH("$push"), ADDTOSET(
+				"$addToSet");
+
+		private String mongoOperator;
+
+		Accumulators(String mongoOperator) {
+			this.mongoOperator = mongoOperator;
+		}
+
+		/* (non-Javadoc)
+		 * @see java.lang.Enum#toString()
+		 */
+		@Override
+		public String toString() {
+			return mongoOperator;
+		}
+	}
+
+	/**
+	 * Encapsulates {@link Output}s.
+	 * 
+	 * @author Mark Paluch
+	 */
+	protected static class Outputs implements AggregationExpression {
+
+		protected static final Outputs EMPTY = new Outputs();
+
+		private List<Output> outputs;
+
+		/**
+		 * Creates a new, empty {@link Outputs}.
+		 */
+		private Outputs() {
+			this.outputs = new ArrayList<Output>();
+		}
+
+		/**
+		 * Creates new {@link Outputs} containing all given {@link Output}s.
+		 * 
+		 * @param current
+		 * @param output
+		 */
+		private Outputs(Collection<Output> current, Output output) {
+
+			this.outputs = new ArrayList<Output>(current.size() + 1);
+			this.outputs.addAll(current);
+			this.outputs.add(output);
+		}
+
+		/**
+		 * @return the {@link ExposedFields} derived from {@link Output}.
+		 */
+		protected ExposedFields asExposedFields() {
+
+			ExposedFields fields = ExposedFields.from();
+
+			for (Output output : outputs) {
+				fields = fields.and(output.getExposedField());
+			}
+
+			return fields;
+		}
+
+		/**
+		 * Create a new {@link Outputs} that contains the new {@link Output}.
+		 * 
+		 * @param output must not be {@literal null}.
+		 * @return the new {@link Outputs} that contains the new {@link Output}
+		 */
+		protected Outputs and(Output output) {
+
+			Assert.notNull(output, "BucketOutput must not be null!");
+			return new Outputs(this.outputs, output);
+		}
+
+		/**
+		 * @return {@literal true} if {@link Outputs} contains no {@link Output}.
+		 */
+		protected boolean isEmpty() {
+			return outputs.isEmpty();
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.aggregation.AggregationExpression#toDbObject(org.springframework.data.mongodb.core.aggregation.AggregationOperationContext)
+		 */
+		@Override
+		public DBObject toDbObject(AggregationOperationContext context) {
+
+			DBObject dbObject = new BasicDBObject();
+
+			for (Output output : outputs) {
+				dbObject.put(output.getExposedField().getName(), output.toDbObject(context));
+			}
+
+			return dbObject;
+		}
+
+	}
+
+	/**
+	 * Encapsulates an output field in a bucket aggregation stage.
+	 * <p>
+	 * Output fields can be either top-level fields that define a valid field name or nested output fields using
+	 * operators.
+	 * 
+	 * @author Mark Paluch
+	 */
+	protected abstract static class Output implements AggregationExpression {
+
+		private final ExposedField field;
+
+		/**
+		 * Creates new {@link Projection} for the given {@link Field}.
+		 *
+		 * @param field must not be {@literal null}.
+		 */
+		protected Output(Field field) {
+
+			Assert.notNull(field, "Field must not be null!");
+			this.field = new ExposedField(field, true);
+		}
+
+		/**
+		 * Returns the field exposed by the {@link Output}.
+		 *
+		 * @return will never be {@literal null}.
+		 */
+		protected ExposedField getExposedField() {
+			return field;
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.aggregation.AggregationExpression#toDbObject(org.springframework.data.mongodb.core.aggregation.AggregationOperationContext)
+		 */
+		@Override
+		public abstract DBObject toDbObject(AggregationOperationContext context);
+	}
+
+	/**
+	 * Output field that uses a Mongo operation (expression object) to generate an output field value.
+	 * <p>
+	 * {@link OperationOutput} is used either with a regular field name or an operation keyword (e.g.
+	 * {@literal $sum, $count}).
+	 *
+	 * @author Mark Paluch
+	 */
+	protected static class OperationOutput extends Output {
+
+		private final String operation;
+		private final List<Object> values;
+
+		/**
+		 * Creates a new {@link Output} for the given field.
+		 *
+		 * @param operation the actual operation key, must not be {@literal null} or empty.
+		 * @param values the values to pass into the operation, must not be {@literal null}.
+		 */
+		public OperationOutput(String operation, Collection<? extends Object> values) {
+
+			super(Fields.field(operation));
+
+			Assert.hasText(operation, "Operation must not be null or empty!");
+			Assert.notNull(values, "Values must not be null!");
+
+			this.operation = operation;
+			this.values = new ArrayList<Object>(values);
+		}
+
+		private OperationOutput(Field field, OperationOutput operationOutput) {
+
+			super(field);
+
+			this.operation = operationOutput.operation;
+			this.values = operationOutput.values;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.aggregation.ProjectionOperation.Projection#toDBObject(org.springframework.data.mongodb.core.aggregation.AggregationOperationContext)
+		 */
+		@Override
+		public DBObject toDbObject(AggregationOperationContext context) {
+
+			List<Object> operationArguments = getOperationArguments(context);
+			return new BasicDBObject(operation,
+					operationArguments.size() == 1 ? operationArguments.get(0) : operationArguments);
+		}
+
+		protected List<Object> getOperationArguments(AggregationOperationContext context) {
+
+			List<Object> result = new ArrayList<Object>(values != null ? values.size() : 1);
+
+			for (Object element : values) {
+
+				if (element instanceof Field) {
+					result.add(context.getReference((Field) element).toString());
+				} else if (element instanceof Fields) {
+					for (Field field : (Fields) element) {
+						result.add(context.getReference(field).toString());
+					}
+				} else if (element instanceof AggregationExpression) {
+					result.add(((AggregationExpression) element).toDbObject(context));
+				} else {
+					result.add(element);
+				}
+			}
+
+			return result;
+		}
+
+		/**
+		 * Returns the field that holds the {@link ProjectionOperationBuilder.OperationProjection}.
+		 *
+		 * @return
+		 */
+		protected Field getField() {
+			return getExposedField();
+		}
+
+		/**
+		 * Creates a new instance of this {@link OperationOutput} with the given alias.
+		 *
+		 * @param alias the alias to set
+		 * @return
+		 */
+		public OperationOutput withAlias(String alias) {
+
+			final Field aliasedField = Fields.field(alias);
+			return new OperationOutput(aliasedField, this) {
+
+				@Override
+				protected Field getField() {
+					return aliasedField;
+				}
+
+				@Override
+				protected List<Object> getOperationArguments(AggregationOperationContext context) {
+
+					// We have to make sure that we use the arguments from the "previous" OperationOutput that we replace
+					// with this new instance.
+
+					return OperationOutput.this.getOperationArguments(context);
+				}
+			};
+		}
+	}
+
+	/**
+	 * A {@link Output} based on a SpEL expression.
+	 */
+	static class SpelExpressionOutput extends Output {
+
+		private static final SpelExpressionTransformer TRANSFORMER = new SpelExpressionTransformer();
+
+		private final String expression;
+		private final Object[] params;
+
+		/**
+		 * Creates a new {@link SpelExpressionOutput} for the given field, SpEL expression and parameters.
+		 *
+		 * @param expression must not be {@literal null} or empty.
+		 * @param parameters must not be {@literal null}.
+		 */
+		public SpelExpressionOutput(String expression, Object[] parameters) {
+
+			super(Fields.field(expression));
+
+			Assert.hasText(expression, "Expression must not be null!");
+			Assert.notNull(parameters, "Parameters must not be null!");
+
+			this.expression = expression;
+			this.params = parameters.clone();
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.aggregation.BucketOperationSupport.Output#toDbObject(org.springframework.data.mongodb.core.aggregation.AggregationOperationContext)
+		 */
+		@Override
+		public DBObject toDbObject(AggregationOperationContext context) {
+			return (DBObject) toMongoExpression(context, expression, params);
+		}
+
+		protected static Object toMongoExpression(AggregationOperationContext context, String expression, Object[] params) {
+			return TRANSFORMER.transform(expression, context, params);
+		}
+	}
+
+	/**
+	 * @author Mark Paluch
+	 */
+	private static class AggregationExpressionOutput extends Output {
+
+		private final AggregationExpression expression;
+
+		/**
+		 * Creates a new {@link AggregationExpressionOutput}.
+		 *
+		 * @param field
+		 * @param expression
+		 */
+		protected AggregationExpressionOutput(Field field, AggregationExpression expression) {
+
+			super(field);
+
+			this.expression = expression;
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.aggregation.BucketOperationSupport.Output#toDbObject(org.springframework.data.mongodb.core.aggregation.AggregationOperationContext)
+		 */
+		@Override
+		public DBObject toDbObject(AggregationOperationContext context) {
+			return expression.toDbObject(context);
+		}
+	}
+}

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

@@ -148,6 +148,7 @@ public class AggregationTests {
 		mongoTemplate.dropCollection(Sales.class);
 		mongoTemplate.dropCollection(Sales2.class);
 		mongoTemplate.dropCollection(Employee.class);
+		mongoTemplate.dropCollection(Art.class);
 	}
 
 	/**
@@ -1667,6 +1668,46 @@ public class AggregationTests {
 		assertThat((DBObject) list.get(1), isBsonObject().containing("name", "Eliot").containing("depth", 0L));
 	}
 
+	/**
+	 * @see DATAMONGO-1552
+	 */
+	@Test
+	public void bucketShouldCollectDocumentsIntoABucket() {
+
+		assumeTrue(mongoVersion.isGreaterThanOrEqualTo(THREE_DOT_FOUR));
+
+		Art a1 = Art.builder().id(1).title("The Pillars of Society").artist("Grosz").year(1926).price(199.99).build();
+		Art a2 = Art.builder().id(2).title("Melancholy III").artist("Munch").year(1902).price(280.00).build();
+		Art a3 = Art.builder().id(3).title("Dancer").artist("Miro").year(1925).price(76.04).build();
+		Art a4 = Art.builder().id(4).title("The Great Wave off Kanagawa").artist("Hokusai").price(167.30).build();
+
+		mongoTemplate.insert(Arrays.asList(a1, a2, a3, a4), Art.class);
+
+		TypedAggregation<Art> aggregation = newAggregation(Art.class, //
+				bucket("price") //
+						.withBoundaries(0, 100, 200) //
+						.withDefaultBucket("other") //
+						.andOutputCount().as("count") //
+						.andOutput("title").push().as("titles") //
+						.andOutputExpression("price * 10").sum().as("sum"));
+
+		AggregationResults<DBObject> result = mongoTemplate.aggregate(aggregation, DBObject.class);
+		assertThat(result.getMappedResults().size(), is(3));
+
+		// { "_id" : 0 , "count" : 1 , "titles" : [ "Dancer"] , "sum" : 760.4000000000001}
+		DBObject bound0 = result.getMappedResults().get(0);
+		assertThat(bound0, isBsonObject().containing("count", 1).containing("titles.[0]", "Dancer"));
+		assertThat((Double) bound0.get("sum"), is(closeTo(760.40, 0.1)));
+
+		// { "_id" : 100 , "count" : 2 , "titles" : [ "The Pillars of Society" , "The Great Wave off Kanagawa"] , "sum" :
+		// 3672.9}
+		DBObject bound100 = result.getMappedResults().get(1);
+		assertThat(bound100, isBsonObject().containing("count", 2).containing("_id", 100));
+		assertThat((List<String>) bound100.get("titles"),
+				hasItems("The Pillars of Society", "The Great Wave off Kanagawa"));
+		assertThat((Double) bound100.get("sum"), is(closeTo(3672.9, 0.1)));
+	}
+
 	private void createUsersWithReferencedPersons() {
 
 		mongoTemplate.dropCollection(User.class);
@@ -1956,4 +1997,18 @@ public class AggregationTests {
 		String name;
 		String reportsTo;
 	}
+
+	/**
+	 * @see DATAMONGO-1552
+	 */
+	@lombok.Data
+	@Builder
+	static class Art {
+
+		int id;
+		String title;
+		String artist;
+		Integer year;
+		double price;
+	}
 }

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

@@ -0,0 +1,254 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.mongodb.core.aggregation;
+
+import static org.hamcrest.core.Is.*;
+import static org.junit.Assert.*;
+import static org.springframework.data.mongodb.core.aggregation.Aggregation.*;
+
+import org.junit.Test;
+import org.springframework.data.mongodb.core.aggregation.AggregationExpressions.ArithmeticOperators;
+
+import com.mongodb.DBObject;
+import com.mongodb.util.JSON;
+
+/**
+ * Unit tests for {@link BucketOperation}.
+ * 
+ * @author Mark Paluch
+ */
+public class BucketOperationUnitTests {
+
+	/**
+	 * @see DATAMONGO-1552
+	 */
+	@Test(expected = IllegalArgumentException.class)
+	public void rejectsNullFields() {
+		new BucketOperation((Field) null);
+	}
+
+	/**
+	 * @see DATAMONGO-1552
+	 */
+	@Test
+	public void shouldRenderBucketOutputExpressions() {
+
+		BucketOperation operation = Aggregation.bucket("field") //
+				.andOutputExpression("(netPrice + surCharge) * taxrate * [0]", 2).as("grossSalesPrice") //
+				.andOutput("title").push().as("titles");
+
+		DBObject dbObject = operation.toDBObject(Aggregation.DEFAULT_CONTEXT);
+		assertThat(extractOutput(dbObject), is(JSON.parse(
+				"{ \"grossSalesPrice\" : { \"$multiply\" : [ { \"$add\" : [ \"$netPrice\" , \"$surCharge\"]} , \"$taxrate\" , 2]} , \"titles\" : { $push: \"$title\" } }}")));
+	}
+
+	/**
+	 * @see DATAMONGO-1552
+	 */
+	@Test(expected = IllegalStateException.class)
+	public void shouldRenderEmptyAggregationExpression() {
+		bucket("groupby").andOutput("field").as("alias");
+	}
+
+	/**
+	 * @see DATAMONGO-1552
+	 */
+	@Test
+	public void shouldRenderBucketOutputOperators() {
+
+		BucketOperation operation = Aggregation.bucket("field") //
+				.andOutputCount().as("titles");
+
+		DBObject dbObject = operation.toDBObject(Aggregation.DEFAULT_CONTEXT);
+		assertThat(extractOutput(dbObject), is(JSON.parse("{ titles : { $sum: 1 } }")));
+	}
+
+	/**
+	 * @see DATAMONGO-1552
+	 */
+	@Test
+	public void shouldRenderSumAggregationExpression() {
+
+		DBObject agg = bucket("field") //
+				.andOutput(ArithmeticOperators.valueOf("quizzes").sum()).as("quizTotal") //
+				.toDBObject(Aggregation.DEFAULT_CONTEXT);
+
+		assertThat(agg, is(JSON.parse(
+				"{ $bucket: { groupBy: \"$field\", boundaries: [],  output : { quizTotal: { $sum: \"$quizzes\"} } } }")));
+	}
+
+	/**
+	 * @see DATAMONGO-1552
+	 */
+	@Test
+	public void shouldRenderDefault() {
+
+		DBObject agg = bucket("field").withDefaultBucket("default bucket").toDBObject(Aggregation.DEFAULT_CONTEXT);
+
+		assertThat(agg,
+				is(JSON.parse("{ $bucket: { groupBy: \"$field\", boundaries: [],  default: \"default bucket\" } }")));
+	}
+
+	/**
+	 * @see DATAMONGO-1552
+	 */
+	@Test
+	public void shouldRenderBoundaries() {
+
+		DBObject agg = bucket("field") //
+				.withDefaultBucket("default bucket") //
+				.withBoundaries(0) //
+				.withBoundaries(10, 20).toDBObject(Aggregation.DEFAULT_CONTEXT);
+
+		assertThat(agg,
+				is(JSON.parse("{ $bucket: { boundaries: [0, 10, 20],  default: \"default bucket\", groupBy: \"$field\" } }")));
+	}
+
+	/**
+	 * @see DATAMONGO-1552
+	 */
+	@Test
+	public void shouldRenderSumOperator() {
+
+		BucketOperation operation = bucket("field") //
+				.andOutput("score").sum().as("cummulated_score");
+
+		DBObject dbObject = operation.toDBObject(Aggregation.DEFAULT_CONTEXT);
+		assertThat(extractOutput(dbObject), is(JSON.parse("{ cummulated_score : { $sum: \"$score\" } }")));
+	}
+
+	/**
+	 * @see DATAMONGO-1552
+	 */
+	@Test
+	public void shouldRenderSumWithValueOperator() {
+
+		BucketOperation operation = bucket("field") //
+				.andOutput("score").sum(4).as("cummulated_score");
+
+		DBObject dbObject = operation.toDBObject(Aggregation.DEFAULT_CONTEXT);
+		assertThat(extractOutput(dbObject), is(JSON.parse("{ cummulated_score : { $sum: 4 } }")));
+	}
+
+	/**
+	 * @see DATAMONGO-1552
+	 */
+	@Test
+	public void shouldRenderAvgOperator() {
+
+		BucketOperation operation = bucket("field") //
+				.andOutput("score").avg().as("average");
+
+		DBObject dbObject = operation.toDBObject(Aggregation.DEFAULT_CONTEXT);
+		assertThat(extractOutput(dbObject), is(JSON.parse("{ average : { $avg: \"$score\" } }")));
+	}
+
+	/**
+	 * @see DATAMONGO-1552
+	 */
+	@Test
+	public void shouldRenderFirstOperator() {
+
+		BucketOperation operation = bucket("field") //
+				.andOutput("title").first().as("first_title");
+
+		DBObject dbObject = operation.toDBObject(Aggregation.DEFAULT_CONTEXT);
+		assertThat(extractOutput(dbObject), is(JSON.parse("{ first_title : { $first: \"$title\" } }")));
+	}
+
+	/**
+	 * @see DATAMONGO-1552
+	 */
+	@Test
+	public void shouldRenderLastOperator() {
+
+		BucketOperation operation = bucket("field") //
+				.andOutput("title").last().as("last_title");
+
+		DBObject dbObject = operation.toDBObject(Aggregation.DEFAULT_CONTEXT);
+		assertThat(extractOutput(dbObject), is(JSON.parse("{ last_title : { $last: \"$title\" } }")));
+	}
+
+	/**
+	 * @see DATAMONGO-1552
+	 */
+	@Test
+	public void shouldRenderMinOperator() {
+
+		BucketOperation operation = bucket("field") //
+				.andOutput("score").min().as("min_score");
+
+		DBObject dbObject = operation.toDBObject(Aggregation.DEFAULT_CONTEXT);
+		assertThat(extractOutput(dbObject), is(JSON.parse("{ min_score : { $min: \"$score\" } }")));
+	}
+
+	/**
+	 * @see DATAMONGO-1552
+	 */
+	@Test
+	public void shouldRenderPushOperator() {
+
+		BucketOperation operation = bucket("field") //
+				.andOutput("title").push().as("titles");
+
+		DBObject dbObject = operation.toDBObject(Aggregation.DEFAULT_CONTEXT);
+		assertThat(extractOutput(dbObject), is(JSON.parse("{ titles : { $push: \"$title\" } }")));
+	}
+
+	/**
+	 * @see DATAMONGO-1552
+	 */
+	@Test
+	public void shouldRenderAddToSetOperator() {
+
+		BucketOperation operation = bucket("field") //
+				.andOutput("title").addToSet().as("titles");
+
+		DBObject dbObject = operation.toDBObject(Aggregation.DEFAULT_CONTEXT);
+		assertThat(extractOutput(dbObject), is(JSON.parse("{ titles : { $addToSet: \"$title\" } }")));
+	}
+
+	/**
+	 * @see DATAMONGO-1552
+	 */
+	@Test
+	public void shouldRenderSumWithExpression() {
+
+		BucketOperation operation = bucket("field") //
+				.andOutputExpression("netPrice + tax").sum().as("total");
+
+		DBObject dbObject = operation.toDBObject(Aggregation.DEFAULT_CONTEXT);
+		assertThat(extractOutput(dbObject), is(JSON.parse("{ total : { $sum: { $add : [\"$netPrice\", \"$tax\"]} } }")));
+	}
+
+	/**
+	 * @see DATAMONGO-1552
+	 */
+	@Test
+	public void shouldRenderSumWithOwnOutputExpression() {
+
+		BucketOperation operation = bucket("field") //
+				.andOutputExpression("netPrice + tax").apply("$multiply", 5).as("total");
+
+		DBObject dbObject = operation.toDBObject(Aggregation.DEFAULT_CONTEXT);
+		assertThat(extractOutput(dbObject),
+				is(JSON.parse("{ total : { $multiply: [ {$add : [\"$netPrice\", \"$tax\"]}, 5] } }")));
+	}
+
+	private static DBObject extractOutput(DBObject fromBucketClause) {
+		return (DBObject) ((DBObject) fromBucketClause.get("$bucket")).get("output");
+	}
+}