read schema in fully async way

.github/workflows/full.yml

@@ -1,7 +1,7 @@
 name: 'Full Workflow'
 
 env:
-  VERSION: 4.3.2
+  VERSION: 4.3.3
   ASM_VERSION: 4.0.0
 
 on:

src/Parquet.Test/ParquetReaderTest.cs

@@ -9,6 +9,7 @@
 using System.Linq;
 using System.Threading.Tasks;
 using Path = System.IO.Path;
+using System.Threading;
 
 namespace Parquet.Test {
     public class ParquetReaderTest : TestBase {
@@ -55,6 +56,15 @@ public async Task Read_simple_map(string parquetFile) {
             }
         }
 
+        [Fact(Skip = "todo: for some reason .Read (sync) is still called")]
+        public async Task Reading_schema_uses_async_only_methods() {
+            using Stream tf = OpenTestFile("map_simple.parquet");
+            using Stream ao = new AsyncOnlyStream(tf, 3);
+            using ParquetReader reader = await ParquetReader.CreateAsync(ao);
+
+            Assert.NotNull(reader.Schema);
+        }
+
         [Theory]
         [InlineData("complex-primitives.parquet")]
         [InlineData("complex-primitives.v2.parquet")]
@@ -280,5 +290,42 @@ public ReadableAndSeekableStream(Stream master) : base(master) {
             public override bool CanRead => true;
 
         }
+
+        class AsyncOnlyStream : Stream {
+            private readonly Stream _baseStream;
+            private readonly int _maxSyncReads;
+            private int _syncReads = 0;
+
+            public AsyncOnlyStream(Stream baseStream, int maxSyncReads) {
+                _baseStream = baseStream;
+                _maxSyncReads = maxSyncReads;
+            }
+
+            public override bool CanRead => _baseStream.CanRead;
+
+            public override bool CanSeek => _baseStream.CanSeek;
+
+            public override bool CanWrite => _baseStream.CanWrite;
+
+            public override long Length => _baseStream.Length;
+
+            public override long Position { get => _baseStream.Position; set => _baseStream.Position = value; }
+
+            public override void Flush() => _baseStream.Flush();
+
+            public override int Read(byte[] buffer, int offset, int count) {
+                _syncReads++;
+                if(_syncReads > _maxSyncReads) {
+                    throw new IOException($"limit of {_maxSyncReads} reached");
+                }
+                return _baseStream.Read(buffer, offset, count);
+            }
+
+            public override long Seek(long offset, SeekOrigin origin) => _baseStream.Seek(offset, origin);
+
+            public override void SetLength(long value) => _baseStream.SetLength(value);
+
+            public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();
+        }
     }
 }

src/Parquet/Extensions/ArrayExtensions.cs

@@ -19,5 +19,9 @@ public static void Explode(this Array dictionary,
                 }
             }
         }
+
+        public static string ToHexString(this byte[] b) {
+            return BitConverter.ToString(b).Replace("-", "");
+        }
     }
 }

src/Parquet/Extensions/StreamExtensions.cs

@@ -1,5 +1,6 @@
 using System;
 using System.IO;
