Added db.aggregate method

JAVA-3116

Author: rozza

driver-async/src/main/com/mongodb/async/client/AggregateIterableImpl.java

@@ -22,6 +22,7 @@
 import com.mongodb.WriteConcern;
 import com.mongodb.async.AsyncBatchCursor;
 import com.mongodb.async.SingleResultCallback;
+import com.mongodb.client.model.AggregationLevel;
 import com.mongodb.client.model.Collation;
 import com.mongodb.client.model.FindOptions;
 import com.mongodb.internal.operation.AsyncOperations;
@@ -46,6 +47,7 @@ class AggregateIterableImpl<TDocument, TResult> extends MongoIterableImpl<TResul
     private final Class<TResult> resultClass;
     private final CodecRegistry codecRegistry;
     private final List<? extends Bson> pipeline;
+    private final AggregationLevel aggregationLevel;
 
     private Boolean allowDiskUse;
     private long maxTimeMS;
@@ -56,17 +58,26 @@ class AggregateIterableImpl<TDocument, TResult> extends MongoIterableImpl<TResul
     private String comment;
     private Bson hint;
 
+    AggregateIterableImpl(@Nullable final ClientSession clientSession, final String databaseName, final Class<TDocument> documentClass,
+                          final Class<TResult> resultClass, final CodecRegistry codecRegistry, final ReadPreference readPreference,
+                          final ReadConcern readConcern, final WriteConcern writeConcern, final OperationExecutor executor,
+                          final List<? extends Bson> pipeline, final AggregationLevel aggregationLevel) {
+        this(clientSession, new MongoNamespace(databaseName, "ignored"), documentClass, resultClass, codecRegistry, readPreference,
+                readConcern, writeConcern, executor, pipeline, aggregationLevel);
+    }
+
     AggregateIterableImpl(@Nullable final ClientSession clientSession, final MongoNamespace namespace, final Class<TDocument> documentClass,
                           final Class<TResult> resultClass, final CodecRegistry codecRegistry, final ReadPreference readPreference,
                           final ReadConcern readConcern, final WriteConcern writeConcern, final OperationExecutor executor,
-                          final List<? extends Bson> pipeline) {
+                          final List<? extends Bson> pipeline, final AggregationLevel aggregationLevel) {
         super(clientSession, executor, readConcern, readPreference);
         this.operations = new AsyncOperations<TDocument>(namespace, documentClass, readPreference, codecRegistry, writeConcern, false);
         this.namespace = notNull("namespace", namespace);
         this.documentClass = notNull("documentClass", documentClass);
         this.resultClass = notNull("resultClass", resultClass);
         this.codecRegistry = notNull("codecRegistry", codecRegistry);
         this.pipeline = notNull("pipeline", pipeline);
+        this.aggregationLevel = notNull("aggregationLevel", aggregationLevel);
     }
 
     @Override
@@ -77,7 +88,7 @@ public void toCollection(final SingleResultCallback<Void> callback) {
         }
 
         getExecutor().execute(operations.aggregateToCollection(pipeline, maxTimeMS, allowDiskUse, bypassDocumentValidation, collation, hint,
-                comment), getReadConcern(), getClientSession(), callback);
+                comment, aggregationLevel), getReadConcern(), getClientSession(), callback);
     }
 
     @Override
@@ -144,7 +155,8 @@ AsyncReadOperation<AsyncBatchCursor<TResult>> asAsyncReadOperation() {
 
         if (outCollection != null) {
             AsyncWriteOperation<Void> aggregateToCollectionOperation =
-                    operations.aggregateToCollection(pipeline, maxTimeMS, allowDiskUse, bypassDocumentValidation, collation, hint, comment);
+                    operations.aggregateToCollection(pipeline, maxTimeMS, allowDiskUse, bypassDocumentValidation, collation, hint, comment,
+                            aggregationLevel);
 
             FindOptions findOptions = new FindOptions().collation(collation);
             Integer batchSize = getBatchSize();
@@ -158,7 +170,7 @@ AsyncReadOperation<AsyncBatchCursor<TResult>> asAsyncReadOperation() {
             return new WriteOperationThenCursorReadOperation<TResult>(aggregateToCollectionOperation, findOperation);
         } else {
             return operations.aggregate(pipeline, resultClass, maxTimeMS, maxAwaitTimeMS, getBatchSize(), collation,
-                    hint, comment, allowDiskUse, useCursor);
+                    hint, comment, allowDiskUse, useCursor, aggregationLevel);
         }
 
     }

driver-async/src/main/com/mongodb/async/client/MongoCollectionImpl.java

