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

@@ -4,28 +4,21 @@
 namespace Microsoft.Azure.Cosmos
 {
  using System.Collections.Generic;
- using System.IO;
  using System.Net;
- using Microsoft.Azure.Cosmos.CosmosElements;
- using Microsoft.Azure.Cosmos.Json;
- using Microsoft.Azure.Cosmos.Serializer;
 
  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 +40,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 +50,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.FromFeedStream<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

@@ -7,11 +7,9 @@ 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 +18,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;
 
@@ -63,6 +62,12 @@ internal T FromStream<T>(Stream stream)
  return serializer.FromStream<T>(stream);
  }
 
+ internal IReadOnlyList<T> FromFeedStream<T>(Stream stream)
+ {
+ CosmosSerializer serializer = this.GetSerializer<T>();
+ return serializer.FromStream<List<T>>(stream);
+ }
+
  internal Stream ToStream<T>(T input)
  {
  CosmosSerializer serializer = this.GetSerializer<T>();
@@ -102,19 +107,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)