[Internal] Binary Encoding: Adds Binary Encoding Support for Point Operations

Microsoft.Azure.Cosmos/src/Handler/RequestInvokerHandler.cs

@@ -24,6 +24,7 @@ namespace Microsoft.Azure.Cosmos.Handlers
     internal class RequestInvokerHandler : RequestHandler
     {
         private static readonly HttpMethod httpPatchMethod = new HttpMethod(HttpConstants.HttpMethods.Patch);
+        private static readonly string BinarySerializationFormat = SupportedSerializationFormats.CosmosBinary.ToString();
         private static (bool, ResponseMessage) clientIsValid = (false, null);
 
         private readonly CosmosClient client;
@@ -67,6 +68,12 @@ public override async Task<ResponseMessage> SendAsync(
                 request.Headers.Add(HttpConstants.HttpHeaders.Prefer, HttpConstants.HttpHeaderValues.PreferReturnMinimal);
             }
 
+            if (ConfigurationManager.IsBinaryEncodingEnabled()
+                && RequestInvokerHandler.IsPointOperationSupportedForBinaryEncoding(request))
+            {
+                request.Headers.Add(HttpConstants.HttpHeaders.SupportedSerializationFormats, RequestInvokerHandler.BinarySerializationFormat);
+            }
+
             await this.ValidateAndSetConsistencyLevelAsync(request);
             this.SetPriorityLevel(request);
 
@@ -94,6 +101,14 @@ public override async Task<ResponseMessage> SendAsync(
                 ((CosmosTraceDiagnostics)response.Diagnostics).Value.AddOrUpdateDatum("ExcludedRegions", request.RequestOptions.ExcludeRegions);
             }
 
+            if (ConfigurationManager.IsBinaryEncodingEnabled()
+                && RequestInvokerHandler.IsPointOperationSupportedForBinaryEncoding(request)
+                && response.Content != null
+                && response.Content is not CloneableStream)
+            {
+                response.Content = await StreamExtension.AsClonableStreamAsync(response.Content, default);
+            }
+
             return response;
         }
 
@@ -547,6 +562,16 @@ private static bool IsItemNoRepsonseSet(bool enableContentResponseOnWrite, Opera
               operationType == OperationType.Patch);
         }
 
