AvroEncoders.cs

Go to the documentation of this file.

using System;
using System.Buffers;
using System.Buffers.Binary;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Linq.Expressions;
using System.Reflection;
using System.Runtime.CompilerServices;
using System.Text;
using Avro;

namespace kinetica;

#region Avro Varint Encoding Utilities

    internal static class AvroEncoding
    {
        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public static int WriteVarInt(byte[] buffer, int position, int value)
        {
            return WriteVarLong(buffer, position, value);
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public static int WriteVarLong(byte[] buffer, int position, long value)
        {
            // Zig-zag encoding
            ulong n = (ulong)((value << 1) ^ (value >> 63));

            while ((n & ~0x7FUL) != 0)
            {
                buffer[position++] = (byte)((n & 0x7F) | 0x80);
                n >>= 7;
            }
            buffer[position++] = (byte)n;

            return position;
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public static int GetVarIntSize(long value)
        {
            ulong n = (ulong)((value << 1) ^ (value >> 63));
            int size = 1;
            while ((n & ~0x7FUL) != 0)
            {
                size++;
                n >>= 7;
            }
            return size;
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public static void EnsureCapacity(ref byte[] buffer, int position, int required)
        {
            if (position + required > buffer.Length)
            {
                var newBuffer = new byte[Math.Max(buffer.Length * 2, position + required + 1024)];
                Buffer.BlockCopy(buffer, 0, newBuffer, 0, position);
                buffer = newBuffer;
            }
        }
    }

    #endregion

    #region Direct Avro Record Encoder

    internal sealed class DirectAvroEncoder<T>
    {
        private readonly FieldEncoder[] _fieldEncoders;
        private readonly int _fieldCount;
        private readonly int _estimatedRecordSize;

        // Cache of encoders by type
        private static readonly ConcurrentDictionary<(Type, string), DirectAvroEncoder<T>> _encoderCache = new();

        public static DirectAvroEncoder<T> GetOrCreate(KineticaType ktype)
        {
            var schemaString = ktype.getSchemaString();
            var key = (typeof(T), schemaString);

            return _encoderCache.GetOrAdd(key, _ => new DirectAvroEncoder<T>(ktype));
        }

        private DirectAvroEncoder(KineticaType ktype)
        {
            var schema = ktype.getSchema() as RecordSchema
                ?? throw new ArgumentException("Invalid schema: not a record schema");

            _fieldCount = schema.Fields.Count;
            _fieldEncoders = new FieldEncoder[_fieldCount];
            _estimatedRecordSize = 64; // Initial estimate, grows as needed

            var type = typeof(T);
            var properties = type.GetProperties(BindingFlags.Public | BindingFlags.Instance);
            var propertyMap = properties.ToDictionary(
                p => p.Name.ToLowerInvariant(),
                p => p,
                StringComparer.OrdinalIgnoreCase);

            for (int i = 0; i < _fieldCount; i++)
            {
                var field = schema.Fields[i];
                var fieldNameLower = field.Name.ToLowerInvariant();

                if (propertyMap.TryGetValue(fieldNameLower, out var property))
                {
                    _fieldEncoders[i] = CreateFieldEncoder(field.Schema, property);
                }
                else
                {
                    // Property not found - create null encoder
                    _fieldEncoders[i] = new NullFieldEncoder();
                }
            }
        }

        private FieldEncoder CreateFieldEncoder(Schema fieldSchema, PropertyInfo property)
        {
            // Handle union types (nullable)
            if (fieldSchema is UnionSchema unionSchema)
            {
                // Find the non-null schema in the union
                Schema? nonNullSchema = null;
                int nonNullIndex = -1;
                for (int i = 0; i < unionSchema.Schemas.Count; i++)
                {
                    if (unionSchema.Schemas[i].Tag != Schema.Type.Null)
                    {
                        nonNullSchema = unionSchema.Schemas[i];
                        nonNullIndex = i;
                        break;
                    }
                }

                if (nonNullSchema != null)
                {
                    return CreateNullableFieldEncoder(nonNullSchema, property, nonNullIndex);
                }
            }

            return CreateNonNullableFieldEncoder(fieldSchema, property);
        }

        private FieldEncoder CreateNullableFieldEncoder(Schema schema, PropertyInfo property, int nonNullIndex)
        {
            var propertyType = property.PropertyType;
            var underlyingType = Nullable.GetUnderlyingType(propertyType) ?? propertyType;

            return schema.Tag switch
            {
                Schema.Type.Int => new NullableIntEncoder(CreateGetter<int?>(property), nonNullIndex),
                Schema.Type.Long => new NullableLongEncoder(CreateGetter<long?>(property), nonNullIndex),
                Schema.Type.Float => new NullableFloatEncoder(CreateGetter<float?>(property), nonNullIndex),
                Schema.Type.Double => new NullableDoubleEncoder(CreateGetter<double?>(property), nonNullIndex),
                Schema.Type.Boolean => new NullableBoolEncoder(CreateGetter<bool?>(property), nonNullIndex),
                Schema.Type.String => new NullableStringEncoder(CreateGetter<string?>(property), nonNullIndex),
                Schema.Type.Bytes => new NullableBytesEncoder(CreateGetter<byte[]?>(property), nonNullIndex),
                _ => new NullFieldEncoder()
            };
        }

        private FieldEncoder CreateNonNullableFieldEncoder(Schema schema, PropertyInfo property)
        {
            return schema.Tag switch
            {
                Schema.Type.Int => new IntEncoder(CreateGetter<int>(property)),
                Schema.Type.Long => new LongEncoder(CreateGetter<long>(property)),
                Schema.Type.Float => new FloatEncoder(CreateGetter<float>(property)),
                Schema.Type.Double => new DoubleEncoder(CreateGetter<double>(property)),
                Schema.Type.Boolean => new BoolEncoder(CreateGetter<bool>(property)),
                Schema.Type.String => new StringEncoder(CreateGetter<string>(property)),
                Schema.Type.Bytes => new BytesEncoder(CreateGetter<byte[]>(property)),
                _ => new NullFieldEncoder()
            };
        }

        private static Func<T, TValue> CreateGetter<TValue>(PropertyInfo property)
        {
            var parameter = Expression.Parameter(typeof(T), "obj");
            Expression propertyAccess = Expression.Property(parameter, property);

            // Handle type conversions
            if (propertyAccess.Type != typeof(TValue))
            {
                propertyAccess = Expression.Convert(propertyAccess, typeof(TValue));
            }

            return Expression.Lambda<Func<T, TValue>>(propertyAccess, parameter).Compile();
        }

        // Thread-local buffer to avoid allocations - each thread gets its own resizable buffer
        [ThreadStatic]
        private static byte[]? t_buffer;

        public byte[] Encode(T record)
        {
            // Get or create thread-local buffer (start with larger buffer for large strings)
            var buffer = t_buffer ?? (t_buffer = new byte[4096]);
            int position = 0;

            for (int i = 0; i < _fieldCount; i++)
            {
                // Encoders handle their own capacity checks and can resize buffer via ref
                position = _fieldEncoders[i].Encode(record, ref buffer, position);
            }

            // Update thread-local buffer if it was resized
            t_buffer = buffer;

            // Allocate exact-sized result array (unavoidable - this is what goes in the request)
            var result = new byte[position];
            Buffer.BlockCopy(buffer, 0, result, 0, position);
            return result;
        }

        public int EncodeTo(T record, ref byte[] buffer, int offset)
        {
            int position = offset;

            for (int i = 0; i < _fieldCount; i++)
            {
                position = _fieldEncoders[i].Encode(record, ref buffer, position);
            }

            return position - offset;
        }

        public int CalculateEncodedSize(T record)
        {
            // Use thread-local buffer
            var buffer = t_buffer ?? (t_buffer = new byte[4096]);
            var size = EncodeTo(record, ref buffer, 0);
            t_buffer = buffer; // Update if resized
            return size;
        }

        public byte[][] EncodeMany(IReadOnlyList<T> records)
        {
            var count = records.Count;
            var results = new byte[count][];

            if (count < 100)
            {
                for (int i = 0; i < count; i++)
                {
                    results[i] = Encode(records[i]);
                }
            }
            else
            {
                System.Threading.Tasks.Parallel.For(0, count, i =>
                {
                    results[i] = Encode(records[i]);
                });
            }

            return results;
        }

        public List<byte[]> EncodeManyAsList(IReadOnlyList<T> records)
        {
            return new List<byte[]>(EncodeMany(records));
        }

        public (byte[] Buffer, (int Offset, int Length)[] Segments) EncodeManyZeroCopy(IReadOnlyList<T> records)
        {
            var count = records.Count;
            if (count == 0)
            {
                return (Array.Empty<byte>(), Array.Empty<(int, int)>());
            }

            // Estimate total size (will grow if needed via ref in encoders)
            int estimatedTotalSize = count * _estimatedRecordSize;
            var buffer = new byte[Math.Max(estimatedTotalSize, 4096)];
            var segments = new (int Offset, int Length)[count];
            int position = 0;

            for (int i = 0; i < count; i++)
            {
                int startPos = position;

                // Encode directly into the shared buffer (encoders handle capacity via ref)
                for (int f = 0; f < _fieldCount; f++)
                {
                    position = _fieldEncoders[f].Encode(records[i], ref buffer, position);
                }

                segments[i] = (startPos, position - startPos);
            }

            // Trim buffer to exact size if significantly oversized
            if (buffer.Length > position * 2)
            {
                var trimmed = new byte[position];
                Buffer.BlockCopy(buffer, 0, trimmed, 0, position);
                buffer = trimmed;
            }

            return (buffer, segments);
        }

        public (byte[] Buffer, (int Offset, int Length)[] Segments) EncodeManyZeroCopyParallel(IReadOnlyList<T> records)
        {
            var count = records.Count;
            if (count == 0)
            {
                return (Array.Empty<byte>(), Array.Empty<(int, int)>());
            }

            // For small batches, use sequential zero-copy
            if (count < 100)
            {
                return EncodeManyZeroCopy(records);
            }

            // Phase 1: Parallel encode to individual arrays (uses thread-local buffers)
            var encodedRecords = new byte[count][];
            System.Threading.Tasks.Parallel.For(0, count, i =>
            {
                encodedRecords[i] = Encode(records[i]);
            });

            // Phase 2: Calculate total size and allocate single buffer
            int totalSize = 0;
            for (int i = 0; i < count; i++)
            {
                totalSize += encodedRecords[i].Length;
            }

            var buffer = new byte[totalSize];
            var segments = new (int Offset, int Length)[count];
            int position = 0;

            // Phase 3: Copy into contiguous buffer (fast sequential memcpy)
            for (int i = 0; i < count; i++)
            {
                var encoded = encodedRecords[i];
                segments[i] = (position, encoded.Length);
                Buffer.BlockCopy(encoded, 0, buffer, position, encoded.Length);
                position += encoded.Length;
            }

            return (buffer, segments);
        }

        #region Field Encoders

        private abstract class FieldEncoder
        {
            public abstract int Encode(T record, ref byte[] buffer, int position);
        }

        private sealed class NullFieldEncoder : FieldEncoder
        {
            public override int Encode(T record, ref byte[] buffer, int position) => position;
        }

        // Non-nullable encoders
        private sealed class IntEncoder : FieldEncoder
        {
            private readonly Func<T, int> _getter;
            public IntEncoder(Func<T, int> getter) => _getter = getter;

            public override int Encode(T record, ref byte[] buffer, int position)
            {
                AvroEncoding.EnsureCapacity(ref buffer, position, 10);
                return AvroEncoding.WriteVarInt(buffer, position, _getter(record));
            }
        }

        private sealed class LongEncoder : FieldEncoder
        {
            private readonly Func<T, long> _getter;
            public LongEncoder(Func<T, long> getter) => _getter = getter;

            public override int Encode(T record, ref byte[] buffer, int position)
            {
                AvroEncoding.EnsureCapacity(ref buffer, position, 10);
                return AvroEncoding.WriteVarLong(buffer, position, _getter(record));
            }
        }

        private sealed class FloatEncoder : FieldEncoder
        {
            private readonly Func<T, float> _getter;
            public FloatEncoder(Func<T, float> getter) => _getter = getter;

            public override int Encode(T record, ref byte[] buffer, int position)
            {
                AvroEncoding.EnsureCapacity(ref buffer, position, 4);
                BinaryPrimitives.WriteSingleLittleEndian(buffer.AsSpan(position), _getter(record));
                return position + 4;
            }
        }

        private sealed class DoubleEncoder : FieldEncoder
        {
            private readonly Func<T, double> _getter;
            public DoubleEncoder(Func<T, double> getter) => _getter = getter;

            public override int Encode(T record, ref byte[] buffer, int position)
            {
                AvroEncoding.EnsureCapacity(ref buffer, position, 8);
                BinaryPrimitives.WriteDoubleLittleEndian(buffer.AsSpan(position), _getter(record));
                return position + 8;
            }
        }

        private sealed class BoolEncoder : FieldEncoder
        {
            private readonly Func<T, bool> _getter;
            public BoolEncoder(Func<T, bool> getter) => _getter = getter;

            public override int Encode(T record, ref byte[] buffer, int position)
            {
                AvroEncoding.EnsureCapacity(ref buffer, position, 1);
                buffer[position] = _getter(record) ? (byte)1 : (byte)0;
                return position + 1;
            }
        }

        private sealed class StringEncoder : FieldEncoder
        {
            private readonly Func<T, string> _getter;
            public StringEncoder(Func<T, string> getter) => _getter = getter;

            public override int Encode(T record, ref byte[] buffer, int position)
            {
                var value = _getter(record);
                if (value == null)
                {
                    AvroEncoding.EnsureCapacity(ref buffer, position, 1);
                    return AvroEncoding.WriteVarLong(buffer, position, 0);
                }

                var bytes = Encoding.UTF8.GetBytes(value);
                // Ensure buffer has enough capacity for length prefix + string bytes
                AvroEncoding.EnsureCapacity(ref buffer, position, bytes.Length + 10);
                position = AvroEncoding.WriteVarLong(buffer, position, bytes.Length);
                Buffer.BlockCopy(bytes, 0, buffer, position, bytes.Length);
                return position + bytes.Length;
            }
        }

        private sealed class BytesEncoder : FieldEncoder
        {
            private readonly Func<T, byte[]> _getter;
            public BytesEncoder(Func<T, byte[]> getter) => _getter = getter;

            public override int Encode(T record, ref byte[] buffer, int position)
            {
                var value = _getter(record);
                if (value == null)
                {
                    AvroEncoding.EnsureCapacity(ref buffer, position, 1);
                    return AvroEncoding.WriteVarLong(buffer, position, 0);
                }

                AvroEncoding.EnsureCapacity(ref buffer, position, value.Length + 10);
                position = AvroEncoding.WriteVarLong(buffer, position, value.Length);
                Buffer.BlockCopy(value, 0, buffer, position, value.Length);
                return position + value.Length;
            }
        }

        // Nullable encoders (for union types)
        private sealed class NullableIntEncoder : FieldEncoder
        {
            private readonly Func<T, int?> _getter;
            private readonly int _nonNullIndex;
            public NullableIntEncoder(Func<T, int?> getter, int nonNullIndex) { _getter = getter; _nonNullIndex = nonNullIndex; }

            public override int Encode(T record, ref byte[] buffer, int position)
            {
                AvroEncoding.EnsureCapacity(ref buffer, position, 20);
                var value = _getter(record);
                if (!value.HasValue)
                {
                    return AvroEncoding.WriteVarLong(buffer, position, _nonNullIndex == 0 ? 1 : 0);
                }
                position = AvroEncoding.WriteVarLong(buffer, position, _nonNullIndex);
                return AvroEncoding.WriteVarInt(buffer, position, value.Value);
            }
        }

        private sealed class NullableLongEncoder : FieldEncoder
        {
            private readonly Func<T, long?> _getter;
            private readonly int _nonNullIndex;
            public NullableLongEncoder(Func<T, long?> getter, int nonNullIndex) { _getter = getter; _nonNullIndex = nonNullIndex; }

            public override int Encode(T record, ref byte[] buffer, int position)
            {
                AvroEncoding.EnsureCapacity(ref buffer, position, 20);
                var value = _getter(record);
                if (!value.HasValue)
                {
                    return AvroEncoding.WriteVarLong(buffer, position, _nonNullIndex == 0 ? 1 : 0);
                }
                position = AvroEncoding.WriteVarLong(buffer, position, _nonNullIndex);
                return AvroEncoding.WriteVarLong(buffer, position, value.Value);
            }
        }

        private sealed class NullableFloatEncoder : FieldEncoder
        {
            private readonly Func<T, float?> _getter;
            private readonly int _nonNullIndex;
            public NullableFloatEncoder(Func<T, float?> getter, int nonNullIndex) { _getter = getter; _nonNullIndex = nonNullIndex; }

            public override int Encode(T record, ref byte[] buffer, int position)
            {
                AvroEncoding.EnsureCapacity(ref buffer, position, 14);
                var value = _getter(record);
                if (!value.HasValue)
                {
                    return AvroEncoding.WriteVarLong(buffer, position, _nonNullIndex == 0 ? 1 : 0);
                }
                position = AvroEncoding.WriteVarLong(buffer, position, _nonNullIndex);
                BinaryPrimitives.WriteSingleLittleEndian(buffer.AsSpan(position), value.Value);
                return position + 4;
            }
        }

        private sealed class NullableDoubleEncoder : FieldEncoder
        {
            private readonly Func<T, double?> _getter;
            private readonly int _nonNullIndex;
            public NullableDoubleEncoder(Func<T, double?> getter, int nonNullIndex) { _getter = getter; _nonNullIndex = nonNullIndex; }

            public override int Encode(T record, ref byte[] buffer, int position)
            {
                AvroEncoding.EnsureCapacity(ref buffer, position, 18);
                var value = _getter(record);
                if (!value.HasValue)
                {
                    return AvroEncoding.WriteVarLong(buffer, position, _nonNullIndex == 0 ? 1 : 0);
                }
                position = AvroEncoding.WriteVarLong(buffer, position, _nonNullIndex);
                BinaryPrimitives.WriteDoubleLittleEndian(buffer.AsSpan(position), value.Value);
                return position + 8;
            }
        }

        private sealed class NullableBoolEncoder : FieldEncoder
        {
            private readonly Func<T, bool?> _getter;
            private readonly int _nonNullIndex;
            public NullableBoolEncoder(Func<T, bool?> getter, int nonNullIndex) { _getter = getter; _nonNullIndex = nonNullIndex; }

            public override int Encode(T record, ref byte[] buffer, int position)
            {
                AvroEncoding.EnsureCapacity(ref buffer, position, 11);
                var value = _getter(record);
                if (!value.HasValue)
                {
                    return AvroEncoding.WriteVarLong(buffer, position, _nonNullIndex == 0 ? 1 : 0);
                }
                position = AvroEncoding.WriteVarLong(buffer, position, _nonNullIndex);
                buffer[position] = value.Value ? (byte)1 : (byte)0;
                return position + 1;
            }
        }

        private sealed class NullableStringEncoder : FieldEncoder
        {
            private readonly Func<T, string?> _getter;
            private readonly int _nonNullIndex;
            public NullableStringEncoder(Func<T, string?> getter, int nonNullIndex) { _getter = getter; _nonNullIndex = nonNullIndex; }

            public override int Encode(T record, ref byte[] buffer, int position)
            {
                var value = _getter(record);
                if (value == null)
                {
                    AvroEncoding.EnsureCapacity(ref buffer, position, 10);
                    return AvroEncoding.WriteVarLong(buffer, position, _nonNullIndex == 0 ? 1 : 0);
                }
                var bytes = Encoding.UTF8.GetBytes(value);
                AvroEncoding.EnsureCapacity(ref buffer, position, bytes.Length + 20);
                position = AvroEncoding.WriteVarLong(buffer, position, _nonNullIndex);
                position = AvroEncoding.WriteVarLong(buffer, position, bytes.Length);
                Buffer.BlockCopy(bytes, 0, buffer, position, bytes.Length);
                return position + bytes.Length;
            }
        }

        private sealed class NullableBytesEncoder : FieldEncoder
        {
            private readonly Func<T, byte[]?> _getter;
            private readonly int _nonNullIndex;
            public NullableBytesEncoder(Func<T, byte[]?> getter, int nonNullIndex) { _getter = getter; _nonNullIndex = nonNullIndex; }

            public override int Encode(T record, ref byte[] buffer, int position)
            {
                var value = _getter(record);
                if (value == null)
                {
                    AvroEncoding.EnsureCapacity(ref buffer, position, 10);
                    return AvroEncoding.WriteVarLong(buffer, position, _nonNullIndex == 0 ? 1 : 0);
                }
                AvroEncoding.EnsureCapacity(ref buffer, position, value.Length + 20);
                position = AvroEncoding.WriteVarLong(buffer, position, _nonNullIndex);
                position = AvroEncoding.WriteVarLong(buffer, position, value.Length);
                Buffer.BlockCopy(value, 0, buffer, position, value.Length);
                return position + value.Length;
            }
        }

        #endregion
    }

    #endregion

    #region Direct Request Encoder

    internal static class DirectRequestEncoder
    {
        // Pre-encoded constants
        private static readonly byte[] BinaryEncodingBytes = Encoding.UTF8.GetBytes("binary");

        public static byte[] Encode(
            string tableName,
            IReadOnlyList<byte[]> records,
            IDictionary<string, string>? options)
        {
            // Estimate size: table name + records + overhead
            int estimatedSize = 256 + tableName.Length * 2;
            foreach (var record in records)
            {
                estimatedSize += record.Length + 10; // +10 for length prefix
            }

            var buffer = new byte[estimatedSize];
            int position = 0;

            // 1. Encode table_name (Avro string: length-prefixed UTF-8)
            position = WriteString(buffer, position, tableName);

            // 2. Encode list (Avro array of bytes)
            position = WriteBytesArray(buffer, position, records, ref buffer);

            // 3. Encode list_str (empty array)
            buffer[position++] = 0; // Empty array marker

            // 4. Encode list_encoding ("binary")
            position = WriteBytes(buffer, position, BinaryEncodingBytes);

            // 5. Encode options (Avro map)
            position = WriteStringMap(buffer, position, options, ref buffer);

            // Return exact-sized result
            var result = new byte[position];
            Buffer.BlockCopy(buffer, 0, result, 0, position);
            return result;
        }

        public static byte[] EncodeZeroCopy(
            string tableName,
            byte[] recordsBuffer,
            (int Offset, int Length)[] recordSegments,
            IDictionary<string, string>? options)
        {
            // Calculate exact size
            int size = 0;

            // Table name
            var tableNameBytes = Encoding.UTF8.GetBytes(tableName);
            size += AvroEncoding.GetVarIntSize(tableNameBytes.Length) + tableNameBytes.Length;

            // Records array: block header + sum of (length prefix + data)
            int recordsCount = recordSegments.Length;
            if (recordsCount > 0)
            {
                size += AvroEncoding.GetVarIntSize(recordsCount); // Block count
                foreach (var (offset, length) in recordSegments)
                {
                    size += AvroEncoding.GetVarIntSize(length) + length;
                }
                size += 1; // End of array marker
            }
            else
            {
                size += 1; // Empty array
            }

            // Empty list_str
            size += 1;

            // list_encoding "binary"
            size += AvroEncoding.GetVarIntSize(BinaryEncodingBytes.Length) + BinaryEncodingBytes.Length;

            // Options map
            int optionsCount = options?.Count ?? 0;
            if (optionsCount > 0)
            {
                size += AvroEncoding.GetVarIntSize(optionsCount);
                foreach (var kvp in options!)
                {
                    var keyBytes = Encoding.UTF8.GetBytes(kvp.Key);
                    var valueBytes = Encoding.UTF8.GetBytes(kvp.Value);
                    size += AvroEncoding.GetVarIntSize(keyBytes.Length) + keyBytes.Length;
                    size += AvroEncoding.GetVarIntSize(valueBytes.Length) + valueBytes.Length;
                }
                size += 1; // End of map marker
            }
            else
            {
                size += 1; // Empty map
            }

            // Allocate exact-sized buffer
            var result = new byte[size];
            int position = 0;

            // 1. Write table_name
            position = AvroEncoding.WriteVarInt(result, position, tableNameBytes.Length);
            Buffer.BlockCopy(tableNameBytes, 0, result, position, tableNameBytes.Length);
            position += tableNameBytes.Length;

            // 2. Write records array
            if (recordsCount > 0)
            {
                position = AvroEncoding.WriteVarInt(result, position, recordsCount);
                foreach (var (offset, length) in recordSegments)
                {
                    position = AvroEncoding.WriteVarInt(result, position, length);
                    Buffer.BlockCopy(recordsBuffer, offset, result, position, length);
                    position += length;
                }
                result[position++] = 0; // End of array
            }
            else
            {
                result[position++] = 0; // Empty array
            }

            // 3. Write empty list_str
            result[position++] = 0;

            // 4. Write list_encoding
            position = AvroEncoding.WriteVarInt(result, position, BinaryEncodingBytes.Length);
            Buffer.BlockCopy(BinaryEncodingBytes, 0, result, position, BinaryEncodingBytes.Length);
            position += BinaryEncodingBytes.Length;

            // 5. Write options map
            if (optionsCount > 0)
            {
                position = AvroEncoding.WriteVarInt(result, position, optionsCount);
                foreach (var kvp in options!)
                {
                    var keyBytes = Encoding.UTF8.GetBytes(kvp.Key);
                    var valueBytes = Encoding.UTF8.GetBytes(kvp.Value);

                    position = AvroEncoding.WriteVarInt(result, position, keyBytes.Length);
                    Buffer.BlockCopy(keyBytes, 0, result, position, keyBytes.Length);
                    position += keyBytes.Length;

                    position = AvroEncoding.WriteVarInt(result, position, valueBytes.Length);
                    Buffer.BlockCopy(valueBytes, 0, result, position, valueBytes.Length);
                    position += valueBytes.Length;
                }
                result[position++] = 0; // End of map
            }
            else
            {
                result[position++] = 0; // Empty map
            }

            return result;
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        private static int WriteString(byte[] buffer, int position, string value)
        {
            var bytes = Encoding.UTF8.GetBytes(value);
            position = AvroEncoding.WriteVarInt(buffer, position, bytes.Length);
            AvroEncoding.EnsureCapacity(ref buffer, position, bytes.Length);
            Buffer.BlockCopy(bytes, 0, buffer, position, bytes.Length);
            return position + bytes.Length;
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        private static int WriteBytes(byte[] buffer, int position, byte[] bytes)
        {
            position = AvroEncoding.WriteVarInt(buffer, position, bytes.Length);
            AvroEncoding.EnsureCapacity(ref buffer, position, bytes.Length);
            Buffer.BlockCopy(bytes, 0, buffer, position, bytes.Length);
            return position + bytes.Length;
        }

        private static int WriteBytesArray(byte[] buffer, int position, IReadOnlyList<byte[]> items, ref byte[] bufferRef)
        {
            int count = items.Count;
            if (count == 0)
            {
                buffer[position++] = 0; // Empty array
                return position;
            }

            // Write block count (positive = block with count items)
            position = AvroEncoding.WriteVarInt(buffer, position, count);

            // Write each item
            for (int i = 0; i < count; i++)
            {
                var item = items[i];
                AvroEncoding.EnsureCapacity(ref buffer, position, item.Length + 10);
                bufferRef = buffer;

                position = AvroEncoding.WriteVarInt(buffer, position, item.Length);
                Buffer.BlockCopy(item, 0, buffer, position, item.Length);
                position += item.Length;
            }

            // End of array marker
            buffer[position++] = 0;
            return position;
        }

        private static int WriteStringMap(byte[] buffer, int position, IDictionary<string, string>? map, ref byte[] bufferRef)
        {
            if (map == null || map.Count == 0)
            {
                buffer[position++] = 0; // Empty map
                return position;
            }

            // Write block count
            position = AvroEncoding.WriteVarInt(buffer, position, map.Count);

            // Write each key-value pair
            foreach (var kvp in map)
            {
                AvroEncoding.EnsureCapacity(ref buffer, position, 256);
                bufferRef = buffer;

                // Write key
                var keyBytes = Encoding.UTF8.GetBytes(kvp.Key);
                position = AvroEncoding.WriteVarInt(buffer, position, keyBytes.Length);
                Buffer.BlockCopy(keyBytes, 0, buffer, position, keyBytes.Length);
                position += keyBytes.Length;

                // Write value
                var valueBytes = Encoding.UTF8.GetBytes(kvp.Value);
                position = AvroEncoding.WriteVarInt(buffer, position, valueBytes.Length);
                Buffer.BlockCopy(valueBytes, 0, buffer, position, valueBytes.Length);
                position += valueBytes.Length;
            }

            // End of map marker
            buffer[position++] = 0;
            return position;
        }
    }

    #endregion