CSHARP-1649: Added support for $graphLookup aggregation stage.

Author: rstam

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

@@ -30,6 +30,7 @@ public class Feature
         private static readonly Feature __aggregateCursorResult = new Feature("AggregateCursorResult", new SemanticVersion(2, 6, 0));
         private static readonly Feature __aggregateExplain = new Feature("AggregateExplain", new SemanticVersion(2, 6, 0));
         private static readonly Feature __aggregateFacetStage = new Feature("AggregateFacetStage", new SemanticVersion(3, 4, 0, "rc0"));
+        private static readonly Feature __aggregateGraphLookupStage = new Feature("AggregateGraphLookupStage", new SemanticVersion(3, 4, 0, "rc0"));
         private static readonly Feature __aggregateOut = new Feature("Aggregate", new SemanticVersion(2, 6, 0));
         private static readonly Feature __bypassDocumentValidation = new Feature("BypassDocumentValidation", new SemanticVersion(3, 2, 0));
         private static readonly CollationFeature __collation = new CollationFeature("Collation", new SemanticVersion(3, 3, 11));
@@ -88,6 +89,11 @@ public class Feature
         /// </summary>
         public static Feature AggregateFacetStage => __aggregateFacetStage;
 
+        /// <summary>
+        /// Gets the aggregate $graphLookup stage feature.
+        /// </summary>
+        public static Feature AggregateGraphLookupStage => __aggregateGraphLookupStage;
+
         /// <summary>
         /// Gets the aggregate out feature.
         /// </summary>

src/MongoDB.Driver/AggregateFluent.cs

@@ -16,6 +16,7 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Reflection;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
@@ -255,6 +256,60 @@ public override IAggregateFluent<TNewResult> Facet<TNewResult>(
             return AppendStage<TNewResult>(stage);
         }
 
+        public override IAggregateFluent<TNewResult> GraphLookup<TNewResult, TFrom, TConnect, TConnectFrom, TStartWith, TAs, TAsEnumerable>(
+            IMongoCollection<TFrom> from,
+            FieldDefinition<TFrom, TConnectFrom> connectFromField,
+            FieldDefinition<TFrom, TConnect> connectToField,
+            AggregateExpressionDefinition<TResult, TStartWith> startWith,
+            FieldDefinition<TNewResult, TAsEnumerable> @as,
+            FieldDefinition<TAs, int> depthField,
+            AggregateGraphLookupOptions<TNewResult, TFrom, TConnect, TConnectFrom, TStartWith, TAs, TAsEnumerable> options = null)
+        {
+            Ensure.IsNotNull(from, nameof(from));
+            Ensure.IsNotNull(connectFromField, nameof(connectFromField));
+            Ensure.IsNotNull(connectToField, nameof(connectToField));
+            Ensure.IsNotNull(startWith, nameof(startWith));
+            Ensure.IsNotNull(@as, nameof(@as));
+            Ensure.That(from.Database.DatabaseNamespace.Equals(_collection.Database.DatabaseNamespace), "From collection must be from the same database.", nameof(from));
+            Ensure.That(IsTConnectOrEnumerableTConnect<TConnectFrom, TConnect>(), "TConnectFrom must be either TConnect or a type that implements IEnumerable<TConnect>.", nameof(TConnectFrom));
+            Ensure.That(IsTConnectOrEnumerableTConnect<TStartWith, TConnect>(), "TStartWith must be either TConnect or a type that implements IEnumerable<TConnect>.", nameof(TStartWith));
+
+            const string operatorName = "$graphLookup";
+            var stage = new DelegatedPipelineStageDefinition<TResult, TNewResult>(
+                operatorName,
+                (s, sr) =>
+                {
+                    var resultSerializer = s;
+                    var newResultSerializer = options?.NewResultSerializer ?? sr.GetSerializer<TNewResult>();
+                    var fromSerializer = options?.FromSerializer ?? sr.GetSerializer<TFrom>();
+                    var asSerializer = options?.AsSerializer ?? sr.GetSerializer<TAs>();
+                    var renderedConnectToField = connectToField.Render(fromSerializer, sr);
+                    var renderedStartWith = startWith.Render(resultSerializer, sr);
+                    var renderedConnectFromField = connectFromField.Render(fromSerializer, sr);
+                    var renderedAs = @as.Render(newResultSerializer, sr);
+                    var renderedDepthField = depthField?.Render(asSerializer, sr);
+                    var renderedRestrictSearchWithMatch = options?.RestrictSearchWithMatch?.Render(fromSerializer, sr);
+                    var document = new BsonDocument
+                    {
+                        { operatorName, new BsonDocument
+                            {
+                                { "from", from.CollectionNamespace.CollectionName },
+                                { "connectFromField", renderedConnectFromField.FieldName },
+                                { "connectToField", renderedConnectToField.FieldName },
+                                { "startWith", renderedStartWith },
+                                { "as", renderedAs.FieldName },
+                                { "depthField", () => renderedDepthField.FieldName, renderedDepthField != null },
+                                { "maxDepth", () => options.MaxDepth.Value, options != null && options.MaxDepth.HasValue },
+                                { "restrictSearchWithMatch", renderedRestrictSearchWithMatch, renderedRestrictSearchWithMatch != null }
+                            }
+                        }
+                    };
+                    return new RenderedPipelineStageDefinition<TNewResult>(operatorName, document, newResultSerializer);
+                });
+
+            return AppendStage<TNewResult>(stage);
+        }
+
         public override IAggregateFluent<TNewResult> Group<TNewResult>(ProjectionDefinition<TResult, TNewResult> group)
         {
             const string operatorName = "$group";
@@ -479,5 +534,22 @@ public override string ToString()
             sb.Append("])");
             return sb.ToString();
         }
