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

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

@@ -67,6 +67,17 @@ public override async Task<ResponseMessage> SendAsync(
                 request.Headers.Add(HttpConstants.HttpHeaders.Prefer, HttpConstants.HttpHeaderValues.PreferReturnMinimal);
             }
 
+            if (ConfigurationManager.IsBinaryEncodingEnabled()
+                && request.OperationType.IsPointOperation()
+                && request.ResourceType == ResourceType.Document)
+            {
+                request.Headers.Add(HttpConstants.HttpHeaders.SupportedSerializationFormats, SupportedSerializationFormats.CosmosBinary.ToString());
+                if (request.Content != null)
+                {
+                    request.Headers.Add(HttpConstants.HttpHeaders.ContentSerializationFormat, SupportedSerializationFormats.CosmosBinary.ToString());
+                }
+            }
+
             await this.ValidateAndSetConsistencyLevelAsync(request);
             this.SetPriorityLevel(request);
 

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,26 @@ 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, which can reduce networking and CPU load
+        /// by not sending the resource back over the network and serializing it on the client.
+        /// </summary>
+        /// <example>
+        /// <code language="c#">
+        /// <![CDATA[
+        /// ItemRequestOptions requestOptions = new ItemRequestOptions() { EnableBinaryResponseOnPointOperations = true };
+        /// ItemResponse itemResponse = await this.container.CreateItemAsync<ToDoActivity>(tests, new PartitionKey(test.status), requestOptions);
+        /// Assert.AreEqual(HttpStatusCode.Created, itemResponse.StatusCode);
+        /// Assert.IsNotNull(itemResponse.Resource);
+        /// ]]>
+        /// </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

@@ -51,15 +51,17 @@ public async Task<ResponseMessage> CreateItemStreamAsync(
             ITrace trace,
             ItemRequestOptions requestOptions = null,
             CancellationToken cancellationToken = default)