+        private static bool IsPointOperationSupportedForBinaryEncoding(RequestMessage request)
+        {
+            return request.ResourceType == ResourceType.Document 
+                && (request.OperationType == OperationType.Create
+                    || request.OperationType == OperationType.Replace
+                    || request.OperationType == OperationType.Delete
+                    || request.OperationType == OperationType.Read
+                    || request.OperationType == OperationType.Upsert);
+        }
+
         private static bool IsClientNoResponseSet(CosmosClientOptions clientOptions, OperationType operationType)
         {
             return clientOptions != null

Microsoft.Azure.Cosmos/src/Json/Interop/CosmosDBToNewtonsoftReader.cs

@@ -192,7 +192,7 @@ public override byte[] ReadAsBytes()
         public override DateTime? ReadAsDateTime()
         {
             this.Read();
-            if (this.jsonReader.CurrentTokenType == JsonTokenType.EndArray)
+            if (this.jsonReader.CurrentTokenType == JsonTokenType.Null || this.jsonReader.CurrentTokenType == JsonTokenType.EndArray)
             {
                 return null;
             }
@@ -211,7 +211,7 @@ public override byte[] ReadAsBytes()
         public override DateTimeOffset? ReadAsDateTimeOffset()
         {
             this.Read();
-            if (this.jsonReader.CurrentTokenType == JsonTokenType.EndArray)
+            if (this.jsonReader.CurrentTokenType == JsonTokenType.Null || this.jsonReader.CurrentTokenType == JsonTokenType.EndArray)
             {
                 return null;
             }
@@ -260,7 +260,7 @@ public override byte[] ReadAsBytes()
         public override string ReadAsString()
         {
             this.Read();
-            if (this.jsonReader.CurrentTokenType == JsonTokenType.EndArray)
+            if (this.jsonReader.CurrentTokenType == JsonTokenType.Null || this.jsonReader.CurrentTokenType == JsonTokenType.EndArray)
             {
                 return null;
             }
@@ -278,7 +278,7 @@ public override string ReadAsString()
         private double? ReadNumberValue()
         {
             this.Read();
-            if (this.jsonReader.CurrentTokenType == JsonTokenType.EndArray)
+            if (this.jsonReader.CurrentTokenType == JsonTokenType.Null || this.jsonReader.CurrentTokenType == JsonTokenType.EndArray)
             {
                 return null;
             }

Microsoft.Azure.Cosmos/src/Json/Interop/CosmosDBToNewtonsoftWriter.cs

@@ -266,7 +266,8 @@ public override void WriteValue(bool value)
         /// <param name="value">The <see cref="short"/> value to write.</param>
         public override void WriteValue(short value)
         {
-            base.WriteValue((long)value);
+            base.WriteValue((Int16)value);
+            this.jsonWriter.WriteInt16Value(value);
         }
 
         /// <summary>

Microsoft.Azure.Cosmos/src/RequestOptions/ItemRequestOptions.cs

@@ -127,6 +127,25 @@ public ConsistencyLevel? ConsistencyLevel
         /// </remarks>
         public DedicatedGatewayRequestOptions DedicatedGatewayRequestOptions { get; set; }
 
+        /// <summary>
+        /// Gets or sets the boolean to enable binary response for point operations like Create, Upsert, Read, Patch, and Replace.
+        /// Setting this option to true will cause the response to be in binary format. This request option will remain internal only
+        /// since the consumer of thie flag will be the internal components of the cosmos db ecosystem.
+        /// </summary>
+        /// <example>
+        /// <code language="c#">
+        /// <![CDATA[
+        /// ItemRequestOptions requestOptions = new ItemRequestOptions() { EnableBinaryResponseOnPointOperations = true };
+        /// ResponseMessage responseMessage = await container.CreateItemStreamAsync(createStream, new Cosmos.PartitionKey(comment.pk), requestOptions);
+        /// Assert.AreEqual(HttpStatusCode.Created, responseMessage.StatusCode);
+        /// ]]>
+        /// </code>
+        /// </example>
+        /// <remarks>
+        /// This is optimal for workloads where the returned resource can be processed in binary format.
+        /// </remarks>
+        internal bool EnableBinaryResponseOnPointOperations { get; set; }
+
         /// <summary>
         /// Fill the CosmosRequestMessage headers with the set properties
         /// </summary>

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

@@ -59,7 +59,8 @@ public async Task<ResponseMessage> CreateItemStreamAsync(
                 streamPayload: streamPayload,
                 operationType: OperationType.Create,
                 requestOptions: requestOptions,
-                trace: trace,
+                trace: trace,
+                targetResponseSerializationFormat: JsonSerializationFormat.Text,
                 cancellationToken: cancellationToken);
         }
 
@@ -100,7 +101,8 @@ public async Task<ResponseMessage> ReadItemStreamAsync(
                 streamPayload: null,
                 operationType: OperationType.Read,
                 requestOptions: requestOptions,
-                trace: trace,
+                trace: trace,
+                targetResponseSerializationFormat: JsonSerializationFormat.Text,
                 cancellationToken: cancellationToken);
         }
 
@@ -117,7 +119,8 @@ public async Task<ItemResponse<T>> ReadItemAsync<T>(
                 streamPayload: null,
                 operationType: OperationType.Read,
                 requestOptions: requestOptions,
-                trace: trace,
+                trace: trace,
+                targetResponseSerializationFormat: default,
                 cancellationToken: cancellationToken);
 
             return this.ClientContext.ResponseFactory.CreateItemResponse<T>(response);
@@ -136,7 +139,8 @@ public async Task<ResponseMessage> UpsertItemStreamAsync(
                 streamPayload: streamPayload,
                 operationType: OperationType.Upsert,
                 requestOptions: requestOptions,
-                trace: trace,
+                trace: trace,
+                targetResponseSerializationFormat: JsonSerializationFormat.Text,
                 cancellationToken: cancellationToken);
         }
 
@@ -178,7 +182,8 @@ public async Task<ResponseMessage> ReplaceItemStreamAsync(
                 streamPayload: streamPayload,
                 operationType: OperationType.Replace,
                 requestOptions: requestOptions,
-                trace: trace,
+                trace: trace,
+                targetResponseSerializationFormat: JsonSerializationFormat.Text,
                 cancellationToken: cancellationToken);
         }
 
@@ -225,7 +230,8 @@ public async Task<ResponseMessage> DeleteItemStreamAsync(
                 streamPayload: null,
                 operationType: OperationType.Delete,
                 requestOptions: requestOptions,
-                trace: trace,
+                trace: trace,
+                targetResponseSerializationFormat: JsonSerializationFormat.Text,
                 cancellationToken: cancellationToken);
         }
 