+
+        // private methods
+        private bool IsTConnectOrEnumerableTConnect<TConnectFrom, TConnect>()
+        {
+            if (typeof(TConnect) == typeof(TConnectFrom))
+            {
+                return true;
+            }
+
+            var ienumerableTConnect = typeof(IEnumerable<>).MakeGenericType(typeof(TConnect));
+            if (typeof(TConnectFrom).GetTypeInfo().GetInterfaces().Contains(ienumerableTConnect))
+            {
+                return true;
+            }
+
+            return false;
+        }
     }
 }

src/MongoDB.Driver/AggregateFluentBase.cs

@@ -92,6 +92,20 @@ public virtual IAggregateFluent<TNewResult> Facet<TNewResult>(
             throw new NotImplementedException();
         }
 
+        /// <inheritdoc />
+        public virtual IAggregateFluent<TNewResult> GraphLookup<TNewResult, TFrom, TConnect, TConnectFrom, TStartWith, TAs, TAsEnumerable>(
+            IMongoCollection<TFrom> from,
+            FieldDefinition<TFrom, TConnectFrom> connectFromField,
+            FieldDefinition<TFrom, TConnect> connectToField,
+            AggregateExpressionDefinition<TResult, TStartWith> startWith,
+            FieldDefinition<TNewResult, TAsEnumerable> @as,
+            FieldDefinition<TAs, int> depthField,
+            AggregateGraphLookupOptions<TNewResult, TFrom, TConnect, TConnectFrom, TStartWith, TAs, TAsEnumerable> options = null)
+                where TAsEnumerable : IEnumerable<TAs>
+        {
+            throw new NotImplementedException();
+        }
+
         /// <inheritdoc />
         public abstract IAggregateFluent<TNewResult> Group<TNewResult>(ProjectionDefinition<TResult, TNewResult> group);
 

src/MongoDB.Driver/AggregateGraphLookupOptions.cs

@@ -0,0 +1,56 @@
+/* Copyright 2016 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.Serialization;
+
+namespace MongoDB.Driver
+{
+    /// <summary>
+    /// Represents options for the GraphLookup method.
+    /// </summary>
+    public class AggregateGraphLookupOptions<TNewResult, TFrom, TConnect, TConnectFrom, TStartWith, TAs, TAsEnumerable>
+        where TAsEnumerable : IEnumerable<TAs>
+    {
+        internal AggregateGraphLookupOptions()
+        {
+        }
+
+        /// <summary>
+        /// Gets or sets the TAs serialzier.
+        /// </summary>
+        public IBsonSerializer<TAs> AsSerializer { get; set; }
+
+        /// <summary>
+        /// Gets or sets the TFrom serializer.
+        /// </summary>
+        public IBsonSerializer<TFrom> FromSerializer { get; set; }
+
+        /// <summary>
+        /// Gets or sets the maximum depth.
+        /// </summary>
+        public int? MaxDepth { get; set; }
+
+        /// <summary>
+        /// Gets or sets the TNewResult serializer.
+        /// </summary>
+        public IBsonSerializer<TNewResult> NewResultSerializer { get; set; }
+
+        /// <summary>
+        /// Gets the filter to restrict the search with.
+        /// </summary>
+        public FilterDefinition<TFrom> RestrictSearchWithMatch { get; set; }
+    }
+}

src/MongoDB.Driver/IAggregateFluent.cs

@@ -135,6 +135,34 @@ IAggregateFluent<TNewResult> Facet<TNewResult>(
             IEnumerable<AggregateFacet<TResult>> facets,
             AggregateFacetOptions<TNewResult> options = null);
 
+        /// <summary>
+        /// Appends a $graphLookup stage to the pipeline.
+        /// </summary>
+        /// <typeparam name="TNewResult">The type of the new result (must be same as TResult with an additional as field).</typeparam>
+        /// <typeparam name="TFrom">The type of the from documents.</typeparam>
+        /// <typeparam name="TConnect">The type of the connect field.</typeparam>
+        /// <typeparam name="TConnectFrom">The type of the connect from field (must be either TConnect or a type that implements IEnumerable{TConnect}).</typeparam>
+        /// <typeparam name="TStartWith">The type of the start with expression (must be either TConnect or a type that implements IEnumerable{TConnect}).</typeparam>
+        /// <typeparam name="TAs">The type of the documents in the as field.</typeparam>
+        /// <typeparam name="TAsEnumerable">The type of the enumerable as field.</typeparam>
+        /// <param name="from">The from collection.</param>
+        /// <param name="connectFromField">The connect from field.</param>
+        /// <param name="connectToField">The connect to field.</param>
+        /// <param name="startWith">The start with value.</param>
+        /// <param name="as">The as field.</param>
+        /// <param name="depthField">The depth field.</param>
+        /// <param name="options">The options.</param>
+        /// <returns>The fluent aggregate interface.</returns>
+        IAggregateFluent<TNewResult> GraphLookup<TNewResult, TFrom, TConnect, TConnectFrom, TStartWith, TAs, TAsEnumerable>(
+            IMongoCollection<TFrom> from,
+            FieldDefinition<TFrom, TConnectFrom> connectFromField,
+            FieldDefinition<TFrom, TConnect> connectToField,
+            AggregateExpressionDefinition<TResult, TStartWith> startWith,
+            FieldDefinition<TNewResult, TAsEnumerable> @as,
+            FieldDefinition<TAs, int> depthField,
+            AggregateGraphLookupOptions<TNewResult, TFrom, TConnect, TConnectFrom, TStartWith, TAs, TAsEnumerable> options = null)
+                where TAsEnumerable : IEnumerable<TAs>;
+
         /// <summary>
         /// Appends a group stage to the pipeline.
         /// </summary>