How to deserialize streaming and detect the end of stream correctly?

[marcwittke]

Given a binary deserializer of Record and a huge gzip stream with serialized records, how do I detect the end of stream correctly?

This test demonstrates the problem:

using System;
using System.Collections.Generic;
using System.IO;
using System.IO.Compression;
using System.Threading;
using Chr.Avro.Abstract;
using Chr.Avro.Resolution;
using Chr.Avro.Serialization;
using Xunit;

namespace AvroTests
{
    public class TheSerialization
    {
        private const int Count = 100_000;
        private readonly IBinarySerializer<Record> _serializer;
        private readonly IBinaryDeserializer<Record> _deserializer;

        public TheSerialization()
        {
            var typeResolver = new ReflectionResolver(resolveReferenceTypesAsNullable: true);
            var builder = new SchemaBuilder(typeResolver: typeResolver);
            Schema serializerSchema = builder.BuildSchema<Record>();
            _serializer = new BinarySerializerBuilder().BuildSerializer<Record>(serializerSchema);

            Schema deserializerSchema = builder.BuildSchema<Record>();
            _deserializer = new BinaryDeserializerBuilder().BuildDeserializer<Record>(deserializerSchema);
        }

        [Fact]
        public void CanDeserializeStreaming()
        {
            using (var fileStream = File.Create("records.avro.gz"))
            {
                using (var gzipStream = new GZipStream(fileStream, CompressionLevel.Optimal))
                {
                    for (int i = 0; i < Count; i++)
                    {
                        _serializer.Serialize(new Record(), gzipStream);
                    }
                }
            }

            using (var fileStream = File.OpenRead("records.avro.gz"))
            {
                ISet<Record> deserialized = new HashSet<Record>();
                using (var gzipStream = new GZipStream(fileStream, CompressionMode.Decompress))
                {
                    while (true)   //<---- what to check here? ####
                    {
                        deserialized.Add(_deserializer.Deserialize(gzipStream));
                        Assert.True(deserialized.Count <= Count);
                    }
                }

                Assert.Equal(Count, deserialized.Count);
            }
        }
    }


    public class Record
    {
        private static int _recordCounter;

        public Record()
        {
            int index = Interlocked.Increment(ref _recordCounter);
            Name = $"Name {index}";
            DecimalValue = 234765.234m / index;
            IntNumber = index;
            TimeStamp = DateTime.MinValue.AddDays(index);
        }
        
        public string Name { get; set; }

        public int IntNumber { get; set; }

        public DateTime TimeStamp { get; set; }

        public decimal DecimalValue { get; set; }
    }
}

Currently, the test deserializes the 100.000 records, but ends with an exception when it reaches the end of the underlying stream:

System.OverflowException
Encoded integer exceeds long bounds.
   at lambda_method27(Closure , Stream )
   at AvroTests.TheSerialization.CanDeserializeStreaming()

I also tried to stop when the underlying gzip stream position is at the end, but that will result in the loss of data, since the buffer still contains unread records when this happens.

[dstelljes]

Yeah, that’s a good question. At the decoding level, the library could handle that case better—seems like EndOfStreamException should be thrown instead of OverflowException. That would still require you to catch an exception as part of normal operation, which isn’t ideal.

The better solution is probably something like an Object Container File (#156). You could also make a simpler envelope, such as by writing out the record count at the beginning of the file.

[marcwittke]

It would be sufficient, when the deserializer could return an IEnumerator<T> where I could defensively check if (e.MoveNext() && e.Current !_ null), maybe this would be a cheap fix?

[dstelljes]

Internally, the library would still probably have to do a try-catch to make that work. The API has also changed in 8.x; instead of reading from a Stream, our BinaryReader operates on a ReadOnlySpan<byte> for parity with .NET’s Utf8JsonReader.