Support explain for aggregate operations (#623)

Support explain for aggregate using the explain command, 
which started supporting explaining aggregate commands
in MongoDB 3.6.

JAVA-3909

Author: jyemin

driver-core/src/main/com/mongodb/internal/operation/AggregateExplainOperation.java

@@ -18,27 +18,31 @@
 
 import com.mongodb.ExplainVerbosity;
 import com.mongodb.MongoNamespace;
+import com.mongodb.client.model.Collation;
 import com.mongodb.internal.async.AsyncBatchCursor;
 import com.mongodb.internal.async.SingleResultCallback;
-import com.mongodb.client.model.Collation;
 import com.mongodb.internal.binding.AsyncReadBinding;
 import com.mongodb.internal.binding.ReadBinding;
 import com.mongodb.internal.client.model.AggregationLevel;
+import com.mongodb.internal.connection.NoOpSessionContext;
+import com.mongodb.lang.Nullable;
 import org.bson.BsonDocument;
 import org.bson.BsonValue;
 import org.bson.codecs.Decoder;
 
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 
+import static com.mongodb.internal.operation.ExplainHelper.asExplainCommand;
+
 /**
  * An operation that executes an aggregation query.
  *
  * @param <T> the operations result type.
  * @mongodb.driver.manual aggregation/ Aggregation
  * @since 3.0
  */
-public class AggregateOperation<T> implements AsyncReadOperation<AsyncBatchCursor<T>>, ReadOperation<BatchCursor<T>> {
+public class AggregateOperation<T> implements AsyncExplainableReadOperation<AsyncBatchCursor<T>>, ExplainableReadOperation<BatchCursor<T>> {
     private final AggregateOperationImpl<T> wrapped;
     /**
      * Construct a new instance.
@@ -304,29 +308,26 @@ public void executeAsync(final AsyncReadBinding binding, final SingleResultCallb
     /**
      * Gets an operation whose execution explains this operation.
      *
-     * @param explainVerbosity the explain verbosity
+     * @param verbosity the explain verbosity
      * @return a read operation that when executed will explain this operation
      */
-    public ReadOperation<BsonDocument> asExplainableOperation(final ExplainVerbosity explainVerbosity) {
-        return new AggregateExplainOperation(getNamespace(), getPipeline())
-               .allowDiskUse(getAllowDiskUse())
-               .maxTime(getMaxAwaitTime(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS)
-               .hint(wrapped.getHint())
-               .retryReads(getRetryReads());
+    public <R> ReadOperation<R> asExplainableOperation(@Nullable final ExplainVerbosity verbosity, final Decoder<R> resultDecoder) {
+        return new CommandReadOperation<R>(getNamespace().getDatabaseName(),
+                asExplainCommand(wrapped.getCommand(NoOpSessionContext.INSTANCE), verbosity),
+                resultDecoder);
     }
 
     /**
      * Gets an operation whose execution explains this operation.
      *
-     * @param explainVerbosity the explain verbosity
+     * @param verbosity the explain verbosity
      * @return a read operation that when executed will explain this operation
      */
-    public AsyncReadOperation<BsonDocument> asExplainableOperationAsync(final ExplainVerbosity explainVerbosity) {
-        return new AggregateExplainOperation(getNamespace(), getPipeline())
-                .retryReads(getRetryReads())
-                .allowDiskUse(getAllowDiskUse())
-                .maxTime(getMaxAwaitTime(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS)
-                .hint(wrapped.getHint());
+    public <R> AsyncReadOperation<R> asAsyncExplainableOperation(@Nullable final ExplainVerbosity verbosity,
+                                                                 final Decoder<R> resultDecoder) {
+        return new CommandReadOperation<R>(getNamespace().getDatabaseName(),
+                asExplainCommand(wrapped.getCommand(NoOpSessionContext.INSTANCE), verbosity),
+                resultDecoder);
     }
 
 

driver-core/src/main/com/mongodb/internal/operation/AggregateOperation.java

@@ -202,16 +202,16 @@ private CommandCreator getCommandCreator(final SessionContext sessionContext) {
             @Override
             public BsonDocument create(final ServerDescription serverDescription, final ConnectionDescription connectionDescription) {
                 validateReadConcernAndCollation(connectionDescription, sessionContext.getReadConcern(), collation);
-                return getCommand(connectionDescription, sessionContext);
+                return getCommand(sessionContext);
             }
         };
     }
 
-    private BsonDocument getCommand(final ConnectionDescription description, final SessionContext sessionContext) {
+    BsonDocument getCommand(final SessionContext sessionContext) {
         BsonDocument commandDocument = new BsonDocument("aggregate", aggregateTarget.create());
 
         appendReadConcernToCommand(sessionContext, commandDocument);
-        commandDocument.put("pipeline", pipelineCreator.create(description, sessionContext));
+        commandDocument.put("pipeline", pipelineCreator.create());
         if (maxTimeMS > 0) {
             commandDocument.put("maxTimeMS", maxTimeMS > Integer.MAX_VALUE
                     ? new BsonInt64(maxTimeMS) : new BsonInt32((int) maxTimeMS));
@@ -270,7 +270,7 @@ interface AggregateTarget {
     }
 
     interface PipelineCreator {
-        BsonArray create(ConnectionDescription connectionDescription, SessionContext sessionContext);
+        BsonArray create();
     }
 
     private static AggregateTarget defaultAggregateTarget(final AggregationLevel aggregationLevel, final String collectionName) {
@@ -289,7 +289,7 @@ public BsonValue create() {
     private static PipelineCreator defaultPipelineCreator(final List<BsonDocument> pipeline) {
         return new PipelineCreator() {
             @Override
-            public BsonArray create(final ConnectionDescription connectionDescription, final SessionContext sessionContext) {
+            public BsonArray create() {
                 return new BsonArray(pipeline);
             }
         };

driver-core/src/main/com/mongodb/internal/operation/AggregateOperationImpl.java

@@ -16,13 +16,12 @@
 
 package com.mongodb.internal.operation;
 
-import com.mongodb.ExplainVerbosity;
 import com.mongodb.MongoNamespace;
 import com.mongodb.ReadConcern;
 import com.mongodb.WriteConcern;
-import com.mongodb.internal.async.SingleResultCallback;
 import com.mongodb.client.model.Collation;
 import com.mongodb.connection.ConnectionDescription;
+import com.mongodb.internal.async.SingleResultCallback;
 import com.mongodb.internal.binding.AsyncWriteBinding;
 import com.mongodb.internal.binding.WriteBinding;
 import com.mongodb.internal.client.model.AggregationLevel;
@@ -345,19 +344,6 @@ public AggregateToCollectionOperation hint(final BsonDocument hint) {
         return this;
     }
 
-    /**
-     * Gets an operation whose execution explains this operation.
-     *
-     * @param explainVerbosity the explain verbosity
-     * @return a read operation that when executed will explain this operation
-     */
-    public ReadOperation<BsonDocument> asExplainableOperation(final ExplainVerbosity explainVerbosity) {
-        return new AggregateExplainOperation(namespace, pipeline)
-                .allowDiskUse(allowDiskUse)
-                .maxTime(maxTimeMS, TimeUnit.MILLISECONDS)
-                .hint(hint);
-    }
-
     @Override
     public Void execute(final WriteBinding binding) {
         return withConnection(binding, new CallableWithConnection<Void>() {

driver-core/src/main/com/mongodb/internal/operation/AggregateToCollectionOperation.java

@@ -17,20 +17,18 @@
 package com.mongodb.internal.operation;
 
 import com.mongodb.MongoNamespace;
+import com.mongodb.client.model.Collation;
+import com.mongodb.client.model.changestream.FullDocument;
 import com.mongodb.internal.async.AsyncAggregateResponseBatchCursor;
 import com.mongodb.internal.async.AsyncBatchCursor;
 import com.mongodb.internal.async.SingleResultCallback;
-import com.mongodb.client.model.Collation;
-import com.mongodb.client.model.changestream.FullDocument;
-import com.mongodb.connection.ConnectionDescription;
 import com.mongodb.internal.binding.AsyncConnectionSource;
 import com.mongodb.internal.binding.AsyncReadBinding;
 import com.mongodb.internal.binding.ConnectionSource;
 import com.mongodb.internal.binding.ReadBinding;
 import com.mongodb.internal.client.model.changestream.ChangeStreamLevel;
 import com.mongodb.internal.operation.OperationHelper.AsyncCallableWithSource;
 import com.mongodb.internal.operation.OperationHelper.CallableWithSource;
-import com.mongodb.internal.session.SessionContext;
 import org.bson.BsonArray;
 import org.bson.BsonBoolean;
 import org.bson.BsonDocument;
@@ -404,7 +402,7 @@ public BsonValue create() {
     private AggregateOperationImpl.PipelineCreator getPipelineCreator() {
         return new AggregateOperationImpl.PipelineCreator() {
             @Override
-            public BsonArray create(final ConnectionDescription description, final SessionContext sessionContext) {
+            public BsonArray create() {
                 List<BsonDocument> changeStreamPipeline = new ArrayList<BsonDocument>();
                 BsonDocument changeStream = new BsonDocument();
                 if (fullDocument != FullDocument.DEFAULT) {