[Internal] Binary Encoding: Fixes serializer for non-point operations. (#5058)

# Pull Request Template

## Description

In the current code, CosmosClient used a default version of
CosmosJsonDotNetSerializer as a propertiesSerializer which used the
ConfigurationManager.IsBinaryEncodingEnabled() to determine whether to
use binary encoding during serialization and de-serialization.

This propertiesSerializer is hooked into multiple serializers to use as
internal serializers and caused unexpected behaviors for APIs like
ReadManyItems, ChangeFeed and caused a regression.
## Type of change

There are 3 parts of the change:

- As part of this change we are introducing a binary serializer. The
default serializer will not just use `propertySerializer` but it will
check if the binary encoding flag is enabled and accordingly assign the
serializer.

- When a custom serializer is present, on the response path, no matter
what we always return the stream as Text to the serializer. If the
default serializer is used then we pass the binary stream in the
serializer.

- During some local testing, I realized that `CreateItemAsync` with
`PreTriggers` or `PostTriggers` doesn't work when binary encoding is
enabled. This is because when triggers are present in the item request
options the backend will pass the stream to the javascript engine, which
does not support binary encoded content at the moment. For long term,
since trigger operations won't be supported in the backend, avoiding the
binary encoding in such cases, will be the ideal approach. Therefore,
needed to skip using binary in such situation. For example, the below
code will fail today (before the fix in the PR) if it's used with binary
encoding:

```
ItemRequestOptions requestOptions = new ItemRequestOptions()
{
    PreTriggers = new List<string>() { triggerResponse.Resource.Id },
};

Comment comment = new Comment(
    Guid.NewGuid().ToString(),
    "pk",
    random.Next().ToString(),
    "[email protected]", 
    "This document is intended for binary encoding.");

ItemResponse<Comment> writeResponse = await container.CreateItemAsync<Comment>(
    item: comment,
    partitionKey: new Cosmos.PartitionKey(comment.pk),
    requestOptions: requestOptions
);
```

Please delete options that are not relevant.

- [X] Bug fix (non-breaking change which fixes an issue)

## Closing issues

To automatically close an issue: closes #5043

---------

Co-authored-by: Kiran Kumar Kolli <[email protected]>
Co-authored-by: Debdatta Kunda <[email protected]>
Co-authored-by: Debdatta Kunda <[email protected]>

Author: 3 people

Microsoft.Azure.Cosmos/src/CosmosClientOptions.cs

@@ -74,7 +74,8 @@ public class CosmosClientOptions
         private IWebProxy webProxy;
         private Func<HttpClient> httpClientFactory;
         private string applicationName;
-        private IFaultInjector faultInjector;
+        private IFaultInjector faultInjector;
+        private bool isCustomSerializerProvided;
 
         /// <summary>
         /// Creates a new CosmosClientOptions
@@ -624,7 +625,8 @@ public CosmosSerializer Serializer
                     throw new ArgumentException(
                         $"{nameof(this.Serializer)} is not compatible with {nameof(this.SerializerOptions)} or {nameof(this.UseSystemTextJsonSerializerWithOptions)}. Only one can be set.  ");
                 }
-
+
+                this.isCustomSerializerProvided = true;
                 this.serializerInternal = value;
             }
         }
@@ -1084,6 +1086,11 @@ internal static CosmosClientOptions GetCosmosClientOptionsWithCertificateFlag(st
 
             return clientOptions;
         }
