From 902ff93fc8c0f235c7aeabde873825ba9f236854 Mon Sep 17 00:00:00 2001 From: Rusty Conover Date: Tue, 25 Aug 2026 12:16:45 -0400 Subject: [PATCH 1/2] fix: Return immediately from a zero-length ReadFullBufferAsync/ReadFullBuffer instead of calling into the stream A RecordBatch message body is legitimately zero bytes whenever the batch has no buffers (e.g. a batch built from a zero-column schema). StreamExtensions.ReadFullBufferAsync/ReadFullBuffer called stream.ReadAsync/stream.Read with that zero-length buffer unconditionally. Over a MemoryStream this is a harmless no-op. Over a real socket-backed NetworkStream, a zero-byte ReadAsync/Read does not complete immediately as it logically should -- it blocks as though waiting for the peer to send more data (or close the connection), instead of trivially returning 0. In a lockstep/RPC-style protocol where the peer is itself waiting for a response before sending anything further, this blocks indefinitely. Both read helpers now short-circuit buffer.Length == 0 and return 0 immediately, before ever touching the stream -- matching the Go standard library's documented behavior for io.ReadFull, which special-cases a zero-length buffer and never issues the read at all. Closes #425. --- .../Extensions/StreamExtensions.cs | 20 ++++ .../StreamExtensionsTests.cs | 98 +++++++++++++++++++ 2 files changed, 118 insertions(+) create mode 100644 test/Apache.Arrow.Tests/StreamExtensionsTests.cs diff --git a/src/Apache.Arrow/Extensions/StreamExtensions.cs b/src/Apache.Arrow/Extensions/StreamExtensions.cs index ac1e9751..1ee7df10 100644 --- a/src/Apache.Arrow/Extensions/StreamExtensions.cs +++ b/src/Apache.Arrow/Extensions/StreamExtensions.cs @@ -24,6 +24,19 @@ internal static partial class StreamExtensions { public static async ValueTask ReadFullBufferAsync(this Stream stream, Memory buffer, CancellationToken cancellationToken = default) { + // A zero-length request is trivially satisfied — 0 bytes were asked for, 0 were + // read — and must return WITHOUT ever calling stream.ReadAsync. Socket-backed streams + // (NetworkStream et al.) do not treat a zero-length ReadAsync as an immediate no-op + // the way MemoryStream does: it behaves as a "wait for the socket to become readable" + // probe, blocking until the peer sends *something* (or closes). A RecordBatch message + // body is legitimately zero-length whenever the batch has no buffers (e.g. a + // zero-column schema), so without this fast path, reading such a batch's empty body + // over a real socket blocks indefinitely instead of completing immediately. + if (buffer.Length == 0) + { + return 0; + } + int totalBytesRead = 0; do { @@ -48,6 +61,13 @@ await stream.ReadAsync( public static int ReadFullBuffer(this Stream stream, Memory buffer) { + // See the matching guard in ReadFullBufferAsync above — same + // zero-length-buffer-blocks-on-socket-streams rationale applies to the sync path. + if (buffer.Length == 0) + { + return 0; + } + int totalBytesRead = 0; do { diff --git a/test/Apache.Arrow.Tests/StreamExtensionsTests.cs b/test/Apache.Arrow.Tests/StreamExtensionsTests.cs new file mode 100644 index 00000000..d746a545 --- /dev/null +++ b/test/Apache.Arrow.Tests/StreamExtensionsTests.cs @@ -0,0 +1,98 @@ +// Licensed to the Apache Software Foundation (ASF) under one or more +// contributor license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright ownership. +// The ASF licenses this file to You 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. + +using System; +using System.IO; +using System.Threading; +using System.Threading.Tasks; +using Xunit; + +namespace Apache.Arrow.Tests +{ + public class StreamExtensionsTests + { + /// + /// A stream whose Read/ReadAsync overrides throw if ever invoked, standing in for a + /// socket-backed stream (e.g. NetworkStream) whose zero-length ReadAsync/Read does not + /// complete immediately the way MemoryStream's does — it blocks as though waiting for the + /// peer to send more data. Used to prove ReadFullBufferAsync/ReadFullBuffer never call + /// into the underlying stream for a zero-length request. + /// + private sealed class ThrowsIfReadStream : Stream + { + public override bool CanRead => true; + public override bool CanSeek => false; + public override bool CanWrite => false; + public override long Length => throw new NotSupportedException(); + public override long Position + { + get => throw new NotSupportedException(); + set => throw new NotSupportedException(); + } + + public override int Read(byte[] buffer, int offset, int count) => + throw new InvalidOperationException("Read should not be called for a zero-length buffer."); + + public override Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => + throw new InvalidOperationException("ReadAsync should not be called for a zero-length buffer."); + + public override ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) => + throw new InvalidOperationException("ReadAsync should not be called for a zero-length buffer."); + + public override void Flush() => throw new NotSupportedException(); + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + public override void SetLength(long value) => throw new NotSupportedException(); + public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + } + + [Fact] + public async Task ReadFullBufferAsync_ZeroLengthBuffer_ReturnsWithoutTouchingStream() + { + var stream = new ThrowsIfReadStream(); + int bytesRead = await stream.ReadFullBufferAsync(Memory.Empty); + Assert.Equal(0, bytesRead); + } + + [Fact] + public void ReadFullBuffer_ZeroLengthBuffer_ReturnsWithoutTouchingStream() + { + var stream = new ThrowsIfReadStream(); + int bytesRead = stream.ReadFullBuffer(Memory.Empty); + Assert.Equal(0, bytesRead); + } + + [Fact] + public async Task ReadFullBufferAsync_NonEmptyBuffer_ReadsFromStream() + { + var data = new byte[] { 1, 2, 3, 4 }; + using var stream = new MemoryStream(data); + var buffer = new byte[4]; + int bytesRead = await stream.ReadFullBufferAsync(buffer); + Assert.Equal(4, bytesRead); + Assert.Equal(data, buffer); + } + + [Fact] + public void ReadFullBuffer_NonEmptyBuffer_ReadsFromStream() + { + var data = new byte[] { 1, 2, 3, 4 }; + using var stream = new MemoryStream(data); + var buffer = new byte[4]; + int bytesRead = stream.ReadFullBuffer(buffer); + Assert.Equal(4, bytesRead); + Assert.Equal(data, buffer); + } + } +} From 36a8f2492b9b28f4dc26f8051f5cbf6e8209fb78 Mon Sep 17 00:00:00 2001 From: Rusty Conover Date: Tue, 25 Aug 2026 14:02:40 -0400 Subject: [PATCH 2/2] Fix net462/net472 build break: guard Memory ReadAsync override with #if NETCOREAPP Stream.ReadAsync(Memory, CancellationToken) is only a virtual member of Stream on netcoreapp targets; net462/net472 don't declare it, so overriding it unconditionally in ThrowsIfReadStream broke the .NET Framework builds (CS0115). Addresses review comment from @CurtHagenlocher on #426. --- test/Apache.Arrow.Tests/StreamExtensionsTests.cs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/test/Apache.Arrow.Tests/StreamExtensionsTests.cs b/test/Apache.Arrow.Tests/StreamExtensionsTests.cs index d746a545..20e101df 100644 --- a/test/Apache.Arrow.Tests/StreamExtensionsTests.cs +++ b/test/Apache.Arrow.Tests/StreamExtensionsTests.cs @@ -48,8 +48,12 @@ public override int Read(byte[] buffer, int offset, int count) => public override Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => throw new InvalidOperationException("ReadAsync should not be called for a zero-length buffer."); +#if NETCOREAPP + // Stream.ReadAsync(Memory, CancellationToken) is only overridable on + // netcoreapp targets — net462/net472 don't declare it as virtual on Stream. public override ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) => throw new InvalidOperationException("ReadAsync should not be called for a zero-length buffer."); +#endif public override void Flush() => throw new NotSupportedException(); public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();