This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-11967-7dcc1602c460e25a624fd4f095a786352808b884 in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
commit 9532701801cf2aada8b1450170c691192ff92bc7 Author: Phoenix <[email protected]> AuthorDate: Wed Oct 28 15:08:45 2026 +0000 fix(ipc): decode zero-length bodies without EOS (#11967) # Which issue does this PR close? Closes #11966. # Rationale for this change A schema-only IPC stream without an EOS marker is valid, but `StreamDecoder` consumes its metadata and exits before processing the schema. As a result, `schema()` remains `None` and `finish()` reports `Unexpected End of Stream`. # What changes are included in this PR? Allow the decode loop to continue with empty input when the current message has a zero-length body. This completes the pending message while preserving the wait for more input when a nonempty body is incomplete. # Are these changes tested? Added `test_schema_without_eos`, which writes a schema without an EOS marker and verifies that the schema is decoded and `finish()` succeeds. It fails on the original implementation and passes with this change. All six stream-decoder unit tests pass locally. # Are there any significant user-facing changes? No public API changes. AI assistance was used for diagnosis, the regression test, and the proposed loop-condition change. --- arrow-ipc/src/reader/stream.rs | 26 +++++++++++++++++++++++++- 1 file changed, 25 insertions(+), 1 deletion(-) diff --git a/arrow-ipc/src/reader/stream.rs b/arrow-ipc/src/reader/stream.rs index 3ba712b117..7773429a49 100644 --- a/arrow-ipc/src/reader/stream.rs +++ b/arrow-ipc/src/reader/stream.rs @@ -157,7 +157,12 @@ impl StreamDecoder { /// } /// ``` pub fn decode(&mut self, buffer: &mut Buffer) -> Result<Option<RecordBatch>, ArrowError> { - while !buffer.is_empty() { + while !buffer.is_empty() + || matches!( + &self.state, + DecoderState::Body { message } if message.as_ref().bodyLength() == 0 + ) + { match &mut self.state { DecoderState::Header { buf, @@ -394,6 +399,25 @@ mod tests { assert_eq!(err, "Ipc error: Unexpected End of Stream"); } + #[test] + fn test_schema_without_eos() { + let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)])); + + let mut buf = Vec::new(); + let writer = StreamWriter::try_new(&mut buf, &schema).unwrap(); + // The schema has a zero-length body. Drop without finish to omit EOS. + drop(writer); + + let mut buffer = Buffer::from_vec(buf); + let mut decoder = StreamDecoder::new(); + let output = decoder.decode(&mut buffer).unwrap(); + + assert!(output.is_none()); + assert!(buffer.is_empty()); + assert_eq!(decoder.schema(), Some(schema)); + decoder.finish().unwrap(); + } + #[test] fn test_read_ree_dict_record_batches_from_buffer() { let schema = Schema::new(vec