CSHARP-4706: Support for $out to Time-series collections (#1223)

Author: adelinowona

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

@@ -41,6 +41,7 @@ public class Feature
         private static readonly Feature __aggregateMerge = new Feature("AggregateMerge", WireVersion.Server42);
         private static readonly Feature __aggregateOut = new Feature("AggregateOut", WireVersion.Server26);
         private static readonly Feature __aggregateOutOnSecondary = new Feature("AggregateOutOnSecondary", WireVersion.Server50);
+        private static readonly Feature __aggregateOutTimeSeries = new Feature("AggregateOutTimeSeries", WireVersion.Server70);
         private static readonly Feature __aggregateOutToDifferentDatabase = new Feature("AggregateOutToDifferentDatabase", WireVersion.Server44);
         private static readonly Feature __aggregateToString = new Feature("AggregateToString", WireVersion.Server40);
         private static readonly Feature __aggregateUnionWith = new Feature("AggregateUnionWith", WireVersion.Server44);
@@ -237,10 +238,15 @@ public class Feature
         public static Feature AggregateOut => __aggregateOut;
 
         /// <summary>
-        /// Gets the aggregate out on secondary feature,
+        /// Gets the aggregate out on secondary feature.
         /// </summary>
         public static Feature AggregateOutOnSecondary => __aggregateOutOnSecondary;
 
+        /// <summary>
+        /// Gets the aggregate out to time series feature.
+        /// </summary>
+        public static Feature AggregateOutTimeSeries => __aggregateOutTimeSeries;
+
         /// <summary>
         /// Gets the aggregate out to a different database feature.
         /// </summary>

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

@@ -330,7 +330,7 @@ private IReadOnlyList<BsonDocument> SimplifyOutStageIfOutputDatabaseIsSameAsInpu
         {
             var lastStage = pipeline.Last();
             var lastStageName = lastStage.GetElement(0).Name;
-            if (lastStageName == "$out" && lastStage["$out"] is BsonDocument outDocument)
+            if (lastStageName == "$out" && lastStage["$out"] is BsonDocument outDocument && !outDocument.Contains("timeseries"))
             {
                 if (outDocument.TryGetValue("db", out var db) && db.IsString &&
                     outDocument.TryGetValue("coll", out var coll) && coll.IsString)

src/MongoDB.Driver/AggregateFluent.cs

@@ -210,6 +210,20 @@ public override IAsyncCursor<TResult> Out(string collectionName, CancellationTok
             return Out(outputCollection, cancellationToken);
         }
 
+        public override IAsyncCursor<TResult> Out(IMongoCollection<TResult> outputCollection, TimeSeriesOptions timeSeriesOptions, CancellationToken cancellationToken)
+        {
+            Ensure.IsNotNull(outputCollection, nameof(outputCollection));
+            var aggregate = WithPipeline(_pipeline.Out(outputCollection, timeSeriesOptions));
+            return aggregate.ToCursor(cancellationToken);
+        }
+
+        public override IAsyncCursor<TResult> Out(string collectionName, TimeSeriesOptions timeSeriesOptions, CancellationToken cancellationToken)
+        {
+            Ensure.IsNotNull(collectionName, nameof(collectionName));
+            var outputCollection = Database.GetCollection<TResult>(collectionName);
+            return Out(outputCollection, timeSeriesOptions, cancellationToken);
+        }
+
         public override Task<IAsyncCursor<TResult>> OutAsync(IMongoCollection<TResult> outputCollection, CancellationToken cancellationToken)
         {
             Ensure.IsNotNull(outputCollection, nameof(outputCollection));
@@ -224,6 +238,20 @@ public override Task<IAsyncCursor<TResult>> OutAsync(string collectionName, Canc
             return OutAsync(outputCollection, cancellationToken);
         }
 
+        public override Task<IAsyncCursor<TResult>> OutAsync(IMongoCollection<TResult> outputCollection, TimeSeriesOptions timeSeriesOptions, CancellationToken cancellationToken)
+        {
+            Ensure.IsNotNull(outputCollection, nameof(outputCollection));
+            var aggregate = WithPipeline(_pipeline.Out(outputCollection, timeSeriesOptions));
+            return aggregate.ToCursorAsync(cancellationToken);
+        }
+
+        public override Task<IAsyncCursor<TResult>> OutAsync(string collectionName, TimeSeriesOptions timeSeriesOptions, CancellationToken cancellationToken)
+        {
+            Ensure.IsNotNull(collectionName, nameof(collectionName));
+            var outputCollection = Database.GetCollection<TResult>(collectionName);
+            return OutAsync(outputCollection, timeSeriesOptions, cancellationToken);
+        }
+
         public override IAggregateFluent<TNewResult> Project<TNewResult>(ProjectionDefinition<TResult, TNewResult> projection)
         {
             return WithPipeline(_pipeline.Project(projection));

src/MongoDB.Driver/AggregateFluentBase.cs

@@ -192,6 +192,18 @@ public virtual IAsyncCursor<TResult> Out(string collectionName, CancellationToke
             throw new NotImplementedException();
         }
 
+        /// <inheritdoc />
+        public virtual IAsyncCursor<TResult> Out(IMongoCollection<TResult> outputCollection, TimeSeriesOptions timeSeriesOptions, CancellationToken cancellationToken)
+        {
+            throw new NotImplementedException();
+        }
+
+        /// <inheritdoc />
+        public virtual IAsyncCursor<TResult> Out(string collectionName, TimeSeriesOptions timeSeriesOptions, CancellationToken cancellationToken)
+        {
+            throw new NotImplementedException();
+        }
+
         /// <inheritdoc />
         public virtual Task<IAsyncCursor<TResult>> OutAsync(IMongoCollection<TResult> outputCollection, CancellationToken cancellationToken)
         {
@@ -201,6 +213,15 @@ public virtual Task<IAsyncCursor<TResult>> OutAsync(IMongoCollection<TResult> ou
         /// <inheritdoc />
         public abstract Task<IAsyncCursor<TResult>> OutAsync(string collectionName, CancellationToken cancellationToken);
 
+        /// <inheritdoc />
+        public virtual Task<IAsyncCursor<TResult>> OutAsync(IMongoCollection<TResult> outputCollection, TimeSeriesOptions timeSeriesOptions, CancellationToken cancellationToken)
+        {
+            throw new NotImplementedException();
+        }
+
+        /// <inheritdoc />
+        public abstract Task<IAsyncCursor<TResult>> OutAsync(string collectionName, TimeSeriesOptions timeSeriesOptions, CancellationToken cancellationToken);
+
         /// <inheritdoc />
         public abstract IAggregateFluent<TNewResult> Project<TNewResult>(ProjectionDefinition<TResult, TNewResult> projection);
 

src/MongoDB.Driver/IAggregateFluent.cs

@@ -303,6 +303,24 @@ IAggregateFluent<TNewResult> Lookup<TForeignDocument, TAsElement, TAs, TNewResul
         /// <returns>A cursor.</returns>
         IAsyncCursor<TResult> Out(string collectionName, CancellationToken cancellationToken = default(CancellationToken));
 
+        /// <summary>
+        /// Appends an out 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="timeSeriesOptions">The time series options.</param>
+        /// <param name="cancellationToken">The cancellation token.</param>
+        /// <returns>A cursor.</returns>
+        IAsyncCursor<TResult> Out(IMongoCollection<TResult> outputCollection, TimeSeriesOptions timeSeriesOptions, CancellationToken cancellationToken = default(CancellationToken));
+
+        /// <summary>
+        /// Appends an out stage to the pipeline and executes it, and then returns a cursor to read the contents of the output collection.
+        /// </summary>
+        /// <param name="collectionName">Name of the collection.</param>
+        /// <param name="timeSeriesOptions">The time series options.</param>
+        /// <param name="cancellationToken">The cancellation token.</param>
+        /// <returns>A cursor.</returns>
+        IAsyncCursor<TResult> Out(string collectionName, TimeSeriesOptions timeSeriesOptions, CancellationToken cancellationToken = default(CancellationToken));
+
         /// <summary>
         /// Appends an out stage to the pipeline and executes it, and then returns a cursor to read the contents of the output collection.
         /// </summary>
@@ -319,6 +337,24 @@ IAggregateFluent<TNewResult> Lookup<TForeignDocument, TAsElement, TAs, TNewResul
         /// <returns>A Task whose result is a cursor.</returns>
         Task<IAsyncCursor<TResult>> OutAsync(string collectionName, CancellationToken cancellationToken = default(CancellationToken));
 
+        /// <summary>
+        /// Appends an out 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="timeSeriesOptions">The time series options.</param>
+        /// <param name="cancellationToken">The cancellation token.</param>
+        /// <returns>A Task whose result is a cursor.</returns>
+        Task<IAsyncCursor<TResult>> OutAsync(IMongoCollection<TResult> outputCollection, TimeSeriesOptions timeSeriesOptions, CancellationToken cancellationToken = default(CancellationToken));
+
+        /// <summary>
+        /// Appends an out stage to the pipeline and executes it, and then returns a cursor to read the contents of the output collection.
+        /// </summary>
+        /// <param name="collectionName">Name of the collection.</param>
+        /// <param name="timeSeriesOptions">The time series options.</param>
+        /// <param name="cancellationToken">The cancellation token.</param>
+        /// <returns>A Task whose result is a cursor.</returns>
+        Task<IAsyncCursor<TResult>> OutAsync(string collectionName, TimeSeriesOptions timeSeriesOptions, CancellationToken cancellationToken = default(CancellationToken));
+
         /// <summary>
         /// Appends a project stage to the pipeline.
         /// </summary>

src/MongoDB.Driver/PipelineDefinitionBuilder.cs

@@ -960,14 +960,16 @@ public static PipelineDefinition<TInput, TOutput> Merge<TInput, TIntermediate, T
         /// <typeparam name="TOutput">The type of the output documents.</typeparam>
         /// <param name="pipeline">The pipeline.</param>
         /// <param name="outputCollection">The output collection.</param>
+        /// <param name="timeSeriesOptions">The time series options.</param>
         /// <returns>A new pipeline with an additional stage.</returns>
         /// <exception cref="System.NotSupportedException"></exception>
         public static PipelineDefinition<TInput, TOutput> Out<TInput, TOutput>(
             this PipelineDefinition<TInput, TOutput> pipeline,
-            IMongoCollection<TOutput> outputCollection)
+            IMongoCollection<TOutput> outputCollection,
+            TimeSeriesOptions timeSeriesOptions = null)
         {
             Ensure.IsNotNull(pipeline, nameof(pipeline));
-            return pipeline.AppendStage(PipelineStageDefinitionBuilder.Out<TOutput>(outputCollection));
+            return pipeline.AppendStage(PipelineStageDefinitionBuilder.Out<TOutput>(outputCollection, timeSeriesOptions));
         }
 
         /// <summary>
@@ -1111,7 +1113,7 @@ public static PipelineDefinition<TInput, TOutput> ReplaceWith<TInput, TIntermedi
         /// </param>
         /// <param name="scoreDetails">
         /// Flag that specifies whether to return a detailed breakdown
-        /// of the score for each document in the result. 
+        /// of the score for each document in the result.
         /// </param>
         /// <returns>A new pipeline with an additional stage.</returns>
         public static PipelineDefinition<TInput, TOutput> Search<TInput, TOutput>(

src/MongoDB.Driver/PipelineStageDefinitionBuilder.cs

@@ -1244,14 +1244,21 @@ public static PipelineStageDefinition<TInput, TOutput> OfType<TInput, TOutput>(
         /// </summary>
         /// <typeparam name="TInput">The type of the input documents.</typeparam>
         /// <param name="outputCollection">The output collection.</param>
+        /// <param name="timeSeriesOptions">The time series options.</param>
         /// <returns>The stage.</returns>
         public static PipelineStageDefinition<TInput, TInput> Out<TInput>(
-            IMongoCollection<TInput> outputCollection)
+            IMongoCollection<TInput> outputCollection,
+            TimeSeriesOptions timeSeriesOptions = null)
         {
             Ensure.IsNotNull(outputCollection, nameof(outputCollection));
             var outputDatabaseName = outputCollection.Database.DatabaseNamespace.DatabaseName;
             var outputCollectionName = outputCollection.CollectionNamespace.CollectionName;
-            var outDocument = new BsonDocument { { "db", outputDatabaseName }, { "coll", outputCollectionName } };
+            var outDocument = new BsonDocument
+                {
+                    { "db", outputDatabaseName },
+                    { "coll", outputCollectionName },
+                    { "timeseries", () => timeSeriesOptions.ToBsonDocument(), timeSeriesOptions != null}
+                };
             return new BsonDocumentPipelineStageDefinition<TInput, TInput>(new BsonDocument("$out", outDocument));
         }
 
@@ -1331,7 +1338,7 @@ public static PipelineStageDefinition<TInput, TOutput> Project<TInput, TOutput>(
         /// </param>
         /// <param name="scoreDetails">
         /// Flag that specifies whether to return a detailed breakdown
-        /// of the score for each document in the result. 
+        /// of the score for each document in the result.
         /// </param>
         /// <returns>The stage.</returns>
         public static PipelineStageDefinition<TInput, TInput> Search<TInput>(

tests/MongoDB.Driver.Core.Tests/Core/Operations/AggregateToCollectionOperationTests.cs

@@ -607,6 +607,43 @@ public void Execute_should_return_expected_result(
             result.Should().HaveCount(1);
         }
 
+        [Theory]
+        [ParameterAttributeData]
+        public void Execute_to_time_series_collection_should_work(
+            [Values(false, true)] bool usingDifferentOutputDatabase,
+            [Values(false, true)] bool async)
+        {
+            RequireServer.Check().Supports(Feature.AggregateOutTimeSeries);
+
+            var pipeline = new List<BsonDocument> { BsonDocument.Parse("{ $match : { _id : 1 } }") };
+            var inputDatabaseName = _databaseNamespace.DatabaseName;
+            var inputCollectionName = _collectionNamespace.CollectionName;
+            var outputDatabaseName = usingDifferentOutputDatabase ? $"{inputDatabaseName}-outputdatabase-timeseries" : inputDatabaseName;
+            var outputCollectionName = $"{inputCollectionName}-outputcollection-timeseries";
+
+            pipeline.Add(new BsonDocument { {"$set", new BsonDocument { {"time", DateTime.Now } } } } );
+            pipeline.Add(BsonDocument.Parse($"{{ $out : {{ db : '{outputDatabaseName}', coll : '{outputCollectionName}', timeseries: {{ timeField: 'time' }} }} }}"));
+
+            EnsureTestData();
+            if (usingDifferentOutputDatabase)
+            {
+                EnsureDatabaseExists(outputDatabaseName);
+            }
+            var subject = new AggregateToCollectionOperation(_collectionNamespace, pipeline, _messageEncoderSettings);
+
+            ExecuteOperation(subject, async);
+
+            var databaseNamespace = new DatabaseNamespace(outputDatabaseName);
+            var result = ReadAllFromCollection(new CollectionNamespace(databaseNamespace, outputCollectionName), async);
+
+            result.Should().NotBeNull();
+            result.Should().HaveCount(1);
+
+            var output = ListCollections(databaseNamespace);
+            output["cursor"]["firstBatch"][0][0].ToString().Should().Be($"{outputCollectionName}"); // checking name of collection
+            output["cursor"]["firstBatch"][0][1].ToString().Should().Be("timeseries"); // checking type of collection
+        }
+
         [Theory]
         [ParameterAttributeData]
         public void Execute_should_return_expected_result_when_AllowDiskUse_is_set(

tests/MongoDB.Driver.Core.Tests/Core/Operations/OperationTestBase.cs

@@ -347,6 +347,19 @@ protected void KillOpenTransactions()
             }
         }
 
+        protected BsonDocument ListCollections(DatabaseNamespace databaseNamespace)
+        {
+            var listCollectionsCommand = new BsonDocument
+            {
+                { "listCollections", 1 }, { "filter", new BsonDocument { { "type", "timeseries" } } }
+            };
+
+            var runCommandOperation = new ReadCommandOperation<BsonDocument>(databaseNamespace, listCollectionsCommand,
+                BsonDocumentSerializer.Instance, _messageEncoderSettings);
+
+            return ExecuteOperation(runCommandOperation);
+        }
+
         protected Profiler Profile(DatabaseNamespace databaseNamespace)
         {
             var op = new WriteCommandOperation<BsonDocument>(