@@ -29,6 +29,7 @@
 import com.mongodb.async.SingleResultCallback;
 import com.mongodb.bulk.BulkWriteResult;
 import com.mongodb.bulk.WriteRequest;
+import com.mongodb.client.model.AggregationLevel;
 import com.mongodb.client.model.BulkWriteOptions;
 import com.mongodb.client.model.CountOptions;
 import com.mongodb.internal.client.model.CountStrategy;
@@ -346,7 +347,7 @@ private <TResult> AggregateIterable<TResult> createAggregateIterable(@Nullable f
                                                                          final List<? extends Bson> pipeline,
                                                                          final Class<TResult> resultClass) {
         return new AggregateIterableImpl<TDocument, TResult>(clientSession, namespace, documentClass, resultClass, codecRegistry,
-                readPreference, readConcern, writeConcern, executor, pipeline);
+                readPreference, readConcern, writeConcern, executor, pipeline, AggregationLevel.COLLECTION);
     }
 
     @Override

driver-async/src/main/com/mongodb/async/client/MongoDatabase.java

@@ -502,4 +502,58 @@ void createView(ClientSession clientSession, String viewName, String viewOn, Lis
      */
     <TResult> ChangeStreamIterable<TResult> watch(ClientSession clientSession, List<? extends Bson> pipeline, Class<TResult> resultClass);
 
+    /**
+     * Runs an aggregation framework pipeline on the database for pipeline stages
+     * that do not require an underlying collection, such as {@code $currentOp} and {@code $listLocalSessions}.
+     *
+     * @param pipeline the aggregation pipeline
+     * @return an iterable containing the result of the aggregation operation
+     * @since 3.10
+     * @mongodb.driver.manual reference/command/aggregate/#dbcmd.aggregate Aggregate Command
+     * @mongodb.server.release 3.6
+     */
+    AggregateIterable<Document> aggregate(List<? extends Bson> pipeline);
+
+    /**
+     * Runs an aggregation framework pipeline on the database for pipeline stages
+     * that do not require an underlying collection, such as {@code $currentOp} and {@code $listLocalSessions}.
+     *
+     * @param pipeline    the aggregation pipeline
+     * @param resultClass the class to decode each document into
+     * @param <TResult>   the target document type of the iterable.
+     * @return an iterable containing the result of the aggregation operation
+     * @since 3.10
+     * @mongodb.driver.manual reference/command/aggregate/#dbcmd.aggregate Aggregate Command
+     * @mongodb.server.release 3.6
+     */
+    <TResult> AggregateIterable<TResult> aggregate(List<? extends Bson> pipeline, Class<TResult> resultClass);
+
+    /**
+     * Runs an aggregation framework pipeline on the database for pipeline stages
+     * that do not require an underlying collection, such as {@code $currentOp} and {@code $listLocalSessions}.
+     *
+     * @param clientSession the client session with which to associate this operation
+     * @param pipeline the aggregation pipeline
+     * @return an iterable containing the result of the aggregation operation
+     * @since 3.10
+     * @mongodb.driver.manual reference/command/aggregate/#dbcmd.aggregate Aggregate Command
+     * @mongodb.server.release 3.6
+     */
+    AggregateIterable<Document> aggregate(ClientSession clientSession, List<? extends Bson> pipeline);
+
+    /**
+     * Runs an aggregation framework pipeline on the database for pipeline stages
+     * that do not require an underlying collection, such as {@code $currentOp} and {@code $listLocalSessions}.
+     *
+     * @param clientSession the client session with which to associate this operation
+     * @param pipeline    the aggregation pipeline
+     * @param resultClass the class to decode each document into
+     * @param <TResult>   the target document type of the iterable.
+     * @return an iterable containing the result of the aggregation operation
+     * @since 3.10
+     * @mongodb.driver.manual reference/command/aggregate/#dbcmd.aggregate Aggregate Command
+     * @mongodb.server.release 3.6
+     */
+    <TResult> AggregateIterable<TResult> aggregate(ClientSession clientSession, List<? extends Bson> pipeline, Class<TResult> resultClass);
+
 }

driver-async/src/main/com/mongodb/async/client/MongoDatabaseImpl.java

@@ -23,6 +23,7 @@
 import com.mongodb.ReadPreference;
 import com.mongodb.WriteConcern;
 import com.mongodb.async.SingleResultCallback;
+import com.mongodb.client.model.AggregationLevel;
 import com.mongodb.client.model.CreateCollectionOptions;
 import com.mongodb.client.model.CreateViewOptions;
 import com.mongodb.client.model.IndexOptionDefaults;
