Query and change feed: Add serialization optimization by using array instead of by item

Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/FeedManagement/RemainingWorkEstimatorCore.cs

@@ -175,9 +175,9 @@ private static IEnumerable<JObject> GetItemsFromResponse(ResponseMessage respons
                 return new Collection<JObject>();
             }
 
-            return CosmosContainerExtensions.DefaultJsonSerializer.FromFeedResponseStream<JObject>(
-                response.Content,
-                ResourceType.Document);
+            return CosmosFeedResponseSerializer.FromFeedResponseStream<JObject>(
+                CosmosContainerExtensions.DefaultJsonSerializer,
+                response.Content);
         }
 
         private async Task<long> GetRemainingWorkAsync(DocumentServiceLease existingLease, CancellationToken cancellationToken)

Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/FeedProcessing/FeedProcessorCore.cs

@@ -118,9 +118,9 @@ private Task DispatchChangesAsync(ResponseMessage response, CancellationToken ca
             IEnumerable<T> asFeedResponse;
             try
             {
-                asFeedResponse = this.serializerCore.FromFeedResponseStream<T>(
-                    response.Content,
-                    Documents.ResourceType.Document);
+                asFeedResponse = CosmosFeedResponseSerializer.FromFeedResponseStream<T>(
+                    this.serializerCore,
+                    response.Content);
             }
             catch (Exception serializationException)
             {

Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/LeaseManagement/DocumentServiceLeaseContainerCosmos.cs

@@ -58,9 +58,9 @@ private async Task<IReadOnlyList<DocumentServiceLeaseCore>> ListDocumentsAsync(s
                 using (ResponseMessage responseMessage = await iterator.ReadNextAsync().ConfigureAwait(false))
                 {
                     responseMessage.EnsureSuccessStatusCode();
-                    leases.AddRange(CosmosContainerExtensions.DefaultJsonSerializer.FromFeedResponseStream<DocumentServiceLeaseCore>(
-                        responseMessage.Content,
-                        Documents.ResourceType.Document));
+                    leases.AddRange(CosmosFeedResponseSerializer.FromFeedResponseStream<DocumentServiceLeaseCore>(
+                        CosmosContainerExtensions.DefaultJsonSerializer,
+                        responseMessage.Content));
                 }   
             }
 

Microsoft.Azure.Cosmos/src/Query/v3Query/ReadFeedResponse.cs

@@ -14,18 +14,15 @@ internal class ReadFeedResponse<T> : FeedResponse<T>
     {
         protected ReadFeedResponse(
             HttpStatusCode httpStatusCode,
-            CosmosArray cosmosArray,
-            CosmosSerializerCore serializerCore,
+            IReadOnlyList<T> resources,
             Headers responseMessageHeaders,
             CosmosDiagnostics diagnostics)
         {
-            this.Count = cosmosArray != null ? cosmosArray.Count : 0;
+            this.Count = resources != null ? resources.Count : 0;
             this.Headers = responseMessageHeaders;
             this.StatusCode = httpStatusCode;
             this.Diagnostics = diagnostics;
-            this.Resource = CosmosElementSerializer.GetResources<T>(
-                cosmosArray: cosmosArray,
-                serializerCore: serializerCore);
+            this.Resource = resources;
         }
 
         public override int Count { get; }
@@ -47,8 +44,7 @@ public override IEnumerator<T> GetEnumerator()
 
         internal static ReadFeedResponse<TInput> CreateResponse<TInput>(
             ResponseMessage responseMessage,
-            CosmosSerializerCore serializerCore,
-            Documents.ResourceType resourceType)
+            CosmosSerializerCore serializerCore)
         {
             using (responseMessage)
             {
@@ -58,19 +54,13 @@ internal static ReadFeedResponse<TInput> CreateResponse<TInput>(
                     responseMessage.EnsureSuccessStatusCode();
                 }
 
-                CosmosArray cosmosArray = null;
-                if (responseMessage.Content != null)
-                {
-                    cosmosArray = CosmosElementSerializer.ToCosmosElements(
-                        responseMessage.Content,
-                        resourceType,
-                        null);
-                }
+                IReadOnlyList<TInput> resources = CosmosFeedResponseSerializer.FromFeedResponseStream<TInput>(
+                        serializerCore,
+                        responseMessage.Content);
 
                 ReadFeedResponse<TInput> readFeedResponse = new ReadFeedResponse<TInput>(
                     httpStatusCode: responseMessage.StatusCode,
-                    cosmosArray: cosmosArray,
-                    serializerCore: serializerCore,
+                    resources: resources,
                     responseMessageHeaders: responseMessage.Headers,
                     diagnostics: responseMessage.Diagnostics);
 

Microsoft.Azure.Cosmos/src/Resource/CosmosResponseFactoryCore.cs

@@ -31,17 +31,7 @@ public override FeedResponse<T> CreateChangeFeedUserTypeResponse<T>(
             ResponseMessage responseMessage)
         {
             return this.CreateChangeFeedResponseHelper<T>(
-                responseMessage,
-                Documents.ResourceType.Document);
-        }
-
-        public override FeedResponse<T> CreateChangeFeedUserTypeResponse<T>(
-            ResponseMessage responseMessage,
-            Documents.ResourceType resourceType)
-        {
-            return this.CreateChangeFeedResponseHelper<T>(
-                responseMessage,
-                resourceType);
+                responseMessage);
         }
 
         public override FeedResponse<T> CreateQueryFeedUserTypeResponse<T>(
@@ -75,18 +65,15 @@ private FeedResponse<T> CreateQueryFeedResponseHelper<T>(
 
             return ReadFeedResponse<T>.CreateResponse<T>(
                        cosmosResponseMessage,
-                       this.serializerCore,
-                       resourceType);
+                       this.serializerCore);
         }
 
         private FeedResponse<T> CreateChangeFeedResponseHelper<T>(
-            ResponseMessage cosmosResponseMessage,
-            Documents.ResourceType resourceType)
+            ResponseMessage cosmosResponseMessage)
         {
             return ReadFeedResponse<T>.CreateResponse<T>(
                        cosmosResponseMessage,
-                       this.serializerCore,
-                       resourceType);
+                       this.serializerCore);
         }
 
         public override ItemResponse<T> CreateItemResponse<T>(

Microsoft.Azure.Cosmos/src/Resource/CosmosResponseFactoryInternal.cs

@@ -12,10 +12,6 @@ internal abstract class CosmosResponseFactoryInternal : CosmosResponseFactory
         public abstract FeedResponse<T> CreateChangeFeedUserTypeResponse<T>(
             ResponseMessage responseMessage);
 
-        public abstract FeedResponse<T> CreateChangeFeedUserTypeResponse<T>(
-            ResponseMessage responseMessage,
-            Documents.ResourceType resourceType);
-
         public abstract FeedResponse<T> CreateQueryFeedUserTypeResponse<T>(
             ResponseMessage responseMessage);
 

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

@@ -22,30 +22,6 @@ namespace Microsoft.Azure.Cosmos.Serializer
 #endif
     static class CosmosElementSerializer
     {
-        /// <summary>
-        /// Converts a list of CosmosElements into a memory stream.
-        /// </summary>
-        /// <param name="stream">The stream response from Azure Cosmos</param>
-        /// <param name="resourceType">The resource type</param>
-        /// <param name="cosmosSerializationOptions">The custom serialization options. This allows custom serialization types like BSON, JSON, or other formats</param>
-        /// <returns>Returns a memory stream of cosmos elements. By default the memory stream will contain JSON.</returns>
-        internal static CosmosArray ToCosmosElements(
-            Stream stream,
-            ResourceType resourceType,
-            CosmosSerializationFormatOptions cosmosSerializationOptions = null)
-        {
-            MemoryStream memoryStream = stream as MemoryStream;
-            if (memoryStream == null)
-            {
-                memoryStream = new MemoryStream();
-                stream.CopyTo(memoryStream);
-            }
-
-            return CosmosElementSerializer.ToCosmosElements(
-                memoryStream,
-                resourceType,
-                cosmosSerializationOptions);
-        }
         /// <summary>
         /// Converts a list of CosmosElements into a memory stream.
         /// </summary>
@@ -284,7 +260,7 @@ private static IEnumerable<T> GetResourcesHelper<T>(
         /// <param name="cosmosElement">The cosmos elements</param>
         /// <param name="cosmosSerializationOptions">The custom serialization options. This allows custom serialization types like BSON, JSON, or other formats</param>
         /// <returns>Returns a memory stream of cosmos elements. By default the memory stream will contain JSON.</returns>
-        private static MemoryStream ElementToMemoryStream(
+        internal static MemoryStream ElementToMemoryStream(
             CosmosElement cosmosElement,
             CosmosSerializationFormatOptions cosmosSerializationOptions = null)
         {

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

@@ -0,0 +1,118 @@
+//------------------------------------------------------------
+// Copyright (c) Microsoft Corporation.  All rights reserved.
+//------------------------------------------------------------
+
+namespace Microsoft.Azure.Cosmos
+{
+    using System;
+    using System.Collections.Generic;
+    using System.IO;
+    using System.Runtime.InteropServices;
+    using System.Text;
+
+#nullable enable
+    internal static class CosmosFeedResponseSerializer
+    {
+        private static readonly byte ArrayStart = Encoding.UTF8.GetBytes("[")[0];
+        private static readonly byte ArrayEnd = Encoding.UTF8.GetBytes("]")[0];
+
+        /// <summary>
+        /// The service returns feed responses in an envelope. This removes the envelope
+        /// and serializes all the items into a list
+        /// </summary>
+        /// <param name="serializerCore">The cosmos serializer</param>
+        /// <param name="streamWithServiceEnvelope">A stream with the service envelope like: { "ContainerRid":"Test", "Documents":[{ "id":"MyItem"}], "count":1}</param>
+        /// <returns>A read only list of the serialized items</returns>
+        internal static IReadOnlyList<T> FromFeedResponseStream<T>(
+            CosmosSerializerCore serializerCore,
+            Stream streamWithServiceEnvelope)
+        {
+            if (streamWithServiceEnvelope == null)
+            {
+                return new List<T>();
+            }
+
+            using (streamWithServiceEnvelope)
+            {
+                using (MemoryStream stream = GetStreamWithoutServiceEnvelope(
+                                streamWithServiceEnvelope))
+                {
+                    return serializerCore.FromStream<List<T>>(stream);
+                }
+            }
+        }
+
+        /// <summary>
+        /// The service returns feed responses in an envelope. This removes the envelope
+        /// so it only returns the array of items.
+        /// </summary>
+        /// <param name="streamWithServiceEnvelope">A stream with the service envelope like: { "ContainerRid":"Test", "Documents":[{ "id":"MyItem"}], "count":1}</param>
+        /// <returns>A stream containing only the array of items</returns>
+        internal static MemoryStream GetStreamWithoutServiceEnvelope(
+            Stream streamWithServiceEnvelope)
+        {
+            using (streamWithServiceEnvelope)
+            {
+                ReadOnlyMemory<byte> content;
+                MemoryStream? memoryStreamWithEnvelope = streamWithServiceEnvelope as MemoryStream;
+                if (memoryStreamWithEnvelope == null)
+                {
+                    memoryStreamWithEnvelope = new MemoryStream();
+                    streamWithServiceEnvelope.CopyTo(memoryStreamWithEnvelope);
+                    memoryStreamWithEnvelope.Position = 0;
+                }
+
+                if (memoryStreamWithEnvelope.TryGetBuffer(out ArraySegment<byte> buffer))
+                {
+                    content = buffer;
+                }
+                else
+                {
+                    content = memoryStreamWithEnvelope.ToArray();
+                }
+
+                int start = GetArrayStartPosition(content);
+                int end = GetArrayEndPosition(content);
+
+                ReadOnlyMemory<byte> spanwithOnlyArray = content.Slice(start, end - start + 1);
+                if (!MemoryMarshal.TryGetArray(spanwithOnlyArray, out ArraySegment<byte> resultAsArray))
+                {
+                    resultAsArray = new ArraySegment<byte>(spanwithOnlyArray.ToArray());
+                }
+
+                MemoryStream arrayOnlyStream = new MemoryStream(resultAsArray.Array, resultAsArray.Offset, resultAsArray.Count, writable: false, publiclyVisible: true);
+                return arrayOnlyStream;
+            }
+        }
+
+        private static int GetArrayStartPosition(
+            ReadOnlyMemory<byte> memoryByte)
+        {
+            ReadOnlySpan<byte> span = memoryByte.Span;
+            for (int i = 0; i < span.Length; i++)
+            {
+                if (span[i] == ArrayStart)
+                {
+                    return i;
+                }
+            }
+
+            throw new ArgumentException("ArrayStart not found");
+        }
+
+        private static int GetArrayEndPosition(
+            ReadOnlyMemory<byte> memoryByte)
+        {
+            ReadOnlySpan<byte> span = memoryByte.Span;
+            for (int i = span.Length - 1; i < span.Length; i--)
+            {
+                if (span[i] == ArrayEnd)
+                {
+                    return i;
+                }
+            }
+
+            throw new ArgumentException("ArrayEnd not found");
+        }
+    }
+}

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

@@ -5,13 +5,10 @@
 namespace Microsoft.Azure.Cosmos
 {
     using System;
-    using System.Collections.Generic;
     using System.IO;
-    using Microsoft.Azure.Cosmos.CosmosElements;
     using Microsoft.Azure.Cosmos.Query.Core;
     using Microsoft.Azure.Cosmos.Query.Core.QueryPlan;
     using Microsoft.Azure.Cosmos.Scripts;
-    using Microsoft.Azure.Cosmos.Serializer;
     using Microsoft.Azure.Documents;
 
     /// <summary>
@@ -20,6 +17,7 @@ namespace Microsoft.Azure.Cosmos
     internal class CosmosSerializerCore
     {
         private static readonly CosmosSerializer propertiesSerializer = new CosmosJsonSerializerWrapper(new CosmosJsonDotNetSerializer());
+
         private readonly CosmosSerializer customSerializer;
         private readonly CosmosSerializer sqlQuerySpecSerializer;
 
@@ -102,19 +100,6 @@ internal CosmosSerializer GetCustomOrDefaultSerializer()
             return CosmosSerializerCore.propertiesSerializer;
         }
 
-        internal IEnumerable<T> FromFeedResponseStream<T>(
-            Stream stream,
-            ResourceType resourceType)
-        {
-            CosmosArray cosmosArray = CosmosElementSerializer.ToCosmosElements(
-                    stream,
-                    resourceType);
-
-            return CosmosElementSerializer.GetResources<T>(
-               cosmosArray: cosmosArray,
-               serializerCore: this);
-        }
-
         private CosmosSerializer GetSerializer<T>()
         {
             if (this.customSerializer == null)