+
+        internal bool IsCustomSerializerProvided()
+        {
+            return this.isCustomSerializerProvided;
+        }
 
         private static T GetValueFromConnectionString<T>(string connectionString, string keyName, T defaultValue)
         {

Microsoft.Azure.Cosmos/src/Resource/Container/ContainerCore.Items.cs

@@ -112,15 +112,15 @@ public async Task<ItemResponse<T>> ReadItemAsync<T>(
             ITrace trace,
             ItemRequestOptions requestOptions = null,
             CancellationToken cancellationToken = default)
-        {
+        {
             ResponseMessage response = await this.ProcessItemStreamAsync(
                 partitionKey: partitionKey,
                 itemId: id,
                 streamPayload: null,
                 operationType: OperationType.Read,
                 requestOptions: requestOptions,
                 trace: trace,
-                targetResponseSerializationFormat: default,
+                targetResponseSerializationFormat: this.GetTargetResponseSerializationFormat(),
                 cancellationToken: cancellationToken);
 
             return this.ClientContext.ResponseFactory.CreateItemResponse<T>(response);
@@ -249,7 +249,7 @@ public async Task<ItemResponse<T>> DeleteItemAsync<T>(
                 operationType: OperationType.Delete,
                 requestOptions: requestOptions,
                 trace: trace,
-                targetResponseSerializationFormat: default,
+                targetResponseSerializationFormat: this.GetTargetResponseSerializationFormat(),
                 cancellationToken: cancellationToken);
 
             return this.ClientContext.ResponseFactory.CreateItemResponse<T>(response);
@@ -847,8 +847,13 @@ private async Task<ResponseMessage> ExtractPartitionKeyAndProcessItemStreamAsync
 
             Stream itemStream;
             using (trace.StartChild("ItemSerialize"))
-            {
-                itemStream = this.ClientContext.SerializerCore.ToStream<T>(item);
+            {
+                // Serializing the item to a binary stream should be avoided when triggers are present in the item request options.
+                // This is because when triggers are present in the request options, the backend will pass the stream to the javascript
+                // engine, which does not support binary encoded content at the moment. For long term, since trigger operations won't
+                // be supported in the backend, avoiding the binary encoding in such cases, will be the ideal approach.
+                bool canUseBinaryEncoding = !ContainerCore.IsTriggerPresentInRequestOptions(requestOptions);
+                itemStream = this.ClientContext.SerializerCore.ToStream<T>(item, canUseBinaryEncodingForPointOperations: canUseBinaryEncoding);
             }
 
             // User specified PK value, no need to extract it
@@ -861,7 +866,7 @@ private async Task<ResponseMessage> ExtractPartitionKeyAndProcessItemStreamAsync
                         operationType,
                         requestOptions,
                         trace: trace,
-                        targetResponseSerializationFormat: default,
+                        targetResponseSerializationFormat: this.GetTargetResponseSerializationFormat(),
                         cancellationToken: cancellationToken);
             }
 
@@ -877,7 +882,7 @@ private async Task<ResponseMessage> ExtractPartitionKeyAndProcessItemStreamAsync
                     operationType,
                     requestOptions,
                     trace: trace,
-                    targetResponseSerializationFormat: default,
+                    targetResponseSerializationFormat: this.GetTargetResponseSerializationFormat(),
                     cancellationToken: cancellationToken);
 
                 if (responseMessage.IsSuccessStatusCode)
@@ -921,10 +926,15 @@ private async Task<ResponseMessage> ProcessItemStreamAsync(
             }
 
             ContainerInternal.ValidatePartitionKey(partitionKey, requestOptions);
-            string resourceUri = this.GetResourceUri(requestOptions, operationType, itemId);
+            string resourceUri = this.GetResourceUri(requestOptions, operationType, itemId);
 
             // Convert Text to Binary Stream.
-            if (ConfigurationManager.IsBinaryEncodingEnabled())
+            // Exception: Serializing a text stream to a binary stream should be avoided when triggers are present in the item request options.
+            // This is because when triggers are present in the request options, the backend will pass the stream to the javascript
+            // engine, which does not support binary encoded content at the moment. For long term, since trigger operations won't
+            // be supported in the backend, avoiding the binary encoding in such cases, will be the ideal approach.
+            if (ConfigurationManager.IsBinaryEncodingEnabled()
+                && !ContainerCore.IsTriggerPresentInRequestOptions(requestOptions))
             {
                 streamPayload = CosmosSerializationUtil.TrySerializeStreamToTargetFormat(
                     targetSerializationFormat: ContainerCore.GetTargetRequestSerializationFormat(),
@@ -1286,6 +1296,16 @@ private ChangeFeedProcessorBuilder GetChangeFeedProcessorBuilderPrivate(
                 applyBuilderConfiguration: changeFeedProcessor.ApplyBuildConfiguration).WithChangeFeedMode(mode);
         }
 
+        private JsonSerializationFormat? GetTargetResponseSerializationFormat()
+        {
+            if (this.ClientContext.ClientOptions.IsCustomSerializerProvided())
+            {
+                return JsonSerializationFormat.Text;
+            }
+
+            return default;
+        }
+
         private static JsonSerializationFormat GetTargetRequestSerializationFormat()
         {
             return ConfigurationManager.IsBinaryEncodingEnabled()
@@ -1607,6 +1627,17 @@ private static bool IsYMaxWithinX(
             Documents.Routing.Range<string> y)
         {
             return x.Max == y.Max || x.Contains(y.Max);
+        }
+
+        /// <summary>
+        /// Checks if the request options contain any triggers (pre or post).
+        /// </summary>
+        /// <param name="requestOptions">An instance of <see cref="ItemRequestOptions"/>.</param>
+        /// <returns>A boolean flag indicating if triggers were present in the request options.</returns>
+        private static bool IsTriggerPresentInRequestOptions(
+            ItemRequestOptions requestOptions)
+        {
+            return requestOptions?.PreTriggers != null || requestOptions?.PostTriggers != null;
         }
     }
 }

Microsoft.Azure.Cosmos/src/Serializer/CosmosSerializerCore.cs

@@ -17,9 +17,12 @@ namespace Microsoft.Azure.Cosmos
     /// </summary>
     internal class CosmosSerializerCore
     {
+        private readonly bool isBinaryEncodingEnabled;
+        private static readonly CosmosSerializer binarySerializer = new CosmosJsonSerializerWrapper(
+            new CosmosJsonDotNetSerializer(binaryEncodingEnabled: true));
+
         private static readonly CosmosSerializer propertiesSerializer = new CosmosJsonSerializerWrapper(
-            new CosmosJsonDotNetSerializer(
-                ConfigurationManager.IsBinaryEncodingEnabled()));
+            new CosmosJsonDotNetSerializer());
 
         private readonly CosmosSerializer customSerializer;
         private readonly CosmosSerializer sqlQuerySpecSerializer;
@@ -28,6 +31,7 @@ internal class CosmosSerializerCore
         internal CosmosSerializerCore(
             CosmosSerializer customSerializer = null)
         {
+            this.isBinaryEncodingEnabled = ConfigurationManager.IsBinaryEncodingEnabled();
             if (customSerializer == null)
             {
                 this.customSerializer = null;
@@ -81,9 +85,9 @@ internal T[] FromFeedStream<T>(Stream stream)
             return serializer.FromStream<T[]>(stream);
         }
 
-        internal Stream ToStream<T>(T input)
+        internal Stream ToStream<T>(T input, bool canUseBinaryEncodingForPointOperations = false)
         {
-            CosmosSerializer serializer = this.GetSerializer<T>();
+            CosmosSerializer serializer = this.GetSerializer<T>(canUseBinaryEncodingForPointOperations);
             return serializer.ToStream<T>(input);
         }
 
@@ -116,31 +120,29 @@ internal CosmosSerializer GetCustomOrDefaultSerializer()
                 return this.customSerializer;
             }
 
-            return CosmosSerializerCore.propertiesSerializer;
+            return this.GetDefaultSerializer();
         }
 
