GH-41110: [C#] Handle empty stream in ArrowStreamReaderImplementation (#47098)

### Rationale for this change
Fixing #41110.

### What changes are included in this PR?
Handle empty stream in `ArrowStreamReaderImplementation`. Similar changes have *not* been made to `ArrowMemoryReaderImplementation` or `ArrowFileReaderImplementation`.

### Are these changes tested?
Two basic unit tests have been created to validate the new behavior. This might not be sufficient to cover all cases where an empty stream should be handled without an exception occurring.

Original change by @ voidstar69; this takes his change and applies the PR feedback to it.

* GitHub Issue: #41110

Lead-authored-by: voidstar69 <[email protected]>
Co-authored-by: Curt Hagenlocher <[email protected]>
Signed-off-by: Curt Hagenlocher <[email protected]>

Author: CurtHagenlocher

csharp/src/Apache.Arrow/Ipc/ArrowStreamReaderImplementation.cs

@@ -68,7 +68,7 @@ protected async ValueTask<RecordBatch> ReadRecordBatchAsync(CancellationToken ca
 
         protected async ValueTask<ReadResult> ReadMessageAsync(CancellationToken cancellationToken)
         {
-            int messageLength = await ReadMessageLengthAsync(throwOnFullRead: false, cancellationToken)
+            int messageLength = await ReadMessageLengthAsync(throwOnFullRead: false, returnOnEmptyStream: false, cancellationToken)
                 .ConfigureAwait(false);
 
             if (messageLength == 0)
@@ -116,8 +116,7 @@ protected RecordBatch ReadRecordBatch()
 
         protected ReadResult ReadMessage()
         {
-            int messageLength = ReadMessageLength(throwOnFullRead: false);
-
+            int messageLength = ReadMessageLength(throwOnFullRead: false, returnOnEmptyStream: false);
             if (messageLength == 0)
             {
                 // reached end
@@ -160,8 +159,12 @@ public override async ValueTask ReadSchemaAsync(CancellationToken cancellationTo
             }
 
             // Figure out length of schema
-            int schemaMessageLength = await ReadMessageLengthAsync(throwOnFullRead: true, cancellationToken)
+            int schemaMessageLength = await ReadMessageLengthAsync(throwOnFullRead: true, returnOnEmptyStream: true, cancellationToken)
                 .ConfigureAwait(false);
+            if (schemaMessageLength == 0)
+            {
+                return;
+            }
 
             using (ArrayPool<byte>.Shared.RentReturn(schemaMessageLength, out Memory<byte> buff))
             {
@@ -182,7 +185,11 @@ public override void ReadSchema()
             }
 
             // Figure out length of schema
-            int schemaMessageLength = ReadMessageLength(throwOnFullRead: true);
+            int schemaMessageLength = ReadMessageLength(throwOnFullRead: true, returnOnEmptyStream: true);
+            if (schemaMessageLength == 0)
+            {
+                return;
+            }
 
             using (ArrayPool<byte>.Shared.RentReturn(schemaMessageLength, out Memory<byte> buff))
             {
@@ -194,13 +201,17 @@ public override void ReadSchema()
             }
         }
 
-        private async ValueTask<int> ReadMessageLengthAsync(bool throwOnFullRead, CancellationToken cancellationToken = default)
+        private async ValueTask<int> ReadMessageLengthAsync(bool throwOnFullRead, bool returnOnEmptyStream, CancellationToken cancellationToken = default)
         {
             int messageLength = 0;
             using (ArrayPool<byte>.Shared.RentReturn(4, out Memory<byte> lengthBuffer))
             {
                 int bytesRead = await BaseStream.ReadFullBufferAsync(lengthBuffer, cancellationToken)
                     .ConfigureAwait(false);
+                if (bytesRead == 0 && returnOnEmptyStream)
+                {
+                    return 0;
+                }
                 if (throwOnFullRead)
                 {
                     EnsureFullRead(lengthBuffer, bytesRead);
@@ -233,12 +244,16 @@ private async ValueTask<int> ReadMessageLengthAsync(bool throwOnFullRead, Cancel
             return messageLength;
         }
 
-        private int ReadMessageLength(bool throwOnFullRead)
+        private int ReadMessageLength(bool throwOnFullRead, bool returnOnEmptyStream)
         {
             int messageLength = 0;
             using (ArrayPool<byte>.Shared.RentReturn(4, out Memory<byte> lengthBuffer))
             {
                 int bytesRead = BaseStream.ReadFullBuffer(lengthBuffer);
+                if (bytesRead == 0 && returnOnEmptyStream)
+                {
+                    return 0;
+                }
                 if (throwOnFullRead)
                 {
                     EnsureFullRead(lengthBuffer, bytesRead);

csharp/test/Apache.Arrow.Tests/ArrowStreamReaderTests.cs

@@ -131,6 +131,17 @@ private static async Task TestReaderFromMemory(
             await verificationFunc(reader, originalBatch);
         }
 
+        [Fact]
+        public void ReadRecordBatch_EmptyStream()
+        {
+            using (MemoryStream stream = new())
+            {
+                ArrowStreamReader reader = new(stream);
+                RecordBatch readBatch = reader.ReadNextRecordBatch();
+                Assert.Null(readBatch);
+            }
+        }
+
         [Theory]
         [InlineData(true, true)]
         [InlineData(true, false)]
@@ -145,6 +156,17 @@ await TestReaderFromStream((reader, originalBatch) =>
             }, writeEnd, createDictionaryArray);
         }
 
+        [Fact]
+        public async Task ReadRecordBatchAsync_EmptyStream()
+        {
+            using (MemoryStream stream = new())
+            {
+                ArrowStreamReader reader = new(stream);
+                RecordBatch readBatch = await reader.ReadNextRecordBatchAsync();
+                Assert.Null(readBatch);
+            }
+        }
+
         [Theory]
         [InlineData(true, true)]
         [InlineData(true, false)]