@@ -368,6 +369,34 @@ public <TResult> ChangeStreamIterable<TResult> watch(final ClientSession clientS
         return createChangeStreamIterable(clientSession, pipeline, resultClass);
     }
 
+    @Override
+    public AggregateIterable<Document> aggregate(final List<? extends Bson> pipeline) {
+        return aggregate(pipeline, Document.class);
+    }
+
+    @Override
+    public <TResult> AggregateIterable<TResult> aggregate(final List<? extends Bson> pipeline, final Class<TResult> resultClass) {
+        return createAggregateIterable(null, pipeline, resultClass);
+    }
+
+    @Override
+    public AggregateIterable<Document> aggregate(final ClientSession clientSession, final List<? extends Bson> pipeline) {
+        return aggregate(clientSession, pipeline, Document.class);
+    }
+
+    @Override
+    public <TResult> AggregateIterable<TResult> aggregate(final ClientSession clientSession, final List<? extends Bson> pipeline,
+                                                          final Class<TResult> resultClass) {
+        return createAggregateIterable(notNull("clientSession", clientSession), pipeline, resultClass);
+    }
+
+    private <TResult> AggregateIterable<TResult> createAggregateIterable(@Nullable final ClientSession clientSession,
+                                                                         final List<? extends Bson> pipeline,
+                                                                         final Class<TResult> resultClass) {
+        return new AggregateIterableImpl<Document, TResult>(clientSession, name, Document.class, resultClass, codecRegistry,
+                readPreference, readConcern, writeConcern, executor, pipeline, AggregationLevel.DATABASE);
+    }
+
     private <TResult> ChangeStreamIterable<TResult> createChangeStreamIterable(@Nullable final ClientSession clientSession,
                                                                                final List<? extends Bson> pipeline,
                                                                                final Class<TResult> resultClass) {

driver-async/src/test/functional/com/mongodb/async/client/CrudTest.java

@@ -19,6 +19,7 @@
 import com.mongodb.MongoNamespace;
 import org.bson.BsonArray;
 import org.bson.BsonDocument;
+import org.bson.BsonString;
 import org.bson.BsonValue;
 import org.junit.Before;
 import org.junit.Test;
@@ -37,21 +38,25 @@
 import static com.mongodb.ClusterFixture.serverVersionGreaterThan;
 import static com.mongodb.ClusterFixture.serverVersionLessThan;
 import static com.mongodb.async.client.Fixture.getDefaultDatabase;
+import static com.mongodb.async.client.Fixture.getMongoClient;
 import static org.junit.Assert.assertEquals;
 
 // See https://github.com/mongodb/specifications/tree/master/source/crud/tests
 @RunWith(Parameterized.class)
 public class CrudTest extends DatabaseTestCase {
     private final String filename;
     private final String description;
+    private final String databaseName;
     private final BsonArray data;
     private final BsonDocument definition;
     private MongoCollection<BsonDocument> collection;
     private JsonPoweredCrudTestHelper helper;
 
-    public CrudTest(final String filename, final String description, final BsonArray data, final BsonDocument definition) {
+    public CrudTest(final String filename, final String description, final String databaseName,
+                    final BsonArray data, final BsonDocument definition) {
         this.filename = filename;
         this.description = description;
+        this.databaseName = databaseName;
         this.data = data;
         this.definition = definition;
     }
@@ -60,9 +65,9 @@ public CrudTest(final String filename, final String description, final BsonArray
     @Override
     public void setUp() {
         super.setUp();
-        collection = Fixture.initializeCollection(new MongoNamespace(getDefaultDatabaseName(), getClass().getName()))
+        collection = Fixture.initializeCollection(new MongoNamespace(databaseName, getClass().getName()))
                 .withDocumentClass(BsonDocument.class);
-        helper = new JsonPoweredCrudTestHelper(description, getDefaultDatabase(), collection);
+        helper = new JsonPoweredCrudTestHelper(description, getMongoClient().getDatabase(databaseName), collection);
         if (!data.isEmpty()) {
             new MongoOperation<Void>() {
                 @Override
@@ -124,6 +129,7 @@ && serverVersionGreaterThan(testDocument.getString("maxServerVersion").getValue(
             }
             for (BsonValue test : testDocument.getArray("tests")) {
                 data.add(new Object[]{file.getName(), test.asDocument().getString("description").getValue(),
+                        testDocument.getString("database_name", new BsonString(getDefaultDatabaseName())).getValue(),
                         testDocument.getArray("data"), test.asDocument()});
             }
         }

driver-async/src/test/functional/com/mongodb/async/client/JsonPoweredCrudTestHelper.java

@@ -57,6 +57,7 @@
 import org.bson.BsonDocument;
 import org.bson.BsonInt32;
 import org.bson.BsonNull;
+import org.bson.BsonString;
 import org.bson.BsonValue;
 import org.junit.AssumptionViolatedException;
 
@@ -66,7 +67,9 @@
 import java.util.Collections;
 import java.util.List;
 
+import static com.mongodb.async.client.Fixture.isSharded;
 import static java.lang.String.format;
+import static org.junit.Assume.assumeTrue;
 
 public class JsonPoweredCrudTestHelper {
     private final String description;
@@ -85,11 +88,11 @@ BsonDocument getOperationResults(final BsonDocument operation) {
     }
 
     BsonDocument getOperationResults(final BsonDocument operation, @Nullable final ClientSession clientSession) {
-        String name = operation.getString("name").getValue();
         BsonDocument collectionOptions = operation.getDocument("collectionOptions", new BsonDocument());
         BsonDocument arguments = operation.getDocument("arguments");
 
-        String methodName = "get" + name.substring(0, 1).toUpperCase() + name.substring(1) + "Result";
+        String methodName = createMethodName(operation.getString("name").getValue(),
+                operation.getString("object", new BsonString("")).getValue());
         try {
             Method method = getClass().getDeclaredMethod(methodName, BsonDocument.class, BsonDocument.class, ClientSession.class);
             return (BsonDocument) method.invoke(this, collectionOptions, arguments, clientSession);
@@ -108,6 +111,19 @@ BsonDocument getOperationResults(final BsonDocument operation, @Nullable final C
         }
     }
 
+    private String createMethodName(final String name, final String object) {
+        StringBuilder builder = new StringBuilder();
+        builder.append("get");
+        if (!object.isEmpty() && !object.equals("collection")) {
+            builder.append(object.substring(0, 1).toUpperCase());
+            builder.append(object.substring(1));
+        }
+        builder.append(name.substring(0, 1).toUpperCase());
+        builder.append(name.substring(1));
+        builder.append("Result");
+        return builder.toString();
+    }
+
     <T> T futureResult(final FutureResultCallback<T> callback) {
         try {
             return callback.get();
@@ -195,6 +211,11 @@ BsonDocument toResult(@Nullable final BsonValue results) {
         return new BsonDocument("result", results != null ? results : BsonNull.VALUE);
     }
 
+    BsonDocument getDatabaseRunCommandResult(final BsonDocument collectionOptions, final BsonDocument arguments,
+                                             @Nullable final ClientSession clientSession) {
+        return getRunCommandResult(collectionOptions, arguments, clientSession);
+    }
+
     BsonDocument getRunCommandResult(final BsonDocument collectionOptions, final BsonDocument arguments,
                                      @Nullable final ClientSession clientSession) {
         BsonDocument command = arguments.getDocument("command");
@@ -247,6 +268,44 @@ BsonDocument getAggregateResult(final BsonDocument collectionOptions, final Bson
         return toResult(iterable);
     }
 
+    BsonDocument getDatabaseAggregateResult(final BsonDocument collectionOptions, final BsonDocument arguments,
+                                            @Nullable final ClientSession clientSession) {
+        assumeTrue(!isSharded());
+        List<BsonDocument> pipeline = new ArrayList<BsonDocument>();
+        for (BsonValue stage : arguments.getArray("pipeline")) {
+            pipeline.add(stage.asDocument());
+        }
+
+        AggregateIterable<BsonDocument> iterable;
+        if (clientSession == null) {
+            iterable = database.aggregate(pipeline, BsonDocument.class);
+        } else {
+            iterable = database.aggregate(clientSession, pipeline, BsonDocument.class);
+        }
+
+        if (arguments.containsKey("allowDiskUse")) {
+            iterable.allowDiskUse(arguments.getBoolean("allowDiskUse").getValue());
+        }
+        if (arguments.containsKey("batchSize")) {
+            iterable.batchSize(arguments.getNumber("batchSize").intValue());
+        }
+        if (arguments.containsKey("collation")) {
+            iterable.collation(getCollation(arguments.getDocument("collation")));
+        }
+
+        BsonDocument results = toResult(iterable);
+        for (BsonValue result : results.getArray("result", new BsonArray())) {
+            if (result.isDocument()) {
+                BsonDocument command = result.asDocument().getDocument("command", new BsonDocument());
+                command.remove("$readPreference");
+                command.remove("$clusterTime");
+                command.remove("signature");
+                command.remove("keyId");
+            }
+        }
+        return results;
+    }
+
     @SuppressWarnings("deprecation")
     BsonDocument getCountResult(final BsonDocument collectionOptions, final BsonDocument arguments,
                                 @Nullable final ClientSession clientSession) {