CSHARP-2634: Add support for $merge.

Author: rstam

src/MongoDB.Driver.Core/Core/Misc/Feature.cs

@@ -35,7 +35,8 @@ public class Feature
         private static readonly Feature __aggregateGraphLookupStage = new Feature("AggregateGraphLookupStage", new SemanticVersion(3, 4, 0, "rc0"));
         private static readonly Feature __aggregateHint = new Feature("AggregateHint", new SemanticVersion(3, 6, 0, "rc0"));
         private static readonly Feature __aggregateLet = new Feature("AggregateLet", new SemanticVersion(3, 6, 0));
-        private static readonly Feature __aggregateOut = new Feature("Aggregate", new SemanticVersion(2, 6, 0));
+        private static readonly Feature __aggregateMerge = new Feature("AggregateMerge", new SemanticVersion(4, 2, 0));
+        private static readonly Feature __aggregateOut = new Feature("AggregateOut", new SemanticVersion(2, 6, 0));
         private static readonly ArrayFiltersFeature __arrayFilters = new ArrayFiltersFeature("ArrayFilters", new SemanticVersion(3, 5, 11));
         private static readonly Feature __bypassDocumentValidation = new Feature("BypassDocumentValidation", new SemanticVersion(3, 2, 0));
         private static readonly Feature __changeStreamStage = new Feature("ChangeStreamStage", new SemanticVersion(3, 5, 11));
@@ -139,6 +140,11 @@ public class Feature
         /// </summary>
         public static Feature AggregateLet => __aggregateLet;
 
+        /// <summary>
+        /// Gets the aggregate merge feature.
+        /// </summary>
+        public static Feature AggregateMerge => __aggregateMerge;
+
         /// <summary>
         /// Gets the aggregate out feature.
         /// </summary>

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