-        private CosmosSerializer GetSerializer<T>()
+        private CosmosSerializer GetSerializer<T>(
+            bool canUseBinaryEncodingForPointOperations = false)
         {
             Type inputType = typeof(T);
             if (inputType == typeof(PatchSpec))
             {
-                if (this.patchOperationSerializer == null)
-                {
-                    this.patchOperationSerializer = PatchOperationsJsonConverter.CreatePatchOperationsSerializer(
+                this.patchOperationSerializer ??= PatchOperationsJsonConverter.CreatePatchOperationsSerializer(
                         cosmosSerializer: this.customSerializer ?? new CosmosJsonDotNetSerializer(),
                         propertiesSerializer: CosmosSerializerCore.propertiesSerializer);
-                }
                 return this.patchOperationSerializer;
             }
 
-            if (this.customSerializer == null)
+            if (CosmosSerializerCore.IsInputTypeInternal(inputType))
             {
                 return CosmosSerializerCore.propertiesSerializer;
             }
 
-            if (CosmosSerializerCore.IsInputTypeInternal(inputType))
+            if (this.customSerializer == null)
             {
-                return CosmosSerializerCore.propertiesSerializer;
+                return this.GetDefaultSerializer(canUseBinaryEncodingForPointOperations);
             }
 
             if (inputType == typeof(SqlQuerySpec))
@@ -165,6 +167,14 @@ private CosmosSerializer GetSerializer<T>()
             return this.customSerializer;
         }
 
+        private CosmosSerializer GetDefaultSerializer(
+            bool canUseBinaryEncodingForPointOperations = false)
+        {
+            return this.isBinaryEncodingEnabled && canUseBinaryEncodingForPointOperations
+                ? CosmosSerializerCore.binarySerializer
+                : CosmosSerializerCore.propertiesSerializer;
+        }
+
         internal static bool IsInputTypeInternal(
             Type inputType)
         {
@@ -184,4 +194,4 @@ internal static bool IsInputTypeInternal(
                 || inputType == typeof(ChangeFeedQuerySpec);
         }
     }
-}
+}