+using System.Threading.Tasks;
 
 namespace Parquet.Extensions {
     static class StreamExtensions {
@@ -9,6 +10,12 @@ public static int ReadInt32(this Stream s) {
             return BitConverter.ToInt32(tmp, 0);
         }
 
+        public static async Task<int> ReadInt32Async(this Stream s) {
+            byte[] tmp = new byte[sizeof(int)];
+            await s.ReadAsync(tmp, 0, sizeof(int));
+            return BitConverter.ToInt32(tmp, 0);
+        }
+
         public static void WriteInt32(this Stream s, int value) {
             byte[] tmp = BitConverter.GetBytes(value);
             s.Write(tmp, 0, sizeof(int));
@@ -39,5 +46,25 @@ public static byte[] ReadBytesExactly(this Stream s, int count) {
 
             return tmp;
         }
+
+        public static async Task<byte[]> ReadBytesExactlyAsync(this Stream s, int count) {
+            byte[] tmp = new byte[count];
+#if NET7_0_OR_GREATER
+            await s.ReadExactlyAsync(tmp, 0, count);
+#else
+            int read = 0;
+            while(read < count) {
+                int r = await s.ReadAsync(tmp, read, count - read);
+                if(r == 0)
+                    break;
+                else
+                    read += r;
+            }
+            if(read < count)
+                throw new IOException($"only {read} out of {count} bytes are available");
+#endif
+
+            return tmp;
+        }
     }
 }

src/Parquet/ParquetActor.cs

@@ -1,8 +1,10 @@
 using System;
 using System.IO;
+using System.Linq;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
+using Parquet.Extensions;
 using Parquet.File;
 
 namespace Parquet {
@@ -11,12 +13,10 @@ namespace Parquet {
     /// </summary>
     public class ParquetActor {
 #pragma warning disable IDE1006
-        private const string MagicString = "PAR1";
-        internal static readonly byte[] MagicBytes = Encoding.ASCII.GetBytes(MagicString);
+        internal static readonly byte[] MagicBytes = Encoding.ASCII.GetBytes("PAR1");
 #pragma warning restore IDE1006
 
         private readonly Stream _fileStream;
-        private BinaryReader _binaryReader;
         private BinaryWriter _binaryWriter;
         private ThriftStream _thriftStream;
 
@@ -28,39 +28,39 @@ internal ParquetActor(Stream fileStream) =>
         /// </summary>
         protected Stream Stream => _fileStream;
 
-        internal BinaryReader Reader => _binaryReader ??= new BinaryReader(_fileStream);
-
         internal BinaryWriter Writer => _binaryWriter ??= new BinaryWriter(_fileStream);
 
         internal ThriftStream ThriftStream => _thriftStream ??= new ThriftStream(_fileStream);
 
-        internal void ValidateFile() {
+        /// <summary>
+        /// Validates that this file is a valid parquet file by reading head and tail of it
+        /// </summary>
+        /// <returns></returns>
+        /// <exception cref="IOException"></exception>
+        protected async Task ValidateFileAsync() {
             _fileStream.Seek(0, SeekOrigin.Begin);
-            char[] head = Reader.ReadChars(4);
-            string shead = new string(head);
+            byte[] head = await _fileStream.ReadBytesExactlyAsync(4);
+
             _fileStream.Seek(-4, SeekOrigin.End);
-            char[] tail = Reader.ReadChars(4);
-            string stail = new string(tail);
-            if(shead != MagicString)
-                throw new IOException($"not a Parquet file(head is '{shead}')");
-            if(stail != MagicString)
-                throw new IOException($"not a Parquet file(tail is '{stail}')");
+            byte[] tail = await _fileStream.ReadBytesExactlyAsync(4);
+
+            if(!MagicBytes.SequenceEqual(head) || !MagicBytes.SequenceEqual(tail))
+                throw new IOException($"not a parquet file, head: {head.ToHexString()}, tail: {tail.ToHexString()}");
         }
 
         internal async Task<Thrift.FileMetaData> ReadMetadataAsync(CancellationToken cancellationToken = default) {
-            GoBeforeFooter();
-
+            await GoBeforeFooterAsync();
             return await ThriftStream.ReadAsync<Thrift.FileMetaData>(cancellationToken);
         }
 
         internal void GoToBeginning() => _fileStream.Seek(0, SeekOrigin.Begin);
 
         internal void GoToEnd() => _fileStream.Seek(0, SeekOrigin.End);
 
-        internal void GoBeforeFooter() {
+        internal async Task GoBeforeFooterAsync() {
             //go to -4 bytes (PAR1) -4 bytes (footer length number)
             _fileStream.Seek(-8, SeekOrigin.End);
-            int footerLength = Reader.ReadInt32();
+            int footerLength = await _fileStream.ReadInt32Async();
 
             //set just before footer starts
             _fileStream.Seek(-8 - footerLength, SeekOrigin.End);

src/Parquet/ParquetReader.cs

@@ -35,7 +35,7 @@ private ParquetReader(Stream input, ParquetOptions parquetOptions = null, bool l
         }
 
         private async Task InitialiseAsync(CancellationToken cancellationToken) {
-            ValidateFile();
+            await ValidateFileAsync();
 
             //read metadata instantly, now
             _meta = await ReadMetadataAsync(cancellationToken);

src/Parquet/ParquetWriter.cs

@@ -83,14 +83,14 @@ private async Task PrepareFileAsync(bool append, CancellationToken cancellationT
                 if(!Stream.CanSeek)
                     throw new IOException("destination stream must be seekable for append operations.");
 
-                ValidateFile();
+                await ValidateFileAsync();
 
                 Thrift.FileMetaData fileMeta = await ReadMetadataAsync(cancellationToken);
                 _footer = new ThriftFooter(fileMeta);
 
                 ValidateSchemasCompatible(_footer, _schema);
 
-                GoBeforeFooter();
+                await GoBeforeFooterAsync();
             }
             else {
                 if(_footer == null) {