CSHARP-2448: Implement MongoDatabase Aggregate method.

Author: rstam

src/MongoDB.Driver.Core/Core/Operations/AggregateToCollectionOperation.cs

@@ -39,6 +39,7 @@ public class AggregateToCollectionOperation : IWriteOperation<BsonDocument>
         private Collation _collation;
         private readonly CollectionNamespace _collectionNamespace;
         private string _comment;
+        private readonly DatabaseNamespace _databaseNamespace;
         private BsonValue _hint;
         private TimeSpan? _maxTime;
         private readonly MessageEncoderSettings _messageEncoderSettings;
@@ -50,18 +51,30 @@ public class AggregateToCollectionOperation : IWriteOperation<BsonDocument>
         /// <summary>
         /// Initializes a new instance of the <see cref="AggregateToCollectionOperation"/> class.
         /// </summary>
-        /// <param name="collectionNamespace">The collection namespace.</param>
+        /// <param name="databaseNamespace">The database namespace.</param>
         /// <param name="pipeline">The pipeline.</param>
         /// <param name="messageEncoderSettings">The message encoder settings.</param>
-        public AggregateToCollectionOperation(CollectionNamespace collectionNamespace, IEnumerable<BsonDocument> pipeline, MessageEncoderSettings messageEncoderSettings)
+        public AggregateToCollectionOperation(DatabaseNamespace databaseNamespace, IEnumerable<BsonDocument> pipeline, MessageEncoderSettings messageEncoderSettings)
         {
-            _collectionNamespace = Ensure.IsNotNull(collectionNamespace, nameof(collectionNamespace));
+            _databaseNamespace = Ensure.IsNotNull(databaseNamespace, nameof(databaseNamespace));
             _pipeline = Ensure.IsNotNull(pipeline, nameof(pipeline)).ToList();
             _messageEncoderSettings = Ensure.IsNotNull(messageEncoderSettings, nameof(messageEncoderSettings));
 
             EnsureIsOutputToCollectionPipeline();
         }
 
+        /// <summary>
+        /// Initializes a new instance of the <see cref="AggregateToCollectionOperation"/> class.
+        /// </summary>
+        /// <param name="collectionNamespace">The collection namespace.</param>
+        /// <param name="pipeline">The pipeline.</param>
+        /// <param name="messageEncoderSettings">The message encoder settings.</param>
+        public AggregateToCollectionOperation(CollectionNamespace collectionNamespace, IEnumerable<BsonDocument> pipeline, MessageEncoderSettings messageEncoderSettings)
+            : this(Ensure.IsNotNull(collectionNamespace, nameof(collectionNamespace)).DatabaseNamespace, pipeline, messageEncoderSettings)
+        {
+            _collectionNamespace = collectionNamespace;
+        }
+
         // properties
         /// <summary>
         /// Gets or sets a value indicating whether the server is allowed to use the disk.
@@ -122,6 +135,17 @@ public string Comment
             set { _comment = value; }
         }
 
+        /// <summary>
+        /// Gets the database namespace.
+        /// </summary>
+        /// <value>
+        /// The database namespace.
+        /// </value>
+        public DatabaseNamespace DatabaseNamespace
+        {
+            get { return _databaseNamespace; }
+        }
+
         /// <summary>
         /// Gets or sets the hint. This must either be a BsonString representing the index name or a BsonDocument representing the key pattern of the index.
         /// </summary>