@@ -242,7 +248,8 @@ public async Task<ItemResponse<T>> DeleteItemAsync<T>(
                 streamPayload: null,
                 operationType: OperationType.Delete,
                 requestOptions: requestOptions,
-                trace: trace,
+                trace: trace,
+                targetResponseSerializationFormat: default,
                 cancellationToken: cancellationToken);
 
             return this.ClientContext.ResponseFactory.CreateItemResponse<T>(response);
@@ -853,7 +860,8 @@ private async Task<ResponseMessage> ExtractPartitionKeyAndProcessItemStreamAsync
                         itemStream,
                         operationType,
                         requestOptions,
-                        trace: trace,
+                        trace: trace,
+                        targetResponseSerializationFormat: default,
                         cancellationToken: cancellationToken);
             }
 
@@ -868,7 +876,8 @@ private async Task<ResponseMessage> ExtractPartitionKeyAndProcessItemStreamAsync
                     itemStream,
                     operationType,
                     requestOptions,
-                    trace: trace,
+                    trace: trace,
+                    targetResponseSerializationFormat: default,
                     cancellationToken: cancellationToken);
 
                 if (responseMessage.IsSuccessStatusCode)
@@ -897,7 +906,8 @@ private async Task<ResponseMessage> ProcessItemStreamAsync(
             Stream streamPayload,
             OperationType operationType,
             ItemRequestOptions requestOptions,
-            ITrace trace,
+            ITrace trace,
+            JsonSerializationFormat? targetResponseSerializationFormat,
             CancellationToken cancellationToken)
         {
             if (trace == null)
@@ -912,6 +922,11 @@ private async Task<ResponseMessage> ProcessItemStreamAsync(
 
             ContainerInternal.ValidatePartitionKey(partitionKey, requestOptions);
             string resourceUri = this.GetResourceUri(requestOptions, operationType, itemId);
+
+            // Convert Text to Binary Stream.
+            streamPayload = CosmosSerializationUtil.TrySerializeStreamToTargetFormat(
+                targetSerializationFormat: ContainerCore.GetTargetRequestSerializationFormat(),
+                inputStream: streamPayload == null ? null : await StreamExtension.AsClonableStreamAsync(streamPayload));
 
             ResponseMessage responseMessage = await this.ClientContext.ProcessResourceOperationStreamAsync(
                 resourceUri: resourceUri,
@@ -925,6 +940,16 @@ private async Task<ResponseMessage> ProcessItemStreamAsync(
                 requestEnricher: null,
                 trace: trace,
                 cancellationToken: cancellationToken);
+
+            // Convert Binary Stream to Text.
+            if (targetResponseSerializationFormat.HasValue
+                && (requestOptions == null || !requestOptions.EnableBinaryResponseOnPointOperations)
+                && responseMessage?.Content is CloneableStream outputCloneableStream)
+            {
+                responseMessage.Content = CosmosSerializationUtil.TrySerializeStreamToTargetFormat(
+                    targetSerializationFormat: targetResponseSerializationFormat.Value,
+                    inputStream: outputCloneableStream);
+            }
 
             return responseMessage;
         }
@@ -1217,7 +1242,8 @@ public Task<ResponseMessage> PatchItemStreamAsync(
                 streamPayload: streamPayload,
                 operationType: OperationType.Patch,
                 requestOptions: requestOptions,
-                trace: trace,
+                trace: trace,
+                targetResponseSerializationFormat: JsonSerializationFormat.Text,
                 cancellationToken: cancellationToken);
         }
 
@@ -1255,6 +1281,13 @@ private ChangeFeedProcessorBuilder GetChangeFeedProcessorBuilderPrivate(
                 applyBuilderConfiguration: changeFeedProcessor.ApplyBuildConfiguration).WithChangeFeedMode(mode);
         }
 
+        private static JsonSerializationFormat GetTargetRequestSerializationFormat()
+        {
+            return ConfigurationManager.IsBinaryEncodingEnabled()
+                ? JsonSerializationFormat.Binary
+                : JsonSerializationFormat.Text;
+        }
+
         /// <summary>
         /// This method is useful for determining if a smaller, more granular feed range (y) is fully contained within a broader feed range (x), which is a common operation in distributed systems to manage partitioned data.
         ///