@@ -416,9 +416,9 @@ private AsyncCursor<TResult> CreateCursorFromInlineResult(BsonDocument command,
 
         private void EnsureIsReadOnlyPipeline()
         {
-            if (Pipeline.Any(s => s.GetElement(0).Name == "$out"))
+            if (Pipeline.Any(s => { var n = s.GetElement(0).Name; return n == "$out" || n == "$merge"; }))
             {
-                throw new ArgumentException("The pipeline for an AggregateOperation contains a $out operator. Use AggregateOutputToCollectionOperation instead.", "pipeline");
+                throw new ArgumentException("The pipeline for an AggregateOperation contains a $out or $merge operator. Use AggregateOutputToCollectionOperation instead.", "pipeline");
             }
         }
 

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

@@ -282,9 +282,10 @@ private WriteCommandOperation<BsonDocument> CreateOperation(ICoreSessionHandle s
         private void EnsureIsOutputToCollectionPipeline()
         {
             var lastStage = _pipeline.LastOrDefault();
-            if (lastStage == null || lastStage.GetElement(0).Name != "$out")
+            var lastStageName = lastStage?.GetElement(0).Name;
+            if (lastStage == null || (lastStageName != "$out" && lastStageName != "$merge"))
             {
-                throw new ArgumentException("The last stage of the pipeline for an AggregateOutputToCollectionOperation must have a $out operator.", "pipeline");
+                throw new ArgumentException("The last stage of the pipeline for an AggregateOutputToCollectionOperation must have a $out or $merge operator.", "pipeline");
             }
         }
     }

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

@@ -403,7 +403,7 @@ private BsonDocument CreateChangeStreamStage()
         {
             var changeStreamOptions = new BsonDocument
             {
-                { "fullDocument", ToString(_fullDocument) },
+                { "fullDocument", () => ToString(_fullDocument), _fullDocument != ChangeStreamFullDocumentOption.Default },
                 { "allChangesForCluster", true, _collectionNamespace == null && _databaseNamespace == null },
                 { "startAfter", _startAfter, _startAfter != null},
                 { "startAtOperationTime", _startAtOperationTime, _startAtOperationTime != null },

src/MongoDB.Driver.Legacy/MongoCollection.cs

@@ -144,8 +144,9 @@ private IEnumerable<BsonDocument> Aggregate(IClientSessionHandle session, Aggreg
 
             var messageEncoderSettings = GetMessageEncoderSettings();
 
-            var last = args.Pipeline.LastOrDefault();
-            if (last != null && last.GetElement(0).Name == "$out")
+            var lastStage = args.Pipeline.LastOrDefault();
+            var lastStageName = lastStage?.GetElement(0).Name;
+            if (lastStage != null && (lastStageName == "$out" || lastStageName== "$merge"))
             {
                 var aggregateOperation = new AggregateToCollectionOperation(_collectionNamespace, args.Pipeline, messageEncoderSettings)
                 {
@@ -158,8 +159,39 @@ private IEnumerable<BsonDocument> Aggregate(IClientSessionHandle session, Aggreg
                 };
                 ExecuteWriteOperation(session, aggregateOperation);
 
-                var outputCollectionName = last[0].AsString;
-                var outputCollectionNamespace = new CollectionNamespace(_collectionNamespace.DatabaseNamespace, outputCollectionName);
+                CollectionNamespace outputCollectionNamespace;
+                var stageName = lastStage.GetElement(0).Name;
+                switch (stageName)
+                {
+                    case "$out":
+                        {
+                            var outputCollectionName = lastStage[0].AsString;
+                            outputCollectionNamespace = new CollectionNamespace(_collectionNamespace.DatabaseNamespace, outputCollectionName);
+                        }
+                        break;
+                    case "$merge":
+                        {
+                            var mergeArguments = lastStage[0].AsBsonDocument;
+                            DatabaseNamespace outputDatabaseNamespace;
+                            string outputCollectionName;
+                            var into = mergeArguments["into"];
+                            if (into.IsString)
+                            {
+                                outputDatabaseNamespace = _collectionNamespace.DatabaseNamespace;
+                                outputCollectionName = into.AsString;
+                            }
+                            else
+                            {
+                                outputDatabaseNamespace = new DatabaseNamespace(into["db"].AsString);
+                                outputCollectionName = into["coll"].AsString;
+                            }
+                            outputCollectionNamespace = new CollectionNamespace(outputDatabaseNamespace, outputCollectionName);
+                        }
+                        break;
+                    default:
+                        throw new ArgumentException($"Unexpected stage name: {stageName}.");
+                }
+
                 var resultSerializer = BsonDocumentSerializer.Instance;
                 var findOperation = new FindOperation<BsonDocument>(outputCollectionNamespace, resultSerializer, messageEncoderSettings)
                 {

src/MongoDB.Driver/AggregateFluent.cs

@@ -156,6 +156,22 @@ public override IAggregateFluent<TResult> Match(FilterDefinition<TResult> filter
             return WithPipeline(_pipeline.Match(filter));
         }
 
+        public override IAsyncCursor<TOutput> Merge<TOutput>(IMongoCollection<TOutput> outputCollection, MergeStageOptions<TOutput> mergeOptions = null, CancellationToken cancellationToken = default)
+        {
+            Ensure.IsNotNull(outputCollection, nameof(outputCollection));
+            mergeOptions = mergeOptions ?? new MergeStageOptions<TOutput>();
+            var aggregate = WithPipeline(_pipeline.Merge<TInput, TResult, TOutput>(outputCollection, mergeOptions));
+            return aggregate.ToCursor(cancellationToken);
+        }
+
+        public override async Task<IAsyncCursor<TOutput>> MergeAsync<TOutput>(IMongoCollection<TOutput> outputCollection, MergeStageOptions<TOutput> mergeOptions = null, CancellationToken cancellationToken = default)
+        {
+            Ensure.IsNotNull(outputCollection, nameof(outputCollection));
+            mergeOptions = mergeOptions ?? new MergeStageOptions<TOutput>();
+            var aggregate = WithPipeline(_pipeline.Merge<TInput, TResult, TOutput>(outputCollection, mergeOptions));
+            return await aggregate.ToCursorAsync(cancellationToken).ConfigureAwait(false);
+        }
+
         public override IAggregateFluent<TNewResult> OfType<TNewResult>(IBsonSerializer<TNewResult> newResultSerializer)
         {
             return WithPipeline(_pipeline.OfType(newResultSerializer));

src/MongoDB.Driver/AggregateFluentBase.cs

@@ -146,6 +146,18 @@ public virtual IAggregateFluent<TNewResult> Lookup<TForeignDocument, TAsElement,
         /// <inheritdoc />
         public abstract IAggregateFluent<TResult> Match(FilterDefinition<TResult> filter);
 
+        /// <inheritdoc />
+        public virtual IAsyncCursor<TOutput> Merge<TOutput>(IMongoCollection<TOutput> outputCollection, MergeStageOptions<TOutput> mergeOptions = null, CancellationToken cancellationToken = default)
+        {
+            throw new NotImplementedException();
+        }
+
+        /// <inheritdoc />
+        public virtual Task<IAsyncCursor<TOutput>> MergeAsync<TOutput>(IMongoCollection<TOutput> outputCollection, MergeStageOptions<TOutput> mergeOptions = null, CancellationToken cancellationToken = default)
+        {
+            throw new NotImplementedException();
+        }
+
         /// <inheritdoc />
         public abstract IAggregateFluent<TNewResult> OfType<TNewResult>(IBsonSerializer<TNewResult> newResultSerializer) where TNewResult : TResult;
 

src/MongoDB.Driver/BulkWriteUpsert.cs

@@ -62,6 +62,30 @@ public int Index
             get { return _index; }
         }
 
+        // public methods
+        /// <inheritdoc />
+        public override bool Equals(object obj)
+        {
+            if (object.ReferenceEquals(obj, null) || obj.GetType() != typeof(BulkWriteUpsert))
+            {
+                return false;
+            }
+
+            var other = (BulkWriteUpsert)obj;
+            return
+                _index == other._index &&
+                _id.Equals(other._id);
+        }
+
+        /// <inheritdoc />
+        public override int GetHashCode()
+        {
+            return new Hasher()
+                .Hash(_index)
+                .Hash(_id)
+                .GetHashCode();
+        }
+
         // internal static methods
         internal static BulkWriteUpsert FromCore(Core.Operations.BulkWriteOperationUpsert upsert)
         {

src/MongoDB.Driver/IAggregateFluent.cs

@@ -234,6 +234,24 @@ IAggregateFluent<TNewResult> Lookup<TForeignDocument, TAsElement, TAs, TNewResul
         /// <returns>The fluent aggregate interface.</returns>
         IAggregateFluent<TResult> Match(FilterDefinition<TResult> filter);
 
+        /// <summary>
+        /// Appends a merge stage to the pipeline and executes it, and then returns a cursor to read the contents of the output collection.
+        /// </summary>
+        /// <param name="outputCollection">The output collection.</param>
+        /// <param name="mergeOptions">The merge options.</param>
+        /// <param name="cancellationToken">The cancellation token.</param>
+        /// <returns>A cursor.</returns>
+        IAsyncCursor<TOutput> Merge<TOutput>(IMongoCollection<TOutput> outputCollection, MergeStageOptions<TOutput> mergeOptions = null, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Appends a merge stage to the pipeline and executes it, and then returns a cursor to read the contents of the output collection.
+        /// </summary>
+        /// <param name="outputCollection">The output collection.</param>
+        /// <param name="mergeOptions">The merge options.</param>
+        /// <param name="cancellationToken">The cancellation token.</param>
+        /// <returns>A cursor.</returns>
+        Task<IAsyncCursor<TOutput>> MergeAsync<TOutput>(IMongoCollection<TOutput> outputCollection, MergeStageOptions<TOutput> mergeOptions = null, CancellationToken cancellationToken = default);
+
         /// <summary>
         /// Appends a match stage to the pipeline that matches derived documents and changes the result type to the derived type.
         /// </summary>

src/MongoDB.Driver/MergeStageOptions.cs

@@ -0,0 +1,147 @@
+/* Copyright 2019-present MongoDB Inc.
+*
+* 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.
+*/
+
+using System.Collections.Generic;
+using MongoDB.Bson;
+using MongoDB.Bson.Serialization;
+
+namespace MongoDB.Driver
+{
+    /// <summary>
+    /// The behavior of $merge is a result document and an existing document in the collection
+    /// have the same value for the specified on field(s).
+    /// </summary>
+    public enum MergeStageWhenMatched
+    {
+        /// <summary>
+        /// Replace the existing document in the output collection with the matching results document.
+        /// </summary>
+        Replace,
+
+        /// <summary>
+        /// Keep the existing document in the output collection.
+        /// </summary>
+        KeepExisting,
+
+        /// <summary>
+        /// Merge the matching documents (similar to the $mergeObjects operator).
+        /// </summary>
+        Merge,
+
+        /// <summary>
+        /// Stop and fail the aggregation. Any changes to the output collection from previous documents are not reverted.
+        /// </summary>
+        Fail,
+
+        /// <summary>
+        /// Use an aggregation pipeline to update the document in the collection.
+        /// </summary>
+        Pipeline
+    }
+
+    /// <summary>
+    /// The behavior of $merge if a result document does not match an existing document in the output collection.
+    /// </summary>
+    public enum MergeStageWhenNotMatched
+    {
+        /// <summary>
+        /// Insert the document into the output collection.
+        /// </summary>
+        Insert,
+
+        /// <summary>
+        /// Discard the document; i.e. $merge does not insert the document into the output collection.
+        /// </summary>
+        Discard,
+
+        /// <summary>
+        /// Stop and fail the aggregation operation. Any changes to the output collection from previous documents are not reverted.
+        /// </summary>
+        Fail
+    }
+
+    /// <summary>
+    /// Options for the $merge aggregation pipeline stage.
+    /// </summary>
+    /// <typeparam name="TOutput">The type of the output documents.</typeparam>
+    public class MergeStageOptions<TOutput>
+    {
+        // private fields
+        private BsonDocument _letVariables;
+        private IReadOnlyList<string> _onFieldNames;
+        private IBsonSerializer<TOutput> _outputSerializer;
+        private MergeStageWhenMatched? _whenMatched;
+        private PipelineDefinition<TOutput, TOutput> _whenMatchedPipeline;
+        private MergeStageWhenNotMatched? _whenNotMatched;
+
+        // public properties
+        /// <summary>
+        /// Specifies variables accessible for use in the WhenMatchedPipeline.
+        /// </summary>
+        public BsonDocument LetVariables
+        {
+            get => _letVariables;
+            set => _letVariables = value;
+        }
+
+        /// <summary>
+        /// Field or fields that act as a unique identifier for a document. The identifier determines if a results
+        /// document matches an already existing document in the output collection.
+        /// </summary>
+        public IReadOnlyList<string> OnFieldNames
+        {
+            get => _onFieldNames;
+            set => _onFieldNames = value;
+        }
+
+        /// <summary>
+        /// The output serializer.
+        /// </summary>
+        public IBsonSerializer<TOutput> OutputSerializer
+        {
+            get => _outputSerializer;
+            set => _outputSerializer = value;
+        }
+
+        /// <summary>
+        /// The behavior of $merge if a result document and an existing document in the collectoin have the
+        /// same value for the specified on field(s).
+        /// </summary>
+        public MergeStageWhenMatched? WhenMatched
+        {
+            get => _whenMatched;
+            set => _whenMatched = value;
+        }
+
+        /// <summary>
+        /// An aggregation pipeline to update the document in the collection.
+        /// Used when WhenMatched is Pipeline.
+        /// </summary>
+        public PipelineDefinition<TOutput, TOutput> WhenMatchedPipeline
+        {
+            get => _whenMatchedPipeline;
+            set => _whenMatchedPipeline = value;
+        }
+
+        /// <summary>
+        /// The behavior of $merge if a result document does not match an existing document in the output collection.
+        /// </summary>
+        public MergeStageWhenNotMatched? WhenNotMatched
+        {
+            get => _whenNotMatched;
+            set => _whenNotMatched = value;
+        }
+    }
+}