@@ -235,7 +259,7 @@ internal BsonDocument CreateCommand(ICoreSessionHandle session, ConnectionDescri
             var writeConcern = WriteConcernHelper.GetWriteConcernForCommandThatWrites(session, _writeConcern, serverVersion);
             return new BsonDocument
             {
-                { "aggregate", _collectionNamespace.CollectionName },
+                { "aggregate", _collectionNamespace == null ? (BsonValue)1 : _collectionNamespace.CollectionName },
                 { "pipeline", new BsonArray(_pipeline) },
                 { "allowDiskUse", () => _allowDiskUse.Value, _allowDiskUse.HasValue },
                 { "bypassDocumentValidation", () => _bypassDocumentValidation.Value, _bypassDocumentValidation.HasValue && Feature.BypassDocumentValidation.IsSupported(serverVersion) },

src/MongoDB.Driver/AggregateFluent.cs

@@ -23,29 +23,22 @@
 
 namespace MongoDB.Driver
 {
-    internal class AggregateFluent<TDocument, TResult> : AggregateFluentBase<TResult>
+    internal abstract class AggregateFluent<TInput, TResult> : AggregateFluentBase<TResult>
     {
         // fields
-        private readonly IMongoCollection<TDocument> _collection;
-        private readonly AggregateOptions _options;
-        private readonly PipelineDefinition<TDocument, TResult> _pipeline;
-        private readonly IClientSessionHandle _session;
+        protected readonly AggregateOptions _options;
+        protected readonly PipelineDefinition<TInput, TResult> _pipeline;
+        protected readonly IClientSessionHandle _session;
 
         // constructors
-        public AggregateFluent(IClientSessionHandle session, IMongoCollection<TDocument> collection, PipelineDefinition<TDocument, TResult> pipeline, AggregateOptions options)
+        protected AggregateFluent(IClientSessionHandle session, PipelineDefinition<TInput, TResult> pipeline, AggregateOptions options)
         {
             _session = session; // can be null
-            _collection = Ensure.IsNotNull(collection, nameof(collection));
             _pipeline = Ensure.IsNotNull(pipeline, nameof(pipeline));
             _options = Ensure.IsNotNull(options, nameof(options));
         }
 
         // properties
-        public override IMongoDatabase Database
-        {
-            get { return _collection.Database; }
-        }
-
         public override AggregateOptions Options
         {
             get { return _options; }
@@ -143,7 +136,7 @@ public override IAggregateFluent<TResult> Limit(int limit)
         public override IAggregateFluent<TNewResult> Lookup<TForeignDocument, TNewResult>(string foreignCollectionName, FieldDefinition<TResult> localField, FieldDefinition<TForeignDocument> foreignField, FieldDefinition<TNewResult> @as, AggregateLookupOptions<TForeignDocument, TNewResult> options)
         {
             Ensure.IsNotNull(foreignCollectionName, nameof(foreignCollectionName));
-            var foreignCollection = _collection.Database.GetCollection<TForeignDocument>(foreignCollectionName);
+            var foreignCollection = Database.GetCollection<TForeignDocument>(foreignCollectionName);
             return WithPipeline(_pipeline.Lookup(foreignCollection, localField, foreignField, @as, options));
         }
 
@@ -218,7 +211,7 @@ public override IOrderedAggregateFluent<TResult> ThenBy(SortDefinition<TResult>
             var combinedSort = Builders<TResult>.Sort.Combine(oldSort, newSort);
             var combinedSortStage = PipelineStageDefinitionBuilder.Sort(combinedSort);
             stages[stages.Count - 1] = combinedSortStage;
-            var newPipeline = new PipelineStagePipelineDefinition<TDocument, TResult>(stages);
+            var newPipeline = new PipelineStagePipelineDefinition<TInput, TResult>(stages);
             return (IOrderedAggregateFluent<TResult>)WithPipeline(newPipeline);
         }
 
@@ -232,6 +225,34 @@ public override IAggregateFluent<TNewResult> Unwind<TNewResult>(FieldDefinition<
             return WithPipeline(_pipeline.Unwind(field, options));
         }
 
+        public override string ToString()
+        {
+            return $"aggregate({_pipeline})";
+        }
+
+        protected abstract IAggregateFluent<TNewResult> WithPipeline<TNewResult>(PipelineDefinition<TInput, TNewResult> pipeline);
+    }
+
+    internal class CollectionAggregateFluent<TDocument, TResult> : AggregateFluent<TDocument, TResult>
+    {
+        // private fields
+        private readonly IMongoCollection<TDocument> _collection;
+
+        // constructors
+        public CollectionAggregateFluent(
+            IClientSessionHandle session,
+            IMongoCollection<TDocument> collection,
+            PipelineDefinition<TDocument, TResult> pipeline,
+            AggregateOptions options)
+            : base(session, pipeline, options)
+        {
+            _collection = Ensure.IsNotNull(collection, nameof(collection));
+        }
+
+        // public properties
+        public override IMongoDatabase Database => _collection.Database;
+
+        // public methods
         public override IAsyncCursor<TResult> ToCursor(CancellationToken cancellationToken)
         {
             if (_session == null)
@@ -256,14 +277,61 @@ public override Task<IAsyncCursor<TResult>> ToCursorAsync(CancellationToken canc
             }
         }
 
-        public override string ToString()
+        // protected methods
+        protected override IAggregateFluent<TNewResult> WithPipeline<TNewResult>(PipelineDefinition<TDocument, TNewResult> pipeline)
         {
-            return $"aggregate({_pipeline})";
+            return new CollectionAggregateFluent<TDocument, TNewResult>(_session, _collection, pipeline, _options);
+        }
+    }
+
+    internal class DatabaseAggregateFluent<TResult> : AggregateFluent<NoPipelineInput, TResult>
+    {
+        // private fields
+        private readonly IMongoDatabase _database;
+
+        // constructors
+        public DatabaseAggregateFluent(
+            IClientSessionHandle session,
+            IMongoDatabase database,
+            PipelineDefinition<NoPipelineInput, TResult> pipeline,
+            AggregateOptions options)
+            : base(session, pipeline, options)
+        {
+            _database = Ensure.IsNotNull(database, nameof(database));
+        }
+
+        // public properties
+        public override IMongoDatabase Database => _database;
+
+        // public methods
+        public override IAsyncCursor<TResult> ToCursor(CancellationToken cancellationToken)
+        {
+            if (_session == null)
+            {
+                return _database.Aggregate(_pipeline, _options, cancellationToken);
+            }
+            else
+            {
+                return _database.Aggregate(_session, _pipeline, _options, cancellationToken);
+            }
+        }
+
+        public override Task<IAsyncCursor<TResult>> ToCursorAsync(CancellationToken cancellationToken)
+        {
+            if (_session == null)
+            {
+                return _database.AggregateAsync(_pipeline, _options, cancellationToken);
+            }
+            else
+            {
+                return _database.AggregateAsync(_session, _pipeline, _options, cancellationToken);
+            }
         }
 
-        public IAggregateFluent<TNewResult> WithPipeline<TNewResult>(PipelineDefinition<TDocument, TNewResult> pipeline)
+        // protected methods
+        protected override IAggregateFluent<TNewResult> WithPipeline<TNewResult>(PipelineDefinition<NoPipelineInput, TNewResult> pipeline)
         {
-            return new AggregateFluent<TDocument, TNewResult>(_session, _collection, pipeline, _options);
+            return new DatabaseAggregateFluent<TNewResult>(_session, _database, pipeline, _options);
         }
     }
 }

src/MongoDB.Driver/IMongoCollectionExtensions.cs

@@ -41,7 +41,7 @@ public static class IMongoCollectionExtensions
         public static IAggregateFluent<TDocument> Aggregate<TDocument>(this IMongoCollection<TDocument> collection, AggregateOptions options = null)
         {
             var emptyPipeline = new EmptyPipelineDefinition<TDocument>(collection.DocumentSerializer);
-            return new AggregateFluent<TDocument, TDocument>(null, collection, emptyPipeline, options ?? new AggregateOptions());
+            return new CollectionAggregateFluent<TDocument, TDocument>(null, collection, emptyPipeline, options ?? new AggregateOptions());
         }
 
         /// <summary>
@@ -58,7 +58,7 @@ public static IAggregateFluent<TDocument> Aggregate<TDocument>(this IMongoCollec
         {
             Ensure.IsNotNull(session, nameof(session));
             var emptyPipeline = new EmptyPipelineDefinition<TDocument>(collection.DocumentSerializer);
-            return new AggregateFluent<TDocument, TDocument>(session, collection, emptyPipeline, options ?? new AggregateOptions());
+            return new CollectionAggregateFluent<TDocument, TDocument>(session, collection, emptyPipeline, options ?? new AggregateOptions());
         }
 
         /// <summary>

src/MongoDB.Driver/IMongoDatabase.cs

@@ -46,6 +46,52 @@ public interface IMongoDatabase
         /// </summary>
         MongoDatabaseSettings Settings { get; }
 
+        /// <summary>
+        /// Runs an aggregation pipeline.
+        /// </summary>
+        /// <typeparam name="TResult">The type of the result.</typeparam>
+        /// <param name="pipeline">The pipeline.</param>
+        /// <param name="options">The options.</param>
+        /// <param name="cancellationToken">The cancellation token.</param>
+        /// <returns>A cursor.</returns>
+        IAsyncCursor<TResult> Aggregate<TResult>(PipelineDefinition<NoPipelineInput, TResult> pipeline, AggregateOptions options = null, CancellationToken cancellationToken = default(CancellationToken));
+
+        /// <summary>
+        /// Runs an aggregation pipeline.
+        /// </summary>
+        /// <typeparam name="TResult">The type of the result.</typeparam>
+        /// <param name="session">The session.</param>
+        /// <param name="pipeline">The pipeline.</param>
+        /// <param name="options">The options.</param>
+        /// <param name="cancellationToken">The cancellation token.</param>
+        /// <returns>
+        /// A cursor.
+        /// </returns>
+        IAsyncCursor<TResult> Aggregate<TResult>(IClientSessionHandle session, PipelineDefinition<NoPipelineInput, TResult> pipeline, AggregateOptions options = null, CancellationToken cancellationToken = default(CancellationToken));
+
+        /// <summary>
+        /// Runs an aggregation pipeline.
+        /// </summary>
+        /// <typeparam name="TResult">The type of the result.</typeparam>
+        /// <param name="pipeline">The pipeline.</param>
+        /// <param name="options">The options.</param>
+        /// <param name="cancellationToken">The cancellation token.</param>
+        /// <returns>A Task whose result is a cursor.</returns>
+        Task<IAsyncCursor<TResult>> AggregateAsync<TResult>(PipelineDefinition<NoPipelineInput, TResult> pipeline, AggregateOptions options = null, CancellationToken cancellationToken = default(CancellationToken));
+
+        /// <summary>
+        /// Runs an aggregation pipeline.
+        /// </summary>
+        /// <typeparam name="TResult">The type of the result.</typeparam>
+        /// <param name="session">The session.</param>
+        /// <param name="pipeline">The pipeline.</param>
+        /// <param name="options">The options.</param>
+        /// <param name="cancellationToken">The cancellation token.</param>
+        /// <returns>
+        /// A Task whose result is a cursor.
+        /// </returns>
+        Task<IAsyncCursor<TResult>> AggregateAsync<TResult>(IClientSessionHandle session, PipelineDefinition<NoPipelineInput, TResult> pipeline, AggregateOptions options = null, CancellationToken cancellationToken = default(CancellationToken));
+
         /// <summary>
         /// Creates the collection with the specified name.
         /// </summary>

src/MongoDB.Driver/IMongoDatabaseExtensions.cs

@@ -26,6 +26,36 @@ namespace MongoDB.Driver
     /// </summary>
     public static class IMongoDatabaseExtensions
     {
+        /// <summary>
+        /// Begins a fluent aggregation interface.
+        /// </summary>
+        /// <param name="database">The database.</param>
+        /// <param name="options">The options.</param>
+        /// <returns>
+        /// A fluent aggregate interface.
+        /// </returns>
+        public static IAggregateFluent<NoPipelineInput> Aggregate(this IMongoDatabase database, AggregateOptions options = null)
+        {
+            var emptyPipeline = new EmptyPipelineDefinition<NoPipelineInput>(NoPipelineInputSerializer.Instance);
+            return new DatabaseAggregateFluent<NoPipelineInput>(null, database, emptyPipeline, options ?? new AggregateOptions());
+        }
+
+        /// <summary>
+        /// Begins a fluent aggregation interface.
+        /// </summary>
+        /// <param name="database">The database.</param>
+        /// <param name="session">The session.</param>
+        /// <param name="options">The options.</param>
+        /// <returns>
+        /// A fluent aggregate interface.
+        /// </returns>
+        public static IAggregateFluent<NoPipelineInput> Aggregate(this IMongoDatabase database, IClientSessionHandle session, AggregateOptions options = null)
+        {
+            Ensure.IsNotNull(session, nameof(session));
+            var emptyPipeline = new EmptyPipelineDefinition<NoPipelineInput>(NoPipelineInputSerializer.Instance);
+            return new DatabaseAggregateFluent<NoPipelineInput>(session, database, emptyPipeline, options ?? new AggregateOptions());
+        }
+
         /// <summary>
         /// Watches changes on all collection in a database.
         /// </summary>

src/MongoDB.Driver/MongoDatabaseBase.cs

@@ -37,6 +37,30 @@ public abstract class MongoDatabaseBase : IMongoDatabase
         public abstract MongoDatabaseSettings Settings { get; }
 
         // public methods
+        /// <inheritdoc />
+        public virtual IAsyncCursor<TResult> Aggregate<TResult>(PipelineDefinition<NoPipelineInput, TResult> pipeline, AggregateOptions options = null, CancellationToken cancellationToken = default(CancellationToken))
+        {
+            throw new NotImplementedException();
+        }
+
+        /// <inheritdoc />
+        public virtual IAsyncCursor<TResult> Aggregate<TResult>(IClientSessionHandle session, PipelineDefinition<NoPipelineInput, TResult> pipeline, AggregateOptions options = null, CancellationToken cancellationToken = default(CancellationToken))
+        {
+            throw new NotImplementedException();
+        }
+
+        /// <inheritdoc />
+        public virtual Task<IAsyncCursor<TResult>> AggregateAsync<TResult>(PipelineDefinition<NoPipelineInput, TResult> pipeline, AggregateOptions options = null, CancellationToken cancellationToken = default(CancellationToken))
+        {
+            throw new NotImplementedException();
+        }
+
+        /// <inheritdoc />
+        public virtual Task<IAsyncCursor<TResult>> AggregateAsync<TResult>(IClientSessionHandle session, PipelineDefinition<NoPipelineInput, TResult> pipeline, AggregateOptions options = null, CancellationToken cancellationToken = default(CancellationToken))
+        {
+            throw new NotImplementedException();
+        }
+
         /// <inheritdoc />
         public virtual void CreateCollection(string name, CreateCollectionOptions options = null, CancellationToken cancellationToken = default(CancellationToken))
         {