-        {
+        {
             return await this.ProcessItemStreamAsync(
                 partitionKey: partitionKey,
                 itemId: null,
                 streamPayload: streamPayload,
                 operationType: OperationType.Create,
                 requestOptions: requestOptions,
                 trace: trace,
-                cancellationToken: cancellationToken);
+                cancellationToken: cancellationToken,
+                targetRequestSerializationFormat: ContainerCore.GetTargetRequestSerializationFormat(),
+                targetResponseSerializationFormat: JsonSerializationFormat.Text);
         }
 
         public async Task<ItemResponse<T>> CreateItemAsync<T>(
@@ -79,7 +81,8 @@ public async Task<ItemResponse<T>> CreateItemAsync<T>(
                 itemId: null,
                 item: item,
                 operationType: OperationType.Create,
-                requestOptions: requestOptions,
+                requestOptions: requestOptions,
+                targetRequestSerializationFormat: ContainerCore.GetTargetRequestSerializationFormat(),
                 trace: trace,
                 cancellationToken: cancellationToken);
 
@@ -92,15 +95,17 @@ public async Task<ResponseMessage> ReadItemStreamAsync(
             ITrace trace,
             ItemRequestOptions requestOptions = null,
             CancellationToken cancellationToken = default)
-        {
+        {
             return await this.ProcessItemStreamAsync(
                 partitionKey: partitionKey,
                 itemId: id,
                 streamPayload: null,
                 operationType: OperationType.Read,
                 requestOptions: requestOptions,
                 trace: trace,
-                cancellationToken: cancellationToken);
+                cancellationToken: cancellationToken,
+                targetRequestSerializationFormat: ContainerCore.GetTargetRequestSerializationFormat(),
+                targetResponseSerializationFormat: JsonSerializationFormat.Text);
         }
 
         public async Task<ItemResponse<T>> ReadItemAsync<T>(
@@ -115,7 +120,8 @@ public async Task<ItemResponse<T>> ReadItemAsync<T>(
                 itemId: id,
                 streamPayload: null,
                 operationType: OperationType.Read,
-                requestOptions: requestOptions,
+                requestOptions: requestOptions,
+                targetRequestSerializationFormat: ContainerCore.GetTargetRequestSerializationFormat(),
                 trace: trace,
                 cancellationToken: cancellationToken);
 
@@ -136,7 +142,9 @@ public async Task<ResponseMessage> UpsertItemStreamAsync(
                 operationType: OperationType.Upsert,
                 requestOptions: requestOptions,
                 trace: trace,
-                cancellationToken: cancellationToken);
+                cancellationToken: cancellationToken,
+                targetRequestSerializationFormat: ContainerCore.GetTargetRequestSerializationFormat(),
+                targetResponseSerializationFormat: JsonSerializationFormat.Text);
         }
 
         public async Task<ItemResponse<T>> UpsertItemAsync<T>(
@@ -156,7 +164,8 @@ public async Task<ItemResponse<T>> UpsertItemAsync<T>(
                 itemId: null,
                 item: item,
                 operationType: OperationType.Upsert,
-                requestOptions: requestOptions,
+                requestOptions: requestOptions,
+                targetRequestSerializationFormat: ContainerCore.GetTargetRequestSerializationFormat(),
                 trace: trace,
                 cancellationToken: cancellationToken);
 
@@ -178,7 +187,9 @@ public async Task<ResponseMessage> ReplaceItemStreamAsync(
                 operationType: OperationType.Replace,
                 requestOptions: requestOptions,
                 trace: trace,
-                cancellationToken: cancellationToken);
+                cancellationToken: cancellationToken,
+                targetRequestSerializationFormat: ContainerCore.GetTargetRequestSerializationFormat(),
+                targetResponseSerializationFormat: JsonSerializationFormat.Text);
         }
 
         public async Task<ItemResponse<T>> ReplaceItemAsync<T>(
@@ -204,7 +215,8 @@ public async Task<ItemResponse<T>> ReplaceItemAsync<T>(
                itemId: id,
                item: item,
                operationType: OperationType.Replace,
-               requestOptions: requestOptions,
+               requestOptions: requestOptions,
+               targetRequestSerializationFormat: ContainerCore.GetTargetRequestSerializationFormat(),
                trace: trace,
                cancellationToken: cancellationToken);
 
@@ -217,15 +229,17 @@ public async Task<ResponseMessage> DeleteItemStreamAsync(
             ITrace trace,
             ItemRequestOptions requestOptions = null,
             CancellationToken cancellationToken = default)
-        {
+        {
             return await this.ProcessItemStreamAsync(
                 partitionKey: partitionKey,
                 itemId: id,
                 streamPayload: null,
                 operationType: OperationType.Delete,
                 requestOptions: requestOptions,
                 trace: trace,
-                cancellationToken: cancellationToken);
+                cancellationToken: cancellationToken,
+                targetRequestSerializationFormat: ContainerCore.GetTargetRequestSerializationFormat(),
+                targetResponseSerializationFormat: JsonSerializationFormat.Text);
         }
 
         public async Task<ItemResponse<T>> DeleteItemAsync<T>(
@@ -240,7 +254,8 @@ public async Task<ItemResponse<T>> DeleteItemAsync<T>(
                 itemId: id,
                 streamPayload: null,
                 operationType: OperationType.Delete,
-                requestOptions: requestOptions,
+                requestOptions: requestOptions,
+                targetRequestSerializationFormat: ContainerCore.GetTargetRequestSerializationFormat(),
                 trace: trace,
                 cancellationToken: cancellationToken);
 
@@ -828,7 +843,8 @@ private async Task<ResponseMessage> ExtractPartitionKeyAndProcessItemStreamAsync
             string itemId,
             T item,
             OperationType operationType,
-            ItemRequestOptions requestOptions,
+            ItemRequestOptions requestOptions,
+            JsonSerializationFormat? targetRequestSerializationFormat,
             ITrace trace,
             CancellationToken cancellationToken)
         {
@@ -840,7 +856,7 @@ private async Task<ResponseMessage> ExtractPartitionKeyAndProcessItemStreamAsync
             Stream itemStream;
             using (trace.StartChild("ItemSerialize"))
             {
-                itemStream = this.ClientContext.SerializerCore.ToStream<T>(item);
+                itemStream = this.ClientContext.SerializerCore.ToStream<T>(item);
             }
 
             // User specified PK value, no need to extract it
@@ -853,7 +869,8 @@ private async Task<ResponseMessage> ExtractPartitionKeyAndProcessItemStreamAsync
                         operationType,
                         requestOptions,
                         trace: trace,
-                        cancellationToken: cancellationToken);
+                        cancellationToken: cancellationToken,
+                        targetRequestSerializationFormat: targetRequestSerializationFormat);
             }
 
             PartitionKeyMismatchRetryPolicy requestRetryPolicy = null;
@@ -866,7 +883,8 @@ private async Task<ResponseMessage> ExtractPartitionKeyAndProcessItemStreamAsync
                     itemId,
                     itemStream,
                     operationType,
-                    requestOptions,
+                    requestOptions,
+                    targetRequestSerializationFormat: ContainerCore.GetTargetRequestSerializationFormat(),
                     trace: trace,
                     cancellationToken: cancellationToken);
 
