DATAMONGO-586 - Added test case for complex aggregation framework use case.

Added Zipcode sample dataset from 10gen. Allow Projections to be used in conjunction with GroupOperations. Integration & Refactoring of github contribution by Tobias Trelle and Sebastian Herold. Switched from builder-style to static factory based DSL construction of aggregation specifications. Introduced embedded DSL for convenient construction of aggregation specifications. Added test cases based on mongodb aggregation framework examples. Added more test cases, additional java doc. Added test case for unwind operation (returnFiveMostCommonLikes) in AggregationTests. Other test cases should now also run in CI environment, due to deterministic result ordering. Adjusted write concern to ensure persistence of sample data.

Introduced TypedAggregation which holds type information of the input type of an aggregation. Cleaned up aggregate methods on MongoOperations. Removed HasToDBObject interface. Cleaned up constructors for Aggregation and TypedAggregation.

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

@@ -19,9 +19,9 @@ import java.util.Collection;
 import java.util.List;
 import java.util.Set;
 
-import org.springframework.data.mongodb.core.aggregation.AggregationOperation;
-import org.springframework.data.mongodb.core.aggregation.AggregationPipeline;
+import org.springframework.data.mongodb.core.aggregation.Aggregation;
 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.geo.GeoResult;
 import org.springframework.data.mongodb.core.geo.GeoResults;
@@ -305,30 +305,57 @@ public interface MongoOperations {
 	 */
 	<T> GroupByResults<T> group(Criteria criteria, String inputCollectionName, GroupBy groupBy, Class<T> entityClass);
 
+	/**
+	 * Execute an aggregation operation. The raw results will be mapped to the given entity class. The name of the
+	 * inputCollection is derived from the inputType of the aggregation.
+	 * 
+	 * @param aggregation The {@link TypedAggregation} specification holding the aggregation operations, must not be
+	 *          {@literal null}.
+	 * @param inputCollectionName The name of the input collection to use for the aggreation.
+	 * @param outputType The parameterized type of the returned list, must not be {@literal null}.
+	 * @return The results of the aggregation operation.
+	 * @since 1.3
+	 */
+	<I, O> AggregationResults<O> aggregate(TypedAggregation<I, O> aggregation, String inputCollectionName,
+			Class<O> outputType);
+
+	/**
+	 * Execute an aggregation operation. The raw results will be mapped to the given entity class. The name of the
+	 * inputCollection is derived from the inputType of the aggregation.
+	 * 
+	 * @param aggregation The {@link TypedAggregation} specification holding the aggregation operations, must not be
+	 *          {@literal null}.
+	 * @param outputType The parameterized type of the returned list, must not be {@literal null}.
+	 * @return The results of the aggregation operation.
+	 * @since 1.3
+	 */
+	<I, O> AggregationResults<O> aggregate(TypedAggregation<I, O> aggregation, Class<O> outputType);
+
 	/**
 	 * Execute an aggregation operation. The raw results will be mapped to the given entity class.
 	 * 
-	 * @param inputCollectionName the collection there the aggregation operation will read from, must not be
-	 *          {@literal null} or empty.
-	 * @param pipeline The pipeline holding the aggregation operations, must not be {@literal null}.
-	 * @param entityClass The parameterized type of the returned list, must not be {@literal null}.
+	 * @param inputType the inputType where the aggregation operation will read from, must not be {@literal null} or
+	 *          empty.
+	 * @param aggregation The {@link Aggregation} specification holding the aggregation operations, must not be
+	 *          {@literal null}.
+	 * @param outputType The parameterized type of the returned list, must not be {@literal null}.
 	 * @return The results of the aggregation operation.
 	 * @since 1.3
 	 */
-	<T> AggregationResults<T> aggregate(String inputCollectionName, AggregationPipeline pipeline, Class<T> entityClass);
+	<I, O> AggregationResults<O> aggregate(Class<I> inputType, Aggregation<I, O> aggregation, Class<O> outputType);
 
 	/**
 	 * Execute an aggregation operation. The raw results will be mapped to the given entity class.
 	 * 
-	 * @param inputCollectionName the collection there the aggregation operation will read from, must not be
+	 * @param inputCollectionName the collection where the aggregation operation will read from, must not be
 	 *          {@literal null} or empty.
-	 * @param entityClass The parameterized type of the returned list, must not be {@literal null}.
-	 * @param operations The aggregation operations, must not be {@literal null}.
+	 * @param aggregation The {@link Aggregation} specification holding the aggregation operations, must not be
+	 *          {@literal null}.
+	 * @param outputType The parameterized type of the returned list, must not be {@literal null}.
 	 * @return The results of the aggregation operation.
 	 * @since 1.3
 	 */
-	<T> AggregationResults<T> aggregate(String inputCollectionName, Class<T> entityClass,
-			AggregationOperation... operations);
+	<O> AggregationResults<O> aggregate(String inputCollectionName, Aggregation<?, O> aggregation, Class<O> outputType);
 
 	/**
 	 * Execute a map-reduce operation. The map-reduce operation will be formed with an output type of INLINE

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

@@ -53,9 +53,9 @@ import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.mapping.model.BeanWrapper;
 import org.springframework.data.mapping.model.MappingException;
 import org.springframework.data.mongodb.MongoDbFactory;
-import org.springframework.data.mongodb.core.aggregation.AggregationOperation;
-import org.springframework.data.mongodb.core.aggregation.AggregationPipeline;
+import org.springframework.data.mongodb.core.aggregation.Aggregation;
 import org.springframework.data.mongodb.core.aggregation.AggregationResults;
+import org.springframework.data.mongodb.core.aggregation.TypedAggregation;
 import org.springframework.data.mongodb.core.convert.MappingMongoConverter;
 import org.springframework.data.mongodb.core.convert.MongoConverter;
 import org.springframework.data.mongodb.core.convert.MongoWriter;
@@ -119,6 +119,7 @@ import com.mongodb.util.JSONParseException;
  * @author Patryk Wasik
  * @author Tobias Trelle
  * @author Sebastian Herold
+ * @author Thomas Darimont
  */
 public class MongoTemplate implements MongoOperations, ApplicationContextAware {
 
@@ -1213,16 +1214,30 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware {
 
 	}
 
-	public <T> AggregationResults<T> aggregate(String inputCollectionName, AggregationPipeline pipeline,
-			Class<T> entityClass) {
+	@Override
+	public <I, O> AggregationResults<O> aggregate(TypedAggregation<I, O> aggregation, Class<O> outputType) {
+		return aggregate(aggregation, determineCollectionName(aggregation.getInputType()), outputType);
+	}
+
+	@Override
+	public <I, O> AggregationResults<O> aggregate(TypedAggregation<I, O> aggregation, String inputCollectionName,
+			Class<O> outputType) {
+		return aggregate(inputCollectionName, aggregation, outputType);
+	}
+
+	public <I, O> AggregationResults<O> aggregate(Class<I> inputType, Aggregation<I, O> aggregation, Class<O> outputType) {
+		return aggregate(determineCollectionName(inputType), aggregation, outputType);
+	}
+
+	public <O> AggregationResults<O> aggregate(String inputCollectionName, Aggregation<? extends Object, O> aggregation,
+			Class<O> outputType) {
 
 		Assert.notNull(inputCollectionName, "Collection name is missing");
-		Assert.notNull(pipeline, "Aggregation pipeline is missing");
-		Assert.notNull(entityClass, "Entity class is missing");
+		Assert.notNull(aggregation, "Aggregation pipeline is missing");
+		Assert.notNull(outputType, "Entity class is missing");
 
 		// prepare command
-		DBObject command = new BasicDBObject("aggregate", inputCollectionName);
-		command.put("pipeline", pipeline.getOperations());
+		DBObject command = aggregation.toDbObject(inputCollectionName);
 
 		// execute command
 		CommandResult commandResult = executeCommand(command);
@@ -1231,18 +1246,13 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware {
 		// map results
 		@SuppressWarnings("unchecked")
 		Iterable<DBObject> resultSet = (Iterable<DBObject>) commandResult.get("result");
-		List<T> mappedResults = new ArrayList<T>();
-		DbObjectCallback<T> callback = new ReadDbObjectCallback<T>(mongoConverter, entityClass);
+		List<O> mappedResults = new ArrayList<O>();
+		DbObjectCallback<O> callback = new ReadDbObjectCallback<O>(mongoConverter, outputType);
 		for (DBObject dbObject : resultSet) {
 			mappedResults.add(callback.doWith(dbObject));
 		}
 
-		return new AggregationResults<T>(mappedResults, commandResult);
-	}
-
-	public <T> AggregationResults<T> aggregate(String inputCollectionName, Class<T> entityClass,
-			AggregationOperation... operations) {
-		return aggregate(inputCollectionName, new AggregationPipeline(operations), entityClass);
+		return new AggregationResults<O>(mappedResults, commandResult);
 	}
 
 	protected String replaceWithResourceIfNecessary(String function) {

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

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2013 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 com.mongodb.BasicDBObject;
+import com.mongodb.DBObject;
+
+/**
+ * @author Thomas Darimont
+ */
+abstract class AbstractAggregateOperation implements AggregationOperation {
+	private final String operationName;
+
+	/**
+	 * @param operationName
+	 */
+	public AbstractAggregateOperation(String operationName) {
+		this.operationName = operationName;
+	}
+
+	/**
+	 * @return the operationName
+	 */
+	public String getOperationName() {
+		return operationName;
+	}
+
+	public String getOperationCommand() {
+		return OPERATOR_PREFIX + getOperationName();
+	}
+
+	/**
+	 * @return the argument for the operation
+	 */
+	public abstract Object getOperationArgument();
+
+	@Override
+	public DBObject toDbObject() {
+		return new BasicDBObject(getOperationCommand(), getOperationArgument());
+	}
+
+	/* (non-Javadoc)
+	 * @see java.lang.Object#toString()
+	 */
+	@Override
+	public String toString() {
+		return String.valueOf(toDbObject());
+	}
+}

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

@@ -0,0 +1,324 @@
+/*
+ * Copyright 2013 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.List;
+
+import org.springframework.data.domain.Sort;
+import org.springframework.data.domain.Sort.Direction;
+import org.springframework.data.mongodb.core.query.Criteria;
+import org.springframework.data.mongodb.core.query.NearQuery;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+import com.mongodb.BasicDBObject;
+import com.mongodb.DBObject;
+
+/**
+ * An {@code Aggregation} is a representation of a list of aggregation steps to be performed by the MongoDB Aggregation
+ * Framework.
+ * 
+ * @author Tobias Trelle - Original API and Implementation
+ * @author Thomas Darimont - Refactoring, embedded DSL
+ */
+public class Aggregation<I, O> {
+
+	private final List<AggregationOperation> operations = new ArrayList<AggregationOperation>();
+
+	/**
+	 * Creates a new {@link Aggregation} from the given {@link AggregationOperation}s.
+	 * 
+	 * @param operations must not be {@literal null} or empty.
+	 */
+	public Aggregation(AggregationOperation... operations) {
+		registerOperations(operations);
+	}
+
+	private void registerOperations(AggregationOperation... operations) {
+
+		Assert.notNull(operations, "Operations must not be null!");
+		Assert.isTrue(operations.length > 0, "operations must not be empty!");
+		for (AggregationOperation operation : operations) {
+			Assert.notNull(operation, "Operation is not allowed to be null");
+			this.operations.add(operation);
+		}
+	}
+
+	/**
+	 * Converts this {@link Aggregation} specification to a {@link DBObject}.
+	 * 
+	 * @param inputCollectionName the name of the input collection
+	 * @return the {@code DBObject} representing this aggregation
+	 */
+	public DBObject toDbObject(String inputCollectionName) {
+
+		DBObject command = new BasicDBObject("aggregate", inputCollectionName);
+		command.put("pipeline", getOperationObjects());
+		return command;
+	}
+
+	private List<DBObject> getOperationObjects() {
+
+		List<DBObject> operationObjects = new ArrayList<DBObject>();
+		for (AggregationOperation operation : operations) {
+			operationObjects.add(operation.toDbObject());
+		}
+		return operationObjects;
+	}
+
+	/* (non-Javadoc)
+	 * @see java.lang.Object#toString()
+	 */
+	@Override
+	public String toString() {
+		return StringUtils.collectionToCommaDelimitedString(operations);
+	}
+
+	/**
+	 * Factory method to create a new {@link GroupOperation} for the given {@code id}.
+	 * 
+	 * @param id, must not be {@literal null}
+	 * @return
+	 */
+	public static GroupOperation group(DBObject id) {
+		return new GroupOperation(id);
+	}
+
+	/**
+	 * Factory method to create a new {@link GroupOperation} for the given {@code idFields}.
+	 * 
+	 * @param idField the first idField to use, must not be {@literal null}.
+	 * @param moreIdFields more id fields to use, can be {@literal null}.
+	 * @return
+	 */
+	public static GroupOperation group(String idField, String... moreIdFields) {
+		return new GroupOperation(idField, moreIdFields);
+	}
+
+	/**
+	 * Factory method to create a new {@link GroupOperation} for the given {@code idFields}.
+	 * 
+	 * @param idFields
+	 * @return
+	 */
+	public static GroupOperation group(Fields idFields) {
+		return new GroupOperation(idFields);
+	}
+
+	/**
+	 * Factory method to create a new {@link ProjectionOperation} for the given {@code fields}. The {@code _id} field is
+	 * implicitly excluded.
+	 * 
+	 * @param fields a list of fields to include in the projection.
+	 * @return The {@link ProjectionOperation}.
+	 */
+	public static ProjectionOperation project(String... fields) {
+		return new ProjectionOperation(fields);
+	}
+
+	/**
+	 * Factory method to create a new {@link ProjectionOperation}.
+	 * 
+	 * @return The {@link ProjectionOperation}.
+	 */
+	public static ProjectionOperation project() {
+		return new ProjectionOperation();
+	}
+
+	/**
+	 * Factory method to create a new {@link ProjectionOperation} for the given {@code targetClass}.
+	 * 
+	 * @param targetClass
+	 * @return
+	 */
+	public static ProjectionOperation project(Class<?> targetClass) {
+		return new ProjectionOperation(targetClass);
+	}
+
+	/**
+	 * Factory method to create a new {@link MatchOperation} for the given {@link Criteria}.
+	 * 
+	 * @param criteria must not be {@literal null}
+	 * @return
+	 */
+	public static MatchOperation match(Criteria criteria) {
+		return new MatchOperation(criteria);
+	}
+
+	/**
+	 * Factory method to create a new {@link UnwindOperation} for the given {@literal fieldName}.
+	 * 
+	 * @param fieldName {@link UnwindOperation}.
+	 * @return
+	 */
+	public static UnwindOperation unwind(String fieldName) {
+		return new UnwindOperation(fieldName);
+	}
+
+	/**
+	 * Factory method to create a new {@link SkipOperation} for the given {@code skipCount}.
+	 * 
+	 * @param skipCount the number of documents to skip.
+	 * @return
+	 */
+	public static SkipOperation skip(int skipCount) {
+		return new SkipOperation(skipCount);
+	}
+
+	/**
+	 * Factory method to create a new {@link LimitOperation} for the given {@code maxElements}.
+	 * 
+	 * @param maxElements, the max number of documents to return.
+	 * @return
+	 */
+	public static LimitOperation limit(int maxElements) {
+		return new LimitOperation(maxElements);
+	}
+
+	/**
+	 * Factory method to create a new {@link GeoNearOperation} for the given {@code nearQuery}.
+	 * 
+	 * @param nearQuery, must not be {@literal null}.
+	 * @return
+	 */
+	public static GeoNearOperation geoNear(NearQuery nearQuery) {
+		return new GeoNearOperation(nearQuery);
+	}
+
+	/**
+	 * Factory method to create a new {@link SortOperation} for the given sort {@link Direction}  {@code direction} and
+	 * {@code fields}.
+	 * 
+	 * @param direction, the sort direction, must not be {@literal null}.
+	 * @param fields must not be {@literal null}.
+	 * @return
+	 */
+	public static SortOperation sort(Sort.Direction direction, String... fields) {
+		return sort(new Sort(direction, fields));
+	}
+
+	/**
+	 * Factory method to create a new {@link SortOperation} for the given {@link Sort}.
+	 * 
+	 * @param sort
+	 * @return
+	 */
+	public static SortOperation sort(Sort sort) {
+		return new SortOperation(sort);
+	}
+
+	/**
+	 * Factory method to create a new empty {@link Fields} container for key-value pairs.
+	 * 
+	 * @return
+	 */
+	public static Fields fields() {
+		return fields(new String[0]);
+	}
+
+	/**
+	 * Factory method to create a new {@link Fields} container for key-value pairs from the given {@code fieldNames}.
+	 * <p>
+	 * A call to fields("a","b","c") generates:
+	 * <p>
+	 * 
+	 * <pre>
+	 * {    
+	 *   a: $a,
+	 *   b: $b,
+	 *   c: $c
+	 * }
+	 * </pre>
+	 * 
+	 * @return
+	 */
+	public static Fields fields(String... fieldNames) {
+		return new BackendFields(fieldNames);
+	}
+
+	/**
+	 * A convenience shortcut to {@link ReferenceUtil#$id()}
+	 * 
+	 * @return
+	 */
+	public static String $id() {
+		return ReferenceUtil.$id();
+	}
+
+	/**
+	 * A convenience shortcut to {@link ReferenceUtil#$(String)}
+	 * 
+	 * @return
+	 */
+	public static String $(String fieldName) {
+		return ReferenceUtil.$(fieldName);
+	}
+
+	/**
+	 * A convenience shortcut to {@link ReferenceUtil#$id(String)}
+	 * 
+	 * @return
+	 */
+	public static String $id(String fieldName) {
+		return ReferenceUtil.$id(fieldName);
+	}
+
+	/**
+	 * A convenience shortcut to {@link ReferenceUtil#ID_KEY}
+	 * 
+	 * @return
+	 */
+	public static String id() {
+		return ReferenceUtil.ID_KEY;
+	}
+
+	/**
+	 * A convenience shortcut to {@link ReferenceUtil#id(String)}
+	 * 
+	 * @return
+	 */
+	public static String id(String fieldName) {
+		return ReferenceUtil.id(fieldName);
+	}
+
+	/**
+	 * Creates a new {@link Aggregation}.
+	 * 
+	 * @param <I> the input type of the {@link Aggregation}.
+	 * @param <O> the output type of the {@link Aggregation}.
+	 * @param inputType
+	 * @param operations
+	 * @return
+	 */
+	public static <I, O> TypedAggregation<I, O> newAggregation(Class<I> inputType, AggregationOperation... operations) {
+		return new TypedAggregation<I, O>(inputType, operations);
+	}
+
+	/**
+	 * Creates a new {@link Aggregation}.
+	 * 
+	 * @param <I> the input type of the {@link Aggregation}.
+	 * @param <O> the output type of the {@link Aggregation}.
+	 * @param inputType
+	 * @param operations
+	 * @return
+	 */
+	public static <I, O> Aggregation<I, O> newAggregation(AggregationOperation... operations) {
+		return new Aggregation<I, O>(operations);
+	}
+
+}

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

@@ -21,14 +21,17 @@ import com.mongodb.DBObject;
  * Represents one single operation in an aggregation pipeline.
  * 
  * @author Sebastian Herold
+ * @author Thomas Darimont
  * @since 1.3
  */
 public interface AggregationOperation {
 
+	String OPERATOR_PREFIX = "$";
+
 	/**
-	 * Gets the {@link DBObject} backing this operation
+	 * Creates a {@link DBObject} representation backing this object.
 	 * 
 	 * @return the DBObject
 	 */
-	DBObject getDBObject();
+	DBObject toDbObject();
 }

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

@@ -1,211 +0,0 @@
-/*
- * Copyright 2013 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.List;
-
-import org.springframework.data.domain.Sort;
-import org.springframework.data.mongodb.core.query.Criteria;
-import org.springframework.util.Assert;
-
-import com.mongodb.BasicDBObject;
-import com.mongodb.DBObject;
-import com.mongodb.util.JSON;
-import com.mongodb.util.JSONParseException;
-
-/**
- * Holds the operations of an aggregation pipeline.
- * 
- * @author Tobias Trelle
- * @since 1.3
- */
-public class AggregationPipeline {
-
-	private static final String OPERATOR_PREFIX = "$";
-
-	private final List<DBObject> operations = new ArrayList<DBObject>();
-
-	public AggregationPipeline() {
-	}
-
-	/**
-	 * Creates a new {@link AggregationPipeline} from the given {@link AggregationOperation}s.
-	 * 
-	 * @param operations must not be {@literal null} or empty.
-	 */
-	public AggregationPipeline(AggregationOperation... operations) {
-
-		Assert.notNull(operations, "Operations must not be null!");
-		Assert.isTrue(operations.length > 0, "operations must not be empty!");
-
-		for (AggregationOperation operation : operations) {
-			Assert.notNull(operation, "Operation is not allowed to be null");
-			this.operations.add(operation.getDBObject());
-		}
-	}
-
-	/**
-	 * Adds a projection operation to the pipeline.
-	 * 
-	 * @param projection JSON string holding the projection, must not be {@literal null} or empty.
-	 * @return The pipeline.
-	 */
-	public AggregationPipeline project(String projection) {
-		return addDocumentOperation("project", projection);
-	}
-
-	/**
-	 * Adds a projection operation to the pipeline.
-	 * 
-	 * @param projection Type safe projection object, must not be {@literal null}.
-	 * @return The pipeline.
-	 */
-	public AggregationPipeline project(Projection projection) {
-
-		Assert.notNull(projection, "Projection must not be null!");
-		return addOperation("project", projection.toDBObject());
-	}
-
-	/**
-	 * Adds an unwind operation to the pipeline.
-	 * 
-	 * @param field Name of the field to unwind (should be an array), must not be {@literal null} or empty.
-	 * @return The pipeline.
-	 */
-	public AggregationPipeline unwind(String field) {
-
-		Assert.hasText(field, "Missing field name");
-
-		if (!field.startsWith(OPERATOR_PREFIX)) {
-			field = OPERATOR_PREFIX + field;
-		}
-
-		return addOperation("unwind", field);
-	}
-
-	/**
-	 * Adds a group operation to the pipeline.
-	 * 
-	 * @param group JSON string holding the group, must not be {@literal null} or empty.
-	 * @return The pipeline.
-	 */
-	public AggregationPipeline group(String group) {
-		return addDocumentOperation("group", group);
-	}
-
-	/**
-	 * Adds a sort operation to the pipeline.
-	 * 
-	 * @param sort JSON string holding the sorting, must not be {@literal null} or empty.
-	 * @return The pipeline.
-	 */
-	public AggregationPipeline sort(String sort) {
-		return addDocumentOperation("sort", sort);
-	}
-
-	/**
-	 * Adds a sort operation to the pipeline.
-	 * 
-	 * @param sort Type safe sort operation, must not be {@literal null}.
-	 * @return The pipeline.
-	 */
-	public AggregationPipeline sort(Sort sort) {
-
-		Assert.notNull(sort);
-		DBObject dbo = new BasicDBObject();
-
-		for (org.springframework.data.domain.Sort.Order order : sort) {
-			dbo.put(order.getProperty(), order.isAscending() ? 1 : -1);
-		}
-		return addOperation("sort", dbo);
-	}
-
-	/**
-	 * Adds a match operation to the pipeline that is basically a query on the collections.
-	 * 
-	 * @param match JSON string holding the criteria, must not be {@literal null} or empty.
-	 * @return The pipeline.
-	 */
-	public AggregationPipeline match(String match) {
-		return addDocumentOperation("match", match);
-	}
-
-	/**
-	 * Adds a match operation to the pipeline that is basically a query on the collection.s
-	 * 
-	 * @param criteria Type safe criteria to filter documents from the collection, must not be {@literal null}.
-	 * @return The pipeline.
-	 */
-	public AggregationPipeline match(Criteria criteria) {
-
-		Assert.notNull(criteria);
-		return addOperation("match", criteria.getCriteriaObject());
-	}
-
-	/**
-	 * Adds an limit operation to the pipeline.
-	 * 
-	 * @param n Number of document to consider.
-	 * @return The pipeline.
-	 */
-	public AggregationPipeline limit(long n) {
-		return addOperation("limit", n);
-	}
-
-	/**
-	 * Adds an skip operation to the pipeline.
-	 * 
-	 * @param n Number of documents to skip.
-	 * @return The pipeline.
-	 */
-	public AggregationPipeline skip(long n) {
-		return addOperation("skip", n);
-	}
-
-	public List<DBObject> getOperations() {
-		return operations;
-	}
-
-	/**
-	 * creates an empty pipeline
-	 * 
-	 * @return the new pipeline
-	 */
-	public static AggregationPipeline pipeline() {
-		return new AggregationPipeline();
-	}
-
-	private AggregationPipeline addDocumentOperation(String opName, String operation) {
-
-		Assert.hasText(operation, "Operation must not be null or empty!");
-		return addOperation(opName, parseJson(operation));
-	}
-
-	private AggregationPipeline addOperation(String key, Object value) {
-		this.operations.add(new BasicDBObject(OPERATOR_PREFIX + key, value));
-		return this;
-	}
-
-	private DBObject parseJson(String json) {
-
-		try {
-			return (DBObject) JSON.parse(json);
-		} catch (JSONParseException e) {
-			throw new IllegalArgumentException("Invalid JSON document: " + json, e);
-		}
-	}
-}

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

@@ -0,0 +1,48 @@
+package org.springframework.data.mongodb.core.aggregation;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Implementation of {@link Fields}
+ * 
+ * @author Thomas Darimont
+ */
+class BackendFields implements Fields {
+
+	private Map<String, Object> values = new HashMap<String, Object>();
+
+	/**
+	 * @param names
+	 */
+	public BackendFields(String... names) {
+		for (String name : names) {
+			and(name);
+		}
+	}
+
+	/**
+	 * @param name
+	 * @return
+	 */
+	public Fields and(String name) {
+		return and(name, ReferenceUtil.safeReference(name));
+	}
+
+	/**
+	 * @param name
+	 * @param value
+	 * @return
+	 */
+	public Fields and(String name, Object value) {
+		this.values.put(name, value);
+		return this;
+	}
+
+	/**
+	 * @return
+	 */
+	public Map<String, Object> getValues() {
+		return new HashMap<String, Object>(this.values);
+	}
+}

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

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2013 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.Map;
+
+/**
+ * Fields is a collection of key-value pairs.
+ * 
+ * @author Thomas Darimont
+ */
+public interface Fields {
+	Map<String, Object> getValues();
+
+	Fields and(String name);
+
+	Fields and(String name, Object value);
+}

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

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2013 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 org.springframework.data.mongodb.core.query.NearQuery;
+import org.springframework.util.Assert;
+
+/**
+ * @author Thomas Darimont
+ */
+public class GeoNearOperation extends AbstractAggregateOperation {
+
+	private final NearQuery nearQuery;
+
+	public GeoNearOperation(NearQuery nearQuery) {
+		super("geoNear");
+		Assert.notNull(nearQuery);
+		this.nearQuery = nearQuery;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AbstractAggregateOperation#getOperationArgument()
+	 */
+	@Override
+	public Object getOperationArgument() {
+		return nearQuery.toDBObject();
+	}
+
+}

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

@@ -29,32 +29,81 @@ import com.mongodb.DBObject;
  * 
  * @see http://docs.mongodb.org/manual/reference/aggregation/group/#stage._S_group
  * @author Sebastian Herold
+ * @author Thomas Darimont
  * @since 1.3
  */
-public class GroupOperation implements AggregationOperation {
+public class GroupOperation extends AbstractAggregateOperation {
 
-	private static final String ID_KEY = "_id";
-
-	private final Object id;
-	private final Map<String, DBObject> fields = new HashMap<String, DBObject>();
+	final Object id;
+	final Map<String, DBObject> fields = new HashMap<String, DBObject>();
 
 	public GroupOperation(Object id) {
+		super("group");
 		this.id = id;
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#getDBObject()
+	/**
+	 * Creates a <code>$group</code> operation with <code>_id</code> referencing to a field of the document. The returned
+	 * db object equals to
+	 * 
+	 * <pre>
+	 * {_id: "$field"}
+	 * </pre>
+	 * 
+	 * @param id
+	 * @param moreIdFields
+	 */
+	public GroupOperation(String idField, String... moreIdFields) {
+		this(createGroupIdFrom(idField, moreIdFields));
+	}
+
+	/**
+	 * @param idField
+	 * @param moreIdFields
+	 * @return
 	 */
-	public DBObject getDBObject() {
+	private static Object createGroupIdFrom(String idField, String[] moreIdFields) {
+
+		Assert.notNull(idField, "idField must not be null!");
+		Object result = ReferenceUtil.safeReference(idField);
 
-		DBObject projection = new BasicDBObject(ID_KEY, id);
+		if (moreIdFields != null && moreIdFields.length > 0) {
+			DBObject idReferences = new BasicDBObject(moreIdFields.length + 1);
+			idReferences.put(ReferenceUtil.safeNonReference(idField), ReferenceUtil.safeReference(idField));
+
+			for (String additionalIdField : moreIdFields) {
+				idReferences.put(ReferenceUtil.safeNonReference(additionalIdField),
+						ReferenceUtil.safeReference(additionalIdField));
+			}
+
+			result = idReferences;
+		}
+
+		return result;
+	}
+
+	public GroupOperation(Fields idFields) {
+		this((Object) idFields.getValues());
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AbstractAggregateOperation#getOperationArgument()
+	 */
+	@Override
+	public Object getOperationArgument() {
+		return createProjection();
+	}
+
+	/**
+	 * @return
+	 */
+	private DBObject createProjection() {
+		DBObject projection = new BasicDBObject(ReferenceUtil.ID_KEY, id);
 
 		for (Entry<String, DBObject> entry : fields.entrySet()) {
 			projection.put(entry.getKey(), entry.getValue());
 		}
-
-		return new BasicDBObject("$group", projection);
+		return projection;
 	}
 
 	public GroupOperation addField(String key, DBObject value) {
@@ -64,7 +113,7 @@ public class GroupOperation implements AggregationOperation {
 
 		String trimmedKey = key.trim();
 
-		if (ID_KEY.equals(trimmedKey)) {
+		if (ReferenceUtil.ID_KEY.equals(trimmedKey)) {
 			throw new IllegalArgumentException("_id field can only be set in constructor");
 		}
 
@@ -263,18 +312,21 @@ public class GroupOperation implements AggregationOperation {
 	}
 
 	/**
-	 * Creates a <code>$group</code> operation with <code>_id</code> referencing to a field of the document. The returned
-	 * db object equals to
+	 * Adds a field with the <a href="http://docs.mongodb.org/manual/reference/aggregation/addToSet/#grp._S_sum">$sum
+	 * operation</a>.
 	 * 
 	 * <pre>
-	 * {_id: "$field"}
+	 *     {$group: {
+	 *          _id: "$id_field",
+	 *          field: {$sum: "$field"}
+	 *     }}
 	 * </pre>
 	 * 
-	 * @param field
+	 * @param field reference to a field of the document
 	 * @return
 	 */
-	public static GroupOperation group(String field) {
-		return new GroupOperation(ReferenceUtil.safeReference(field));
+	public GroupOperation sum(String field) {
+		return sum(field, field);
 	}
 
 	protected GroupOperation addOperation(String operation, String name, String field) {
@@ -303,6 +355,7 @@ public class GroupOperation implements AggregationOperation {
 	 * @param idFields
 	 * @return
 	 */
+	// TODO still relevant? IdField is probably a better abstraction than Fields?!
 	public static GroupOperation group(IdField... idFields) {
 		Assert.notNull(idFields, "Combined id is null");
 

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

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2013 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;
+
+/**
+ * @author Thomas Darimont
+ */
+class LimitOperation extends AbstractAggregateOperation {
+
+	private final long maxElements;
+
+	/**
+	 * @param maxElements Number of documents to consider.
+	 */
+	public LimitOperation(long maxElements) {
+		super("limit");
+		this.maxElements = maxElements;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AbstractAggregateOperation#getOperationArgument()
+	 */
+	@Override
+	public Object getOperationArgument() {
+		return maxElements;
+	}
+}

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

@@ -18,17 +18,16 @@ package org.springframework.data.mongodb.core.aggregation;
 import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.util.Assert;
 
-import com.mongodb.BasicDBObject;
 import com.mongodb.DBObject;
 
 /**
  * Encapsulates the {@code $match}-operation
  * 
  * @author Sebastian Herold
+ * @author Thomas Darimont
  * @since 1.3
  */
-public class MatchOperation implements AggregationOperation {
-
+public class MatchOperation extends AbstractAggregateOperation {
 	private final DBObject criteria;
 
 	/**
@@ -38,26 +37,16 @@ public class MatchOperation implements AggregationOperation {
 	 */
 	public MatchOperation(Criteria criteria) {
 
+		super("match");
 		Assert.notNull(criteria, "Criteria must not be null!");
 		this.criteria = criteria.getCriteriaObject();
 	}
 
-	/**
-	 * Factory method to create a new {@link MatchOperation} for the given {@link Criteria}.
-	 * 
-	 * @param criteria must not be {@literal null}.
-	 * @return
-	 */
-	public static MatchOperation match(Criteria criteria) {
-		return new MatchOperation(criteria);
-	}
-
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#getDBObject()
+	/* (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AbstractAggregateOperation#getOperationArgument()
 	 */
-	public DBObject getDBObject() {
-		return new BasicDBObject("$match", criteria);
+	@Override
+	public Object getOperationArgument() {
+		return criteria;
 	}
-
 }

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

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2013 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 com.mongodb.BasicDBObject;
+import com.mongodb.DBObject;
+
+/**
+ * @author Thomas Darimont
+ */
+public class NoopAggreationOperation implements AggregationOperation {
+	public DBObject toDbObject() {
+		return new BasicDBObject();
+	}
+}

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

@@ -29,26 +29,29 @@ import com.mongodb.BasicDBObject;
 import com.mongodb.DBObject;
 
 /**
- * Projection of field to be used in an {@link AggregationPipeline}.
- * <p/>
+ * Projection of field to be used in an {@link Aggregation}.
+ * <p>
  * A projection is similar to a {@link Field} inclusion/exclusion but more powerful. It can generate new fields, change
  * values of given field etc.
+ * <p>
  * 
  * @author Tobias Trelle
+ * @author Thomas Darimont
  * @since 1.3
  */
-public class Projection {
+public class ProjectionOperation extends AbstractAggregateOperation {
 
-    /** Stack of key names. Size is 0 or 1. */
+	/** Stack of key names. Size is 0 or 1. */
 	private final Stack<String> reference = new Stack<String>();
-	private final DBObject document = new BasicDBObject();
+	private final DBObject projection = new BasicDBObject();
 
 	private DBObject rightHandExpression;
 
 	/**
 	 * Create an empty projection.
 	 */
-	public Projection() {
+	public ProjectionOperation() {
+		super("project");
 	}
 
 	/**
@@ -56,7 +59,8 @@ public class Projection {
 	 * 
 	 * @param includes Keys of field to include, must not be {@literal null} or empty.
 	 */
-	public Projection(String... includes) {
+	public ProjectionOperation(String... includes) {
+		this();
 
 		Assert.notEmpty(includes);
 		exclude("_id");
@@ -66,15 +70,24 @@ public class Projection {
 		}
 	}
 
+	/**
+	 * Create an empty projection.
+	 * 
+	 * @param targetClass
+	 */
+	public ProjectionOperation(Class<?> targetClass) {
+		this();
+	}
+
 	/**
 	 * Excludes a given field.
 	 * 
 	 * @param key The key of the field.
 	 */
-	public final Projection exclude(String key) {
+	public final ProjectionOperation exclude(String key) {
 
 		Assert.hasText(key, "Missing key");
-		document.put(key, 0);
+		projection.put(key, 0);
 		return this;
 	}
 
@@ -83,7 +96,7 @@ public class Projection {
 	 * 
 	 * @param key The key of the field, must not be {@literal null} or empty.
 	 */
-	public final Projection include(String key) {
+	public final ProjectionOperation include(String key) {
 
 		Assert.hasText(key, "Missing key");
 
@@ -98,12 +111,12 @@ public class Projection {
 	 * 
 	 * @param key must not be {@literal null} or empty.
 	 */
-	public final Projection as(String key) {
+	public final ProjectionOperation as(String key) {
 
 		Assert.hasText(key, "Missing key");
 
 		try {
-			document.put(key, rightHandSide(ReferenceUtil.safeReference(reference.pop())));
+			projection.put(key, rightHandSide(ReferenceUtil.safeReference(reference.pop())));
 		} catch (EmptyStackException e) {
 			throw new InvalidDataAccessApiUsageException("Invalid use of as()", e);
 		}
@@ -111,15 +124,32 @@ public class Projection {
 		return this;
 	}
 
-	public final Projection plus(Number n) {
+	/**
+	 * Sets the key for a computed field.
+	 * 
+	 * @param key must not be {@literal null} or empty.
+	 */
+	public final ProjectionOperation asSelf() {
+
+		try {
+			String selfRef = reference.pop();
+			projection.put(selfRef, rightHandSide(ReferenceUtil.safeReference(selfRef)));
+		} catch (EmptyStackException e) {
+			throw new InvalidDataAccessApiUsageException("Invalid use of as()", e);
+		}
+
+		return this;
+	}
+
+	public final ProjectionOperation plus(Number n) {
 		return arithmeticOperation("add", n);
 	}
 
-	public final Projection minus(Number n) {
+	public final ProjectionOperation minus(Number n) {
 		return arithmeticOperation("substract", n);
 	}
 
-	private Projection arithmeticOperation(String op, Number n) {
+	private ProjectionOperation arithmeticOperation(String op, Number n) {
 
 		Assert.notNull(n, "Missing number");
 
@@ -138,18 +168,46 @@ public class Projection {
 	private void safePop() {
 
 		if (!reference.empty()) {
-			document.put(reference.pop(), rightHandSide(1));
+			projection.put(reference.pop(), rightHandSide(1));
 		}
 	}
 
-    private Object rightHandSide(Object defaultValue) {
+	private Object rightHandSide(Object defaultValue) {
 		Object value = rightHandExpression != null ? rightHandExpression : defaultValue;
 		rightHandExpression = null;
 		return value;
 	}
 
-	DBObject toDBObject() {
+	/**
+	 * @param string
+	 * @param projection
+	 * @return
+	 */
+	public ProjectionOperation addField(String key, Object value) {
+
+		Assert.notNull(key, "Missing Key");
+		Assert.notNull(value);
+		this.projection.put(key, value instanceof Fields ? ((Fields) value).getValues() : value);
+		return this;
+	}
+
+	/**
+	 * @param name
+	 * @param value
+	 * @return
+	 */
+	public ProjectionOperation field(String name, Object value) {
+		addField(name, value);
+		return this;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AbstractAggregateOperation#getOperationArgument()
+	 */
+	@Override
+	public Object getOperationArgument() {
+
 		safePop();
-		return document;
+		return projection;
 	}
 }

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

@@ -21,10 +21,13 @@ import org.springframework.util.Assert;
  * Utility class for mongo db reference operator <code>$</code>
  * 
  * @author Sebastian Herold
+ * @author Thomas Darimont
  * @since 1.3
  */
 class ReferenceUtil {
 
+	public static final String ID_KEY = "_id";
+
 	private static final String REFERENCE_PREFIX = "$";
 
 	/**
@@ -60,4 +63,46 @@ class ReferenceUtil {
 
 		return field;
 	}
+
+	/**
+	 * @return $_id
+	 */
+	public static String $id() {
+		return $(ID_KEY);
+	}
+
+	/**
+	 * <pre>
+	 *  $("a") -> $a
+	 * </pre>
+	 * 
+	 * @param fieldName
+	 * @return the field name prefixed with {@literal $}
+	 * @see #safeReference(String)
+	 */
+	public static String $(String fieldName) {
+		return safeReference(fieldName);
+	}
+
+	/**
+	 * <pre>
+	 * 	$id("a") -> $_id.a
+	 * </pre>
+	 * 
+	 * @param fieldName
+	 * @return
+	 * @see #safeNonReference(String)
+	 */
+	public static String $id(String fieldName) {
+		return $id() + "." + safeNonReference(fieldName);
+	}
+
+	/**
+	 * @param fieldName
+	 * @return
+	 */
+	public static String id(String fieldName) {
+		return ID_KEY + "." + fieldName;
+	}
+
 }

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

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2013 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;
+
+/**
+ * @author Thomas Darimont
+ */
+public class SkipOperation extends AbstractAggregateOperation {
+
+	private final long skipCount;
+
+	/**
+	 * @param skipCount number of documents to skip.
+	 */
+	public SkipOperation(long skipCount) {
+		super("skip");
+		this.skipCount = skipCount;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AbstractAggregateOperation#getOperationArgument()
+	 */
+	@Override
+	public Object getOperationArgument() {
+		return skipCount;
+	}
+}

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

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2013 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 org.springframework.data.domain.Sort;
+import org.springframework.util.Assert;
+
+import com.mongodb.BasicDBObject;
+import com.mongodb.DBObject;
+
+/**
+ * @author Thomas Darimont
+ */
+public class SortOperation extends AbstractAggregateOperation {
+
+	private Sort sort;
+
+	/**
+	 * @param sort
+	 */
+	public SortOperation(Sort sort) {
+		super("sort");
+
+		Assert.notNull(sort);
+		this.sort = sort;
+	}
+
+	public SortOperation and(Sort sort) {
+		return new SortOperation(this.sort.and(sort));
+	}
+
+	public SortOperation and(Sort.Direction direction, String... fields) {
+		return and(new Sort(direction, fields));
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AbstractAggregateOperation#getOperationArgument()
+	 */
+	@Override
+	public Object getOperationArgument() {
+		return createSortProperties();
+	}
+
+	/**
+	 * @return
+	 */
+	private DBObject createSortProperties() {
+		DBObject sortProperties = new BasicDBObject();
+		for (org.springframework.data.domain.Sort.Order order : sort) {
+			sortProperties.put(order.getProperty(), order.isAscending() ? 1 : -1);
+		}
+		return sortProperties;
+	}
+}

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

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2013 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;
+
+/**
+ * A {@code TypedAggregation} is a special {@link Aggregation} that holds information of the input aggregation type.
+ * 
+ * @author Thomas Darimont
+ */
+public class TypedAggregation<I, O> extends Aggregation<I, O> {
+
+	private Class<I> inputType;
+
+	/**
+	 * Creates a new {@link TypedAggregation} from the given {@link AggregationOperation}s.
+	 * 
+	 * @param operations must not be {@literal null} or empty.
+	 */
+	public TypedAggregation(Class<I> inputType, AggregationOperation... operations) {
+		super(operations);
+		this.inputType = inputType;
+	}
+
+	/**
+	 * @return the inputType
+	 */
+	public Class<?> getInputType() {
+		return inputType;
+	}
+
+}

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

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2013 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 org.springframework.util.Assert;
+
+/**
+ * @author Thomas Darimont
+ */
+public class UnwindOperation extends AbstractAggregateOperation {
+
+	private final String fieldName;
+
+	public UnwindOperation(String fieldName) {
+		super("unwind");
+		Assert.notNull(fieldName);
+		this.fieldName = fieldName;
+	}
+
+	@Override
+	public Object getOperationArgument() {
+		return ReferenceUtil.safeReference(fieldName);
+	}
+}

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

@@ -0,0 +1,19 @@
+/*
+ * Copyright 2013 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.
+ */
+/**
+ * @author Thomas Darimont
+ */
+package org.springframework.data.mongodb.core.aggregation;

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

@@ -17,14 +17,11 @@ package org.springframework.data.mongodb.core.aggregation;
 
 import static org.hamcrest.CoreMatchers.*;
 import static org.junit.Assert.*;
+import static org.springframework.data.domain.Sort.Direction.*;
 import static org.springframework.data.mongodb.core.DBObjectUtils.*;
+import static org.springframework.data.mongodb.core.aggregation.Aggregation.*;
 
-import java.util.List;
-
-import org.junit.Before;
 import org.junit.Test;
-import org.springframework.data.domain.Sort;
-import org.springframework.data.domain.Sort.Direction;
 import org.springframework.data.mongodb.core.query.Criteria;
 
 import com.mongodb.DBObject;
@@ -34,62 +31,38 @@ import com.mongodb.DBObject;
  * 
  * @see DATAMONGO-586
  * @author Tobias Trelle
+ * @author Thomas Darimont
  */
 public class AggregationPipelineTests {
 
-	AggregationPipeline pipeline;
-
-	@Before
-	public void setUp() {
-		pipeline = new AggregationPipeline();
-	}
-
 	@Test
 	public void limitOperation() {
 
-		pipeline.limit(42);
-
-		List<DBObject> rawPipeline = pipeline.getOperations();
-		assertDBObject("$limit", 42L, rawPipeline);
+		assertSingleDBObject("$limit", 42L, limit(42).toDbObject());
 	}
 
 	@Test
 	public void skipOperation() {
 
-		pipeline.skip(5);
-
-		List<DBObject> rawPipeline = pipeline.getOperations();
-		assertDBObject("$skip", 5L, rawPipeline);
+		assertSingleDBObject("$skip", 5L, skip(5).toDbObject());
 	}
 
 	@Test
 	public void unwindOperation() {
 
-		pipeline.unwind("$field");
-
-		List<DBObject> rawPipeline = pipeline.getOperations();
-		assertDBObject("$unwind", "$field", rawPipeline);
+		assertSingleDBObject("$unwind", "$field", unwind("$field").toDbObject());
 	}
 
 	@Test
 	public void unwindOperationWithAddedPrefix() {
 
-		pipeline.unwind("field");
-
-		List<DBObject> rawPipeline = pipeline.getOperations();
-		assertDBObject("$unwind", "$field", rawPipeline);
+		assertSingleDBObject("$unwind", "$field", unwind("field").toDbObject());
 	}
 
 	@Test
 	public void matchOperation() {
 
-		Criteria criteria = new Criteria("title").is("Doc 1");
-		pipeline.match(criteria);
-
-		List<DBObject> rawPipeline = pipeline.getOperations();
-		assertOneDocument(rawPipeline);
-
-		DBObject match = rawPipeline.get(0);
+		DBObject match = match(new Criteria("title").is("Doc 1")).toDbObject();
 		DBObject criteriaDoc = getAsDBObject(match, "$match");
 		assertThat(criteriaDoc, is(notNullValue()));
 		assertSingleDBObject("title", "Doc 1", criteriaDoc);
@@ -98,13 +71,7 @@ public class AggregationPipelineTests {
 	@Test
 	public void sortOperation() {
 
-		Sort sort = new Sort(new Sort.Order(Direction.ASC, "n"));
-		pipeline.sort(sort);
-
-		List<DBObject> rawPipeline = pipeline.getOperations();
-		assertOneDocument(rawPipeline);
-
-		DBObject sortDoc = rawPipeline.get(0);
+		DBObject sortDoc = sort(ASC, "n").toDbObject();
 		DBObject orderDoc = getAsDBObject(sortDoc, "$sort");
 		assertThat(orderDoc, is(notNullValue()));
 		assertSingleDBObject("n", 1, orderDoc);
@@ -113,30 +80,12 @@ public class AggregationPipelineTests {
 	@Test
 	public void projectOperation() {
 
-		Projection projection = new Projection("a");
-		pipeline.project(projection);
-
-		List<DBObject> rawPipeline = pipeline.getOperations();
-		assertOneDocument(rawPipeline);
-
-		DBObject projectionDoc = rawPipeline.get(0);
+		DBObject projectionDoc = project("a").toDbObject();
 		DBObject fields = getAsDBObject(projectionDoc, "$project");
 		assertThat(fields, is(notNullValue()));
 		assertSingleDBObject("a", 1, fields);
 	}
 
-	private static void assertOneDocument(List<DBObject> result) {
-
-		assertThat(result, is(notNullValue()));
-		assertThat(result.size(), is(1));
-	}
-
-	private static void assertDBObject(String key, Object value, List<DBObject> result) {
-
-		assertOneDocument(result);
-		assertSingleDBObject(key, value, result.get(0));
-	}
-
 	private static void assertSingleDBObject(String key, Object value, DBObject doc) {
 		assertThat(doc.get(key), is(value));
 	}

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

@@ -17,44 +17,57 @@ package org.springframework.data.mongodb.core.aggregation;
 
 import static org.hamcrest.CoreMatchers.*;
 import static org.junit.Assert.*;
+import static org.springframework.data.domain.Sort.Direction.*;
+import static org.springframework.data.mongodb.core.aggregation.Aggregation.*;
+import static org.springframework.data.mongodb.core.query.Criteria.*;
 
+import java.io.BufferedInputStream;
 import java.util.ArrayList;
+import java.util.Date;
 import java.util.List;
+import java.util.Scanner;
 
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.dao.InvalidDataAccessApiUsageException;
-import org.springframework.data.domain.Sort;
-import org.springframework.data.domain.Sort.Direction;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.dao.DataAccessException;
+import org.springframework.data.mongodb.core.CollectionCallback;
 import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 import com.mongodb.BasicDBObject;
 import com.mongodb.DBCollection;
 import com.mongodb.DBObject;
+import com.mongodb.MongoException;
+import com.mongodb.WriteConcern;
+import com.mongodb.util.JSON;
 
 /**
  * Tests for {@link MongoTemplate#aggregate(String, AggregationPipeline, Class)}.
  * 
  * @see DATAMONGO-586
  * @author Tobias Trelle
+ * @author Thomas Darimont
  */
 @RunWith(SpringJUnit4ClassRunner.class)
+@DirtiesContext
 @ContextConfiguration("classpath:infrastructure.xml")
 public class AggregationTests {
 
 	private static final String INPUT_COLLECTION = "aggregation_test_collection";
 
-	@Autowired
-	MongoTemplate mongoTemplate;
+	@Autowired MongoTemplate mongoTemplate;
 
 	@Before
 	public void setUp() {
+		mongoTemplate.setWriteConcern(WriteConcern.SAFE); // safe due to DirtiesContext
 		cleanDb();
+		initSampleDataIfNecessary();
 	}
 
 	@After
@@ -62,9 +75,48 @@ public class AggregationTests {
 		cleanDb();
 	}
 
+	private void cleanDb() {
+		mongoTemplate.dropCollection(INPUT_COLLECTION);
+		mongoTemplate.dropCollection(UserWithLikes.class);
+	}
+
+	/**
+	 * Imports the sample dataset (zips.json) if necessary (e.g. if it doen't exist yet). The dataset can originally be
+	 * found on the mongodb aggregation framework example website:
+	 * 
+	 * @see http://docs.mongodb.org/manual/tutorial/aggregation-examples/.
+	 */
+	private void initSampleDataIfNecessary() {
+
+		if (!mongoTemplate.collectionExists(ZipInfo.class)) {
+			mongoTemplate.dropCollection(ZipInfo.class);
+
+			mongoTemplate.execute(ZipInfo.class, new CollectionCallback<Void>() {
+				@Override
+				public Void doInCollection(DBCollection collection) throws MongoException, DataAccessException {
+					Scanner scanner = null;
+					try {
+						scanner = new Scanner(new BufferedInputStream(new ClassPathResource("zips.json").getInputStream()));
+						while (scanner.hasNextLine()) {
+							String zipInfoRecord = scanner.nextLine();
+							collection.save((DBObject) JSON.parse(zipInfoRecord));
+						}
+					} catch (Exception e) {
+						if (scanner != null) {
+							scanner.close();
+						}
+						throw new RuntimeException("Could not load mongodb sample dataset!", e);
+					}
+
+					return null;
+				}
+			});
+		}
+	}
+
 	@Test(expected = IllegalArgumentException.class)
 	public void shouldHandleMissingInputCollection() {
-		mongoTemplate.aggregate(null, new AggregationPipeline(), TagCount.class);
+		mongoTemplate.aggregate((String) null, new Aggregation<Object, TagCount>(), TagCount.class);
 	}
 
 	@Test(expected = IllegalArgumentException.class)
@@ -74,29 +126,23 @@ public class AggregationTests {
 
 	@Test(expected = IllegalArgumentException.class)
 	public void shouldHandleMissingEntityClass() {
-		mongoTemplate.aggregate(INPUT_COLLECTION, new AggregationPipeline(), null);
-	}
-
-	@Test(expected = IllegalArgumentException.class)
-	public void shouldDetectIllegalJsonInOperation() {
-
-		AggregationPipeline pipeline = new AggregationPipeline().project("{ foo bar");
-		mongoTemplate.aggregate(INPUT_COLLECTION, pipeline, TagCount.class);
+		mongoTemplate.aggregate(INPUT_COLLECTION, new Aggregation<Object, TagCount>(), null);
 	}
 
 	@Test
 	public void shouldAggregate() {
 
-		createDocuments();
+		createTagDocuments();
 
-		AggregationPipeline pipeline = new AggregationPipeline(). //
-				project("{_id:0,tags:1}}"). //
-				unwind("tags"). //
-				group("{_id:\"$tags\", n:{$sum:1}}"). //
-				project("{tag: \"$_id\", n:1, _id:0}"). //
-				sort(new Sort(new Sort.Order(Direction.DESC, "n")));
+		Aggregation<Object, TagCount> agg = newAggregation( //
+				project("tags"), //
+				unwind("tags"), //
+				group($("tags")).count("n"), //
+				project().field("tag", $id()).field("n", 1), //
+				sort(DESC, "n") //
+		);
 
-		AggregationResults<TagCount> results = mongoTemplate.aggregate(INPUT_COLLECTION, pipeline, TagCount.class);
+		AggregationResults<TagCount> results = mongoTemplate.aggregate(INPUT_COLLECTION, agg, TagCount.class);
 
 		assertThat(results, is(notNullValue()));
 		assertThat(results.getServerUsed(), is("/127.0.0.1:27017"));
@@ -111,25 +157,18 @@ public class AggregationTests {
 		assertTagCount("nosql", 1, tagCount.get(2));
 	}
 
-	@Test(expected = InvalidDataAccessApiUsageException.class)
-	public void shouldDetectIllegalAggregationOperation() {
-
-		createDocuments();
-		AggregationPipeline pipeline = new AggregationPipeline().project("{$foobar:{_id:0,tags:1}}");
-		mongoTemplate.aggregate(INPUT_COLLECTION, pipeline, TagCount.class);
-	}
-
 	@Test
 	public void shouldAggregateEmptyCollection() {
 
-		AggregationPipeline pipeline = new AggregationPipeline(). //
-				project("{_id:0,tags:1}}"). //
-				unwind("$tags").//
-				group("{_id:\"$tags\", n:{$sum:1}}").//
-				project("{tag: \"$_id\", n:1, _id:0}").//
-				sort("{n:-1}");
+		Aggregation<Object, TagCount> agg = newAggregation(//
+				project("tags"), //
+				unwind("tags"), //
+				group($("tags")).count("n"), //
+				project().field("tag", $id()).field("n", 1), //
+				sort(DESC, "n") //
+		);
 
-		AggregationResults<TagCount> results = mongoTemplate.aggregate(INPUT_COLLECTION, pipeline, TagCount.class);
+		AggregationResults<TagCount> results = mongoTemplate.aggregate(INPUT_COLLECTION, agg, TagCount.class);
 
 		assertThat(results, is(notNullValue()));
 		assertThat(results.getServerUsed(), is("/127.0.0.1:27017"));
@@ -143,14 +182,15 @@ public class AggregationTests {
 	@Test
 	public void shouldDetectResultMismatch() {
 
-		createDocuments();
-		AggregationPipeline pipeline = new AggregationPipeline(). //
-				project("{_id:0,tags:1}}"). //
-				unwind("$tags"). //
-				group("{_id:\"$tags\", count:{$sum:1}}"). //
-				limit(2);
+		createTagDocuments();
+		Aggregation<Object, TagCount> agg = newAggregation( //
+				project("tags"), //
+				unwind("tags"), //
+				group($("tags")).count("count"), //
+				limit(2) //
+		);
 
-		AggregationResults<TagCount> results = mongoTemplate.aggregate(INPUT_COLLECTION, pipeline, TagCount.class);
+		AggregationResults<TagCount> results = mongoTemplate.aggregate(INPUT_COLLECTION, agg, TagCount.class);
 
 		assertThat(results, is(notNullValue()));
 		assertThat(results.getServerUsed(), is("/127.0.0.1:27017"));
@@ -163,11 +203,241 @@ public class AggregationTests {
 		assertTagCount(null, 0, tagCount.get(1));
 	}
 
-	protected void cleanDb() {
-		mongoTemplate.dropCollection(INPUT_COLLECTION);
+	@Test
+	public void fieldsFactoryMethod() {
+
+		Fields fields = fields("a", "b").and("c").and("d", 42);
+		assertThat(fields, is(notNullValue()));
+		assertThat(fields.getValues(), is(notNullValue()));
+		assertThat(fields.getValues().size(), is(4));
+		assertThat(fields.getValues().get("a"), is((Object) "$a"));
+		assertThat(fields.getValues().get("b"), is((Object) "$b"));
+		assertThat(fields.getValues().get("c"), is((Object) "$c"));
+		assertThat(fields.getValues().get("d"), is((Object) 42));
 	}
 
-	private void createDocuments() {
+	@Test
+	public void groupFactoryMethodWithFieldsAndSumOperation() {
+
+		Fields fields = fields("a", "b").and("c").and("d", 42);
+		GroupOperation groupOperation = group(fields).sum("e");
+
+		assertThat(groupOperation, is(notNullValue()));
+		assertThat(groupOperation.toDbObject(), is(notNullValue()));
+		assertThat(groupOperation.id, is(notNullValue()));
+		assertThat(groupOperation.id, is((Object) fields.getValues()));
+		assertThat(groupOperation.fields, is(notNullValue()));
+		assertThat(groupOperation.fields.size(), is(1));
+		assertThat(groupOperation.fields.containsKey("e"), is(true));
+		assertThat(groupOperation.fields.get("e"), is(notNullValue()));
+		assertThat(groupOperation.fields.get("e").get("$sum"), is(notNullValue()));
+		assertThat(groupOperation.fields.get("e").get("$sum"), is((Object) "$e"));
+	}
+
+	@Test
+	public void complexAggregationFrameworkUsageLargestAndSmallestCitiesByState() {
+		/*
+		 //complex mongodb aggregation framework example from http://docs.mongodb.org/manual/tutorial/aggregation-examples/#largest-and-smallest-cities-by-state
+		db.zipInfo.aggregate( 
+			{
+			   $group: {
+			      _id: {
+			         state: '$state',
+			         city: '$city'
+			      },
+			      pop: {
+			         $sum: '$pop'
+			      }
+			   }
+			},
+			{
+			   $sort: {
+			      pop: 1,
+			      '_id.state': 1,
+			      '_id.city': 1
+			   }
+			},
+			{
+			   $group: {
+			      _id: '$_id.state',
+			      biggestCity: {
+			         $last: '$_id.city'
+			      },
+			      biggestPop: {
+			         $last: '$pop'
+			      },
+			      smallestCity: {
+			         $first: '$_id.city'
+			      },
+			      smallestPop: {
+			         $first: '$pop'
+			      }
+			   }
+			},
+			{
+			   $project: {
+			      _id: 0,
+			      state: '$_id',
+			      biggestCity: {
+			         name: '$biggestCity',
+			         pop: '$biggestPop'
+			      },
+			      smallestCity: {
+			         name: '$smallestCity',
+			         pop: '$smallestPop'
+			      }
+			   }
+			},
+			{
+			   $sort: {
+			      state: 1
+			   }
+			}
+		)
+		*/
+
+		TypedAggregation<ZipInfo, ZipInfoStats> agg = newAggregation(
+				ZipInfo.class, //
+				group("state", "city").sum("pop"), // group("state", "city") -> _id: {state: $state, city: $city}
+				sort(ASC, "pop", id("state"), id("city")), //
+				group($id("state")) // $id("state") -> _id : $_id.state
+						.last("biggestCity", $id("city")) //
+						.last("biggestPop", $("pop")) //
+						.first("smallestCity", $id("city")) //
+						.first("smallestPop", $("pop")), //
+				project(ZipInfoStats.class) //
+						.field("_id", 0) //
+						.field("state", $id()) // $id() -> $_id
+						.field("biggestCity", fields().and("name", $("biggestCity")).and("population", $("biggestPop"))) //
+						.field("smallestCity", fields().and("name", $("smallestCity")).and("population", $("smallestPop"))),
+				sort(ASC, "state") //
+		);
+
+		assertThat(agg, is(notNullValue()));
+		assertThat(agg.toString(), is(notNullValue()));
+
+		AggregationResults<ZipInfoStats> result = mongoTemplate.aggregate(agg, ZipInfoStats.class);
+		assertThat(result, is(notNullValue()));
+		assertThat(result.getAggregationResult(), is(notNullValue()));
+		assertThat(result.getAggregationResult().size(), is(51));
+
+		ZipInfoStats firstZipInfoStats = result.getAggregationResult().get(0);
+		assertThat(firstZipInfoStats, is(notNullValue()));
+		assertThat(firstZipInfoStats.id, is(nullValue()));
+		assertThat(firstZipInfoStats.state, is("AK"));
+		assertThat(firstZipInfoStats.smallestCity, is(notNullValue()));
+		assertThat(firstZipInfoStats.smallestCity.name, is("CHEVAK"));
+		assertThat(firstZipInfoStats.smallestCity.population, is(0));
+		assertThat(firstZipInfoStats.biggestCity, is(notNullValue()));
+		assertThat(firstZipInfoStats.biggestCity.name, is("ANCHORAGE"));
+		assertThat(firstZipInfoStats.biggestCity.population, is(183987));
+
+		ZipInfoStats lastZipInfoStats = result.getAggregationResult().get(50);
+		assertThat(lastZipInfoStats, is(notNullValue()));
+		assertThat(lastZipInfoStats.id, is(nullValue()));
+		assertThat(lastZipInfoStats.state, is("WY"));
+		assertThat(lastZipInfoStats.smallestCity, is(notNullValue()));
+		assertThat(lastZipInfoStats.smallestCity.name, is("LOST SPRINGS"));
+		assertThat(lastZipInfoStats.smallestCity.population, is(6));
+		assertThat(lastZipInfoStats.biggestCity, is(notNullValue()));
+		assertThat(lastZipInfoStats.biggestCity.name, is("CHEYENNE"));
+		assertThat(lastZipInfoStats.biggestCity.population, is(70185));
+	}
+
+	@Test
+	public void findStatesWithPopulationOver10MillionAggregationExample() {
+		/*
+		 //complex mongodb aggregation framework example from http://docs.mongodb.org/manual/tutorial/aggregation-examples/#largest-and-smallest-cities-by-state
+		 db.zipcodes.aggregate( 
+			 	{
+				   $group:{
+				      _id:"$state",
+				      totalPop:{ $sum:"$pop"}
+		 			 }
+				},
+				{ 
+		 			$sort: { _id: 1, "totalPop": 1 } 
+		 		},
+				{
+				   $match:{
+				      totalPop: { $gte:10*1000*1000 }
+				   }
+				}
+		)
+		  */
+
+		TypedAggregation<ZipInfo, StateStats> agg = newAggregation(ZipInfo.class, //
+				group("state").sum("totalPop", $("pop")), // fields("state", "city") -> state: $state, city: $city
+				sort(ASC, id(), "totalPop"), //
+				match(where("totalPop").gte(10 * 1000 * 1000)) //
+		);
+
+		assertThat(agg, is(notNullValue()));
+		assertThat(agg.toString(), is(notNullValue()));
+
+		AggregationResults<StateStats> result = mongoTemplate.aggregate(agg, StateStats.class);
+		assertThat(result, is(notNullValue()));
+		assertThat(result.getAggregationResult(), is(notNullValue()));
+		assertThat(result.getAggregationResult().size(), is(7));
+
+		StateStats stateStats = result.getAggregationResult().get(0);
+		assertThat(stateStats, is(notNullValue()));
+		assertThat(stateStats.id, is("CA"));
+		assertThat(stateStats.state, is(nullValue()));
+		assertThat(stateStats.totalPopulation, is(29760021));
+	}
+
+	/**
+	 * @see http://docs.mongodb.org/manual/tutorial/aggregation-examples/#return-the-five-most-common-likes
+	 */
+	@Test
+	public void returnFiveMostCommonLikesAggregationFrameworkExample() {
+
+		createUserWithLikesDocuments();
+
+		TypedAggregation<UserWithLikes, LikeStats> agg = newAggregation(UserWithLikes.class, //
+				unwind("likes"), //
+				group("likes").count("number"), //
+				sort(DESC, "number"), //
+				limit(5), //
+				sort(ASC, id()) //
+		);
+
+		assertThat(agg, is(notNullValue()));
+		assertThat(agg.toString(), is(notNullValue()));
+
+		AggregationResults<LikeStats> result = mongoTemplate.aggregate(agg, LikeStats.class);
+		assertThat(result, is(notNullValue()));
+		assertThat(result.getAggregationResult(), is(notNullValue()));
+		assertThat(result.getAggregationResult().size(), is(5));
+
+		assertLikeStats(result.getAggregationResult().get(0), "a", 4);
+		assertLikeStats(result.getAggregationResult().get(1), "b", 2);
+		assertLikeStats(result.getAggregationResult().get(2), "c", 4);
+		assertLikeStats(result.getAggregationResult().get(3), "d", 2);
+		assertLikeStats(result.getAggregationResult().get(4), "e", 3);
+	}
+
+	private void assertLikeStats(LikeStats like, String id, long count) {
+
+		assertThat(like, is(notNullValue()));
+		assertThat(like.id, is(id));
+		assertThat(like.count, is(count));
+	}
+
+	private void createUserWithLikesDocuments() {
+		mongoTemplate.insert(new UserWithLikes("u1", new Date(), "a", "b", "c"));
+		mongoTemplate.insert(new UserWithLikes("u2", new Date(), "a"));
+		mongoTemplate.insert(new UserWithLikes("u3", new Date(), "b", "c"));
+		mongoTemplate.insert(new UserWithLikes("u4", new Date(), "c", "d", "e"));
+		mongoTemplate.insert(new UserWithLikes("u5", new Date(), "a", "e", "c"));
+		mongoTemplate.insert(new UserWithLikes("u6", new Date()));
+		mongoTemplate.insert(new UserWithLikes("u7", new Date(), "a"));
+		mongoTemplate.insert(new UserWithLikes("u8", new Date(), "x", "e"));
+		mongoTemplate.insert(new UserWithLikes("u9", new Date(), "y", "d"));
+	}
+
+	private void createTagDocuments() {
 
 		DBCollection coll = mongoTemplate.getCollection(INPUT_COLLECTION);
 
@@ -194,4 +464,5 @@ public class AggregationTests {
 		assertThat(tagCount.getTag(), is(tag));
 		assertThat(tagCount.getN(), is(n));
 	}
+
 }

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

@@ -0,0 +1,11 @@
+package org.springframework.data.mongodb.core.aggregation;
+
+class City {
+
+	String name;
+	int population;
+
+	public String toString() {
+		return "City [name=" + name + ", population=" + population + "]";
+	}
+}

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

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2013 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 org.springframework.data.mongodb.core.mapping.Field;
+
+/**
+ * @author Thomas Darimont
+ */
+public class LikeStats {
+
+	String id;
+	@Field("number") long count;
+}

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

@@ -17,49 +17,40 @@ package org.springframework.data.mongodb.core.aggregation;
 
 import static org.hamcrest.CoreMatchers.*;
 import static org.junit.Assert.*;
+import static org.springframework.data.mongodb.core.aggregation.Aggregation.*;
 
 import java.util.List;
 
-import org.junit.Before;
 import org.junit.Test;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
 
 import com.mongodb.DBObject;
 
 /**
- * Tests of {@link Projection}.
+ * Tests of {@link ProjectionOperation}.
  * 
  * @see DATAMONGO-586
  * @author Tobias Trelle
  */
 public class ProjectionTests {
 
-	Projection projection;
-
-	@Before
-	public void setUp() {
-		projection = new Projection();
-	}
-
 	@Test
 	public void emptyProjection() {
 
-		DBObject raw = projection.toDBObject();
+		DBObject raw = safeExtractDbObjectFromProjection(project());
 		assertThat(raw, is(notNullValue()));
 		assertThat(raw.toMap().isEmpty(), is(true));
 	}
 
 	@Test(expected = IllegalArgumentException.class)
 	public void shouldDetectNullIncludesInConstructor() {
-		new Projection((String[]) null);
+		new ProjectionOperation((String[]) null);
 	}
 
 	@Test
 	public void includesWithConstructor() {
 
-		projection = new Projection("a", "b");
-
-		DBObject raw = projection.toDBObject();
+		DBObject raw = safeExtractDbObjectFromProjection(project("a", "b"));
 		assertThat(raw, is(notNullValue()));
 		assertThat(raw.toMap().size(), is(3));
 		assertThat((Integer) raw.get("_id"), is(0));
@@ -70,47 +61,39 @@ public class ProjectionTests {
 	@Test
 	public void include() {
 
-		projection.include("a");
-
-		DBObject raw = projection.toDBObject();
+		DBObject raw = safeExtractDbObjectFromProjection(project().include("a"));
 		assertSingleDBObject("a", 1, raw);
 	}
 
 	@Test
 	public void exclude() {
 
-		projection.exclude("a");
-
-		DBObject raw = projection.toDBObject();
+		DBObject raw = safeExtractDbObjectFromProjection(project().exclude("a"));
 		assertSingleDBObject("a", 0, raw);
 	}
 
 	@Test
 	public void includeAlias() {
 
-		projection.include("a").as("b");
-
-		DBObject raw = projection.toDBObject();
+		DBObject raw = safeExtractDbObjectFromProjection(project().include("a").as("b"));
 		assertSingleDBObject("b", "$a", raw);
 	}
 
 	@Test(expected = InvalidDataAccessApiUsageException.class)
 	public void shouldDetectAliasWithoutInclude() {
-		projection.as("b");
+		project().as("b");
 	}
 
 	@Test(expected = InvalidDataAccessApiUsageException.class)
 	public void shouldDetectDuplicateAlias() {
-		projection.include("a").as("b").as("c");
+		project().include("a").as("b").as("c");
 	}
 
 	@Test
 	@SuppressWarnings("unchecked")
 	public void plus() {
 
-		projection.include("a").plus(10);
-
-		DBObject raw = projection.toDBObject();
+		DBObject raw = safeExtractDbObjectFromProjection(project().include("a").plus(10));
 		assertNotNullDBObject(raw);
 
 		DBObject addition = (DBObject) raw.get("a");
@@ -127,9 +110,7 @@ public class ProjectionTests {
 	@SuppressWarnings("unchecked")
 	public void plusWithAlias() {
 
-		projection.include("a").plus(10).as("b");
-
-		DBObject raw = projection.toDBObject();
+		DBObject raw = safeExtractDbObjectFromProjection(project().include("a").plus(10).as("b"));
 		assertNotNullDBObject(raw);
 
 		DBObject addition = (DBObject) raw.get("b");
@@ -142,6 +123,29 @@ public class ProjectionTests {
 		assertThat((Integer) summands.get(1), is(10));
 	}
 
+	@Test
+	public void projectionWithFields() {
+		ProjectionOperation projectionOperation = project(ZipInfoStats.class) //
+				.field("_id", 0) //
+				.field("state", $id()) // $id() -> $_id
+				.field("biggestCity", fields().and("name", $("biggestCity")).and("population", $("biggestPop"))) //
+				.field("smallestCity", fields().and("name", $("smallestCity")).and("population", $("smallestPop")));
+
+		assertThat(projectionOperation, is(notNullValue()));
+	}
+
+	private static DBObject safeExtractDbObjectFromProjection(ProjectionOperation projectionOperation) {
+
+		assertThat(projectionOperation, is(notNullValue()));
+		DBObject dbObject = projectionOperation.toDbObject();
+		assertNotNullDBObject(dbObject);
+		Object projection = dbObject.get("$project");
+		assertThat("Expected non null value for key $project ", projection, is(notNullValue()));
+		assertTrue("projection contents should be a " + DBObject.class.getSimpleName(), projection instanceof DBObject);
+
+		return DBObject.class.cast(projection);
+	}
+
 	private static void assertSingleDBObject(String key, Object value, DBObject doc) {
 
 		assertNotNullDBObject(doc);

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

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2013 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 org.springframework.data.annotation.Id;
+import org.springframework.data.mongodb.core.mapping.Field;
+
+/**
+ * @author Thomas Darimont
+ */
+class StateStats {
+	@Id String id;
+	String state;
+	@Field("totalPop") int totalPopulation;
+}

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

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2013 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.Arrays;
+import java.util.Date;
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * @author Thomas Darimont
+ */
+public class UserWithLikes {
+
+	String id;
+	Date joined;
+	Set<String> likes = new HashSet<String>();
+
+	public UserWithLikes(String id, Date joined, String... likes) {
+		this.id = id;
+		this.joined = joined;
+		this.likes = new HashSet<String>(Arrays.asList(likes));
+	}
+}

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

@@ -0,0 +1,25 @@
+package org.springframework.data.mongodb.core.aggregation;
+
+import java.util.Arrays;
+
+import org.springframework.data.mongodb.core.mapping.Field;
+
+/**
+ * Data model from mongodb reference data set
+ * 
+ * @see http://docs.mongodb.org/manual/tutorial/aggregation-examples/
+ * @see http://media.mongodb.org/zips.json
+ */
+class ZipInfo {
+
+	String id;
+	String city;
+	String state;
+	@Field("pop") int population;
+	@Field("loc") double[] location;
+
+	public String toString() {
+		return "ZipInfo [id=" + id + ", city=" + city + ", state=" + state + ", population=" + population + ", location="
+				+ Arrays.toString(location) + "]";
+	}
+}

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

@@ -0,0 +1,14 @@
+package org.springframework.data.mongodb.core.aggregation;
+
+class ZipInfoStats {
+
+	String id;
+	String state;
+	City biggestCity;
+	City smallestCity;
+
+	public String toString() {
+		return "ZipInfoStats [id=" + id + ", state=" + state + ", biggestCity=" + biggestCity + ", smallestCity="
+				+ smallestCity + "]";
+	}
+}