Avoid array copy on Avro deserialization (#2483)

Author: Havret

src/Confluent.SchemaRegistry.Serdes.Avro/GenericDeserializerImpl.cs

@@ -58,8 +58,8 @@ public override async Task<GenericRecord> DeserializeAsync(ReadOnlyMemory<byte>
                 : await Deserialize(context.Topic, context.Headers, data,
                     context.Component == MessageComponentType.Key).ConfigureAwait(false);
         }
-        
-        public async Task<GenericRecord> Deserialize(string topic, Headers headers, ReadOnlyMemory<byte> array, bool isKey)
+
+        private async Task<GenericRecord> Deserialize(string topic, Headers headers, ReadOnlyMemory<byte> array, bool isKey)
         {
             try
             {
@@ -105,7 +105,7 @@ public async Task<GenericRecord> Deserialize(string topic, Headers headers, Read
                 DatumReader<GenericRecord> datumReader;
                 if (migrations.Count > 0)
                 {
-                    using (var stream = new MemoryStream(payload.ToArray()))
+                    using (var stream = new ReadOnlyMemoryStream(payload))
                     {
                         data = new GenericReader<GenericRecord>(writerSchema, writerSchema)
                             .Read(default(GenericRecord), new BinaryDecoder(stream));
@@ -149,7 +149,7 @@ public async Task<GenericRecord> Deserialize(string topic, Headers headers, Read
                     }
                     datumReader = await GetDatumReader(writerSchema, readerSchema).ConfigureAwait(false);
 
-                    using (var stream = new MemoryStream(payload.ToArray()))
+                    using (var stream = new ReadOnlyMemoryStream(payload))
                     {
                         data = datumReader.Read(default(GenericRecord), new BinaryDecoder(stream));
                     }

src/Confluent.SchemaRegistry.Serdes.Avro/ReadOnlyMemoryStream.cs

@@ -0,0 +1,95 @@
+// Copyright 2025 Confluent Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// Refer to LICENSE for more information.
+
+using System;
+using System.IO;
+
+namespace Confluent.SchemaRegistry.Serdes;
+
+/// <summary>
+/// A read-only <see cref="Stream"/> implementation over a <see cref="ReadOnlyMemory{Byte}"/>.
+/// This class is needed to avoid calling <c>ToArray()</c> on <see cref="ReadOnlyMemory{Byte}"/>
+/// when passing data to APIs (such as Avro's <c>BinaryDecoder</c>) that require a <see cref="Stream"/>.
+/// </summary>
+internal class ReadOnlyMemoryStream : Stream
+{
+    private readonly ReadOnlyMemory<byte> data;
+
+    public ReadOnlyMemoryStream(ReadOnlyMemory<byte> data)
+    {
+        this.data = data;
+    }
+
+    public override void Flush()
+    {
+    }
+
+    public override int Read(byte[] buffer, int offset, int count)
+    {
+        if (buffer == null)
+            throw new ArgumentNullException(nameof(buffer));
+        if (offset < 0 || count < 0 || offset + count > buffer.Length)
+            throw new ArgumentOutOfRangeException();
+
+        var remaining = data.Length - (int) Position;
+        if (remaining <= 0)
+            return 0;
+
+        var toRead = Math.Min(count, remaining);
+        data.Slice((int) Position, toRead).Span.CopyTo(buffer.AsSpan(offset, toRead));
+        Position += toRead;
+        return toRead;
+    }
+
+    public override long Seek(long offset, SeekOrigin origin)
+    {
+        switch (origin)
+        {
+            case SeekOrigin.Begin:
+                Position = offset;
+                break;
+            case SeekOrigin.Current:
+                Position += offset;
+                break;
+            case SeekOrigin.End:
+                Position = Length + offset;
+                break;
+            default:
+                throw new ArgumentOutOfRangeException(nameof(origin), origin, null);
+        }
+
+        if (Position < 0 || Position > Length)
+            throw new IOException("Seek operation resulted in an invalid position.");
+
+        return Position;
+    }
+
+    public override void SetLength(long value)
+    {
+        throw new NotSupportedException();
+    }
+
+    public override void Write(byte[] buffer, int offset, int count)
+    {
+        throw new NotSupportedException();
+    }
+
+    public override bool CanRead => true;
+    public override bool CanSeek => true;
+    public override bool CanWrite => false;
+    public override long Length => data.Length;
+    public override long Position { get; set; }
+}

src/Confluent.SchemaRegistry.Serdes.Avro/SpecificDeserializerImpl.cs

@@ -154,9 +154,7 @@ public async Task<T> Deserialize(string topic, Headers headers, ReadOnlyMemory<b
                 DatumReader<T> datumReader = null;
                 if (migrations.Count > 0)
                 {
-                    // TODO: We should be able to write a wrapper around ReadOnlyMemory<byte> that allows us to
-                    // pass it to the BinaryDecoder without copying the data to a MemoryStream.
-                    using (var memoryStream = new MemoryStream(payload.ToArray()))
+                    using (var memoryStream = new ReadOnlyMemoryStream(payload))
                     {
                         data = new GenericReader<GenericRecord>(writerSchema, writerSchema)
                             .Read(default(GenericRecord), new BinaryDecoder(memoryStream));
@@ -187,10 +185,8 @@ public async Task<T> Deserialize(string topic, Headers headers, ReadOnlyMemory<b
                 else
                 {
                     datumReader = await GetDatumReader(writerSchema, ReaderSchema).ConfigureAwait(false);
-
-                    // TODO: We should be able to write a wrapper around ReadOnlyMemory<byte> that allows us to
-                    // pass it to the BinaryDecoder without copying the data to a MemoryStream.
-                    using var stream = new MemoryStream(payload.ToArray());
+                    
+                    using var stream = new ReadOnlyMemoryStream(payload);
                     data = Read(datumReader, new BinaryDecoder(stream));
                 }