@@ -897,7 +915,9 @@ private async Task<ResponseMessage> ProcessItemStreamAsync(
             OperationType operationType,
             ItemRequestOptions requestOptions,
             ITrace trace,
-            CancellationToken cancellationToken)
+            CancellationToken cancellationToken,
+            JsonSerializationFormat? targetRequestSerializationFormat,
+            JsonSerializationFormat? targetResponseSerializationFormat = default)
         {
             if (trace == null)
             {
@@ -910,8 +930,18 @@ 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 (CosmosSerializationUtil.TrySerializeStreamToTargetFormat(
+                expectedSerializationFormat: JsonSerializationFormat.Text,
+                targetSerializationFormat: targetRequestSerializationFormat,
+                inputStream: streamPayload,
+                outputStream: out Stream outputRequestStream))
+            {
+                streamPayload = outputRequestStream;
+            }
+
             ResponseMessage responseMessage = await this.ClientContext.ProcessResourceOperationStreamAsync(
                 resourceUri: resourceUri,
                 resourceType: ResourceType.Document,
@@ -923,9 +953,20 @@ private async Task<ResponseMessage> ProcessItemStreamAsync(
                 streamPayload: streamPayload,
                 requestEnricher: null,
                 trace: trace,
-                cancellationToken: cancellationToken);
-
-            return responseMessage;
+                cancellationToken: cancellationToken);
+
+            // Convert Binary Stream to Text.
+            if ((requestOptions == null || !requestOptions.EnableBinaryResponseOnPointOperations)
+                && CosmosSerializationUtil.TrySerializeStreamToTargetFormat(
+                    expectedSerializationFormat: JsonSerializationFormat.Binary,
+                    targetSerializationFormat: targetResponseSerializationFormat,
+                    inputStream: responseMessage?.Content,
+                    outputStream: out Stream outputResponseStream))
+            {
+                responseMessage.Content = outputResponseStream;
+            }
+
+            return responseMessage;
         }
 
         public override async Task<PartitionKey> GetPartitionKeyValueFromStreamAsync(
@@ -1120,13 +1161,14 @@ public async Task<ItemResponse<T>> PatchItemAsync<T>(
             return this.ClientContext.ResponseFactory.CreateItemResponse<T>(responseMessage);
         }
 
-        public Task<ResponseMessage> PatchItemStreamAsync(
+        public async Task<ResponseMessage> PatchItemStreamAsync(
             string id,
             PartitionKey partitionKey,
             IReadOnlyList<PatchOperation> patchOperations,
             ITrace trace,
             PatchItemRequestOptions requestOptions = null,
-            CancellationToken cancellationToken = default)
+            CancellationToken cancellationToken = default,
+            JsonSerializationFormat? targetResponseSerializationFormat = default)
         {
             if (trace == null)
             {
@@ -1159,8 +1201,8 @@ public Task<ResponseMessage> PatchItemStreamAsync(
             {
                 patchOperationsStream = this.ClientContext.SerializerCore.ToStream(new PatchSpec(patchOperations, requestOptions));
             }
-
-            return this.ClientContext.ProcessResourceOperationStreamAsync(
+
+            ResponseMessage responseMessage = await this.ClientContext.ProcessResourceOperationStreamAsync(
                 resourceUri: this.GetResourceUri(
                     requestOptions,
                     OperationType.Patch,
@@ -1174,7 +1216,19 @@ public Task<ResponseMessage> PatchItemStreamAsync(
                 streamPayload: patchOperationsStream,
                 requestEnricher: null,
                 trace: trace,
-                cancellationToken: cancellationToken);
+                cancellationToken: cancellationToken);
+
+            if ((requestOptions == null || !requestOptions.EnableBinaryResponseOnPointOperations)
+                && CosmosSerializationUtil.TrySerializeStreamToTargetFormat(
+                    expectedSerializationFormat: JsonSerializationFormat.Binary,
+                    targetSerializationFormat: targetResponseSerializationFormat,
+                    inputStream: responseMessage?.Content,
+                    outputStream: out Stream outputResponseStream))
+            {
+                responseMessage.Content = outputResponseStream;
+            }
+
+            return responseMessage;
         }
 
         public Task<ResponseMessage> PatchItemStreamAsync(
@@ -1217,7 +1271,9 @@ public Task<ResponseMessage> PatchItemStreamAsync(
                 operationType: OperationType.Patch,
                 requestOptions: requestOptions,
                 trace: trace,
-                cancellationToken: cancellationToken);
+                cancellationToken: cancellationToken,
+                targetRequestSerializationFormat: ContainerCore.GetTargetRequestSerializationFormat(),
+                targetResponseSerializationFormat: JsonSerializationFormat.Text);
         }
 
         public override ChangeFeedProcessorBuilder GetChangeFeedProcessorBuilderWithAllVersionsAndDeletes<T>(
@@ -1252,6 +1308,13 @@ private ChangeFeedProcessorBuilder GetChangeFeedProcessorBuilderPrivate(
                 container: this,
                 changeFeedProcessor: changeFeedProcessor,
                 applyBuilderConfiguration: changeFeedProcessor.ApplyBuildConfiguration).WithChangeFeedMode(mode);
+        }
+
+        private static JsonSerializationFormat GetTargetRequestSerializationFormat()
+        {
+            return ConfigurationManager.IsBinaryEncodingEnabled()
+                ? JsonSerializationFormat.Binary
+                : JsonSerializationFormat.Text;
         }
     }
 }