Include an AdvancedDatastore-specific method for aggregation

morphia/src/main/java/org/mongodb/morphia/AdvancedDatastore.java

@@ -1,13 +1,15 @@
 package org.mongodb.morphia;
 
+import org.mongodb.morphia.aggregation.AggregationPipeline;
+import org.mongodb.morphia.query.Query;
+import org.mongodb.morphia.query.UpdateOperations;
+
 import com.mongodb.DBDecoderFactory;
 import com.mongodb.DBObject;
 import com.mongodb.DBRef;
 import com.mongodb.ReadPreference;
 import com.mongodb.WriteConcern;
 import com.mongodb.WriteResult;
-import org.mongodb.morphia.query.Query;
-import org.mongodb.morphia.query.UpdateOperations;
 
 /**
  * This interface exposes advanced {@link Datastore} features, like interacting with DBObject and low-level options. It implements matching
@@ -17,6 +19,15 @@
  */
 public interface AdvancedDatastore extends Datastore {
 
+    /**
+     * Returns an {@link AggregationPipeline} bound to the given collection and class.
+     *
+     * @param collection the collection to query
+     * @param clazz The class to create aggregation against
+     * @return the aggregation pipeline
+     */
+    AggregationPipeline createAggregation(String collection, Class<?> clazz);
+
     /**
      * @param <T>        The type of the entity
      * @param collection the collection to query

morphia/src/main/java/org/mongodb/morphia/DatastoreImpl.java

@@ -16,6 +16,7 @@
 import com.mongodb.ReadPreference;
 import com.mongodb.WriteConcern;
 import com.mongodb.WriteResult;
+
 import org.mongodb.morphia.aggregation.AggregationPipeline;
 import org.mongodb.morphia.aggregation.AggregationPipelineImpl;
 import org.mongodb.morphia.annotations.CappedAt;
@@ -138,7 +139,12 @@ public DatastoreImpl copy(final String database) {
      */
     @Override
     public AggregationPipeline createAggregation(final Class source) {
-        return new AggregationPipelineImpl(this, source);
+        return new AggregationPipelineImpl(this, getCollection(source), source);
+    }
+
+    @Override
+    public AggregationPipeline createAggregation(final String collection, final Class<?> clazz) {
+        return new AggregationPipelineImpl(this, db.getCollection(collection), clazz);
     }
 
     @Override

morphia/src/main/java/org/mongodb/morphia/aggregation/AggregationPipelineImpl.java

@@ -37,11 +37,12 @@ public class AggregationPipelineImpl implements AggregationPipeline {
      * Creates an AggregationPipeline
      *
      * @param datastore the datastore to use
+     * @param collection the database collection on which to operate
      * @param source    the source type to aggregate
      */
-    public AggregationPipelineImpl(final DatastoreImpl datastore, final Class source) {
+    public AggregationPipelineImpl(final DatastoreImpl datastore, final DBCollection collection, final Class source) {
         this.datastore = datastore;
-        this.collection = datastore.getCollection(source);
+        this.collection = collection;
         mapper = datastore.getMapper();
         this.source = source;
     }