CSHARP-4633: Avoiding Errors if Change Stream Events Exceed 16MB (#1209)

Author: BorisDog

src/MongoDB.Driver.Core/ChangeStreamDocument.cs

@@ -223,6 +223,11 @@ public TDocument FullDocumentBeforeChange
         /// </value>
         public BsonDocument ResumeToken => GetValue<BsonDocument>(nameof(ResumeToken), null);
 
+        /// <summary>
+        /// Gets the split event.
+        /// </summary>
+        public ChangeStreamSplitEvent SplitEvent => GetValue<ChangeStreamSplitEvent>(nameof(SplitEvent), null);
+
         /// <summary>
         /// Gets the update description.
         /// </summary>

src/MongoDB.Driver.Core/ChangeStreamDocumentSerializer.cs

@@ -52,6 +52,7 @@ public ChangeStreamDocumentSerializer(
             RegisterMember("OperationType", "operationType", ChangeStreamOperationTypeSerializer.Instance);
             RegisterMember("RenameTo", "to", ChangeStreamDocumentCollectionNamespaceSerializer.Instance);
             RegisterMember("ResumeToken", "_id", BsonDocumentSerializer.Instance);
+            RegisterMember("SplitEvent", "splitEvent", ChangeStreamSplitEventSerializer.Instance);
             RegisterMember("UpdateDescription", "updateDescription", ChangeStreamUpdateDescriptionSerializer.Instance);
             RegisterMember("WallTime", "wallTime", DateTimeSerializer.UtcInstance);
         }

src/MongoDB.Driver.Core/ChangeStreamSplitEvent.cs

@@ -0,0 +1,46 @@
+/* Copyright 2010-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 MongoDB.Driver.Core.Misc;
+
+namespace MongoDB.Driver
+{
+    /// <summary>
+    /// Change stream splitEvent data.
+    /// </summary>
+    public sealed class ChangeStreamSplitEvent
+    {
+        /// <summary>
+        /// Initializes a new instance of the <see cref="ChangeStreamSplitEvent" /> class.
+        /// </summary>
+        /// <param name="fragment">Fragment index.</param>
+        /// <param name="of">Total number of fragments.</param>
+        public ChangeStreamSplitEvent(int fragment, int of)
+        {
+            Fragment = Ensure.IsGreaterThanZero(fragment, nameof(of));
+            Of = Ensure.IsGreaterThanZero(of, nameof(of));
+        }
+
+        /// <summary>
+        /// Gets the fragment index, starting at 1.
+        /// </summary>
+        public int Fragment { get; }
+
+        /// <summary>
+        /// Total number of fragments for the event.
+        /// </summary>
+        public int Of { get; }
+    }
+}

src/MongoDB.Driver.Core/ChangeStreamSplitEventSerializer.cs

@@ -0,0 +1,70 @@
+/* Copyright 2010-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;
+using MongoDB.Bson.IO;
+using MongoDB.Bson.Serialization;
+using MongoDB.Bson.Serialization.Serializers;
+
+namespace MongoDB.Driver
+{
+    internal class ChangeStreamSplitEventSerializer : SealedClassSerializerBase<ChangeStreamSplitEvent>
+    {
+        #region static
+
+        public static ChangeStreamSplitEventSerializer Instance { get; } = new ChangeStreamSplitEventSerializer();
+
+        #endregion
+
+        protected override ChangeStreamSplitEvent DeserializeValue(BsonDeserializationContext context, BsonDeserializationArgs args)
+        {
+            var reader = context.Reader;
+            var fragment = -1;
+            var of = -1;
+
+            reader.ReadStartDocument();
+            while (reader.ReadBsonType() != 0)
+            {
+                var fieldName = reader.ReadName();
+                switch (fieldName)
+                {
+                    case "fragment":
+                        fragment = reader.ReadInt32();
+                        break;
+
+                    case "of":
+                        of = reader.ReadInt32();
+                        break;
+                    default:
+                        throw new FormatException($"Invalid field name: \"{fieldName}\".");
+                }
+            }
+            reader.ReadEndDocument();
+
+            return new(fragment, of);
+        }
+
+        protected override void SerializeValue(BsonSerializationContext context, BsonSerializationArgs args, ChangeStreamSplitEvent value)
+        {
+            var writer = context.Writer;
+            writer.WriteStartDocument();
+            writer.WriteName("fragment");
+            writer.WriteInt32(value.Fragment);
+            writer.WriteName("of");
+            writer.WriteInt32(value.Of);
+            writer.WriteEndDocument();
+        }
+    }
+}

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

@@ -51,6 +51,7 @@ public class Feature
         private static readonly Feature __changeStreamForDatabase = new Feature("ChangeStreamForDatabase", WireVersion.Server40);
         private static readonly Feature __changeStreamPostBatchResumeToken = new Feature("ChangeStreamPostBatchResumeToken", WireVersion.Server40);
         private static readonly Feature __changeStreamPrePostImages = new Feature("ChangeStreamPrePostImages", WireVersion.Server60);
+        private static readonly Feature __changeStreamSplitEventStage = new Feature("ChangeStreamSplitEventStage", WireVersion.Server70);
         private static readonly Feature __clientSideEncryption = new Feature("ClientSideEncryption", WireVersion.Server42);
         private static readonly Feature __clusteredIndexes = new Feature("ClusteredIndexes", WireVersion.Server53);
         private static readonly Feature __collation = new Feature("Collation", WireVersion.Server34);
@@ -292,6 +293,11 @@ public class Feature
         /// </summary>
         public static Feature ChangeStreamPrePostImages => __changeStreamPrePostImages;
 
+        /// <summary>
+        /// Gets the change stream splitEvent stage feature.
+        /// </summary>
+        public static Feature ChangeStreamSplitEventStage => __changeStreamSplitEventStage;
+
         /// <summary>
         /// Gets the client side encryption feature.
         /// </summary>

src/MongoDB.Driver/IAggregateFluentExtensions.cs

@@ -150,6 +150,18 @@ public static IAggregateFluent<TNewResult> BucketAutoForLinq2<TResult, TValue, T
             return aggregate.AppendStage(PipelineStageDefinitionBuilder.BucketAutoForLinq2(groupBy, buckets, output, options));
         }
 
+        /// <summary>
+        /// Appends a $changeStreamSplitLargeEvent stage to the pipeline.
+        /// </summary>
+        /// <typeparam name="TResult">The type of the result.</typeparam>
+        /// <param name="aggregate">The aggregate.</param>
+        /// <returns>The fluent aggregate interface.</returns>
+        public static IAggregateFluent<ChangeStreamDocument<TResult>> ChangeStreamSplitLargeEvent<TResult>(this IAggregateFluent<ChangeStreamDocument<TResult>> aggregate)
+        {
+            Ensure.IsNotNull(aggregate, nameof(aggregate));
+            return aggregate.AppendStage(PipelineStageDefinitionBuilder.ChangeStreamSplitLargeEvent<TResult>());
+        }
+
         /// <summary>
         /// Appends a $densify stage to the pipeline.
         /// </summary>

src/MongoDB.Driver/PipelineDefinitionBuilder.cs

@@ -300,6 +300,18 @@ public static PipelineDefinition<TInput, ChangeStreamDocument<TIntermediate>> Ch
             return pipeline.AppendStage(PipelineStageDefinitionBuilder.ChangeStream<TIntermediate>(options));
         }
 
+        /// <summary>
+        /// Appends a $changeStreamSplitLargeEvent stage.
+        /// </summary>
+        /// <typeparam name="TInput">The type of the input documents.</typeparam>
+        /// <param name="pipeline">The pipeline.</param>
+        /// <returns>A new pipeline with an additional stage.</returns>
+        public static PipelineDefinition<ChangeStreamDocument<TInput>, ChangeStreamDocument<TInput>> ChangeStreamSplitLargeEvent<TInput>(
+           this PipelineDefinition<ChangeStreamDocument<TInput>, ChangeStreamDocument<TInput>> pipeline)
+        {
+            return pipeline.AppendStage(PipelineStageDefinitionBuilder.ChangeStreamSplitLargeEvent<TInput>());
+        }
+
         /// <summary>
         /// Appends a $count stage to the pipeline.
         /// </summary>

src/MongoDB.Driver/PipelineStageDefinitionBuilder.cs

@@ -382,6 +382,14 @@ public static PipelineStageDefinition<TInput, ChangeStreamDocument<TInput>> Chan
             return stage;
         }
 
+        /// <summary>
+        /// Creates a $changeStreamSplitLargeEvent stage.
+        /// </summary>
+        /// <typeparam name="TInput">The type of the input documents.</typeparam>
+        /// <returns>The stage.</returns>
+        public static PipelineStageDefinition<ChangeStreamDocument<TInput>, ChangeStreamDocument<TInput>> ChangeStreamSplitLargeEvent<TInput>() =>
+            (PipelineStageDefinition<ChangeStreamDocument<TInput>, ChangeStreamDocument<TInput>>)new BsonDocument("$changeStreamSplitLargeEvent", new BsonDocument());
+
         /// <summary>
         /// Creates a $count stage.
         /// </summary>

tests/MongoDB.Driver.Core.Tests/ChangeStreamDocumentSerializerTests.cs

@@ -1,4 +1,4 @@
-/* Copyright 2017-present MongoDB Inc.
+/* Copyright 2010-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.
@@ -36,7 +36,7 @@ public void constructor_should_initialize_instance()
             var result = new ChangeStreamDocumentSerializer<BsonDocument>(documentSerializer);
 
             result._documentSerializer().Should().BeSameAs(documentSerializer);
-            result._memberSerializationInfo().Count.Should().Be(14);
+            result._memberSerializationInfo().Count.Should().Be(15);
             AssertRegisteredMember(result, "ClusterTime", "clusterTime", BsonTimestampSerializer.Instance);
             AssertRegisteredMember(result, "CollectionNamespace", "ns", ChangeStreamDocumentCollectionNamespaceSerializer.Instance);
             AssertRegisteredMember(result, "CollectionUuid", "ui", GuidSerializer.StandardInstance);
@@ -49,6 +49,7 @@ public void constructor_should_initialize_instance()
             AssertRegisteredMember(result, "OperationType", "operationType", ChangeStreamOperationTypeSerializer.Instance);
             AssertRegisteredMember(result, "RenameTo", "to", ChangeStreamDocumentCollectionNamespaceSerializer.Instance);
             AssertRegisteredMember(result, "ResumeToken", "_id", BsonDocumentSerializer.Instance);
+            AssertRegisteredMember(result, "SplitEvent", "splitEvent", ChangeStreamSplitEventSerializer.Instance);
             AssertRegisteredMember(result, "UpdateDescription", "updateDescription", ChangeStreamUpdateDescriptionSerializer.Instance);
             AssertRegisteredMember(result, "WallTime", "wallTime", DateTimeSerializer.UtcInstance);
         }

tests/MongoDB.Driver.Core.Tests/ChangeStreamDocumentTests.cs

@@ -1,4 +1,4 @@
-/* Copyright 2018-present MongoDB Inc.
+/* Copyright 2010-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.
@@ -461,6 +461,18 @@ public void ResumeToken_should_return_null_when_not_present()
             result.Should().BeNull();
         }
 
+        [Fact]
+        public void SplitEvent_should_return_null_when_not_present()
+        {
+            var value = new BsonDocument("x", 1234);
+            var backingDocument = new BsonDocument { { "other", 1 } };
+            var subject = CreateSubject(backingDocument: backingDocument);
+
+            var result = subject.SplitEvent;
+
+            result.Should().BeNull();
+        }
+
         [Theory]
         [InlineData("{}", new string[0], null)]
         [InlineData("{ x : 1 }", new[] { "a", "b", "c" }, null)]