Phoenix500526 opened a new issue, #11966:
URL: https://github.com/apache/arrow-rs/issues/11966
### Describe the bug
`StreamDecoder::decode()` leaves a complete message unprocessed when its
metadata consumes the remaining input and its `bodyLength` is zero. If there is
no subsequent input or explicit EOS marker, `finish()` reports `Unexpected End
of Stream` even though the stream ends at a complete message boundary.
A schema-only stream without EOS is a minimal reproduction: the input is
consumed, but `schema()` remains `None`. The same state-machine condition also
applies to record batches whose body length is zero.
Reproduced on apache/arrow-rs at `5db01b1fa7ba6f70d5f63abf31c48fabed40929b`,
before applying a local fix.
### To Reproduce
```rust
use arrow_buffer::Buffer;
use arrow_ipc::{reader::StreamDecoder, writer::StreamWriter};
use arrow_schema::{ArrowError, DataType, Field, Schema};
fn main() -> Result<(), ArrowError> {
let schema = Schema::new(vec![Field::new("id", DataType::Int32, false)]);
let mut bytes = Vec::new();
// try_new writes the schema, whose body length is zero.
// Drop the unbuffered writer without finish() to omit the EOS marker.
drop(StreamWriter::try_new(&mut bytes, &schema)?);
let mut buffer = Buffer::from(bytes);
let mut decoder = StreamDecoder::new();
assert!(decoder.decode(&mut buffer)?.is_none());
assert!(buffer.is_empty());
println!("schema decoded: {}", decoder.schema().is_some());
println!("finish: {:?}", decoder.finish());
Ok(())
}
```
Actual output:
```text
schema decoded: false
finish: Err(IpcError("Unexpected End of Stream"))
```
Calling `decode()` again with an empty buffer does not process the pending
message either.
### Expected behavior
The schema should be available immediately after `decode()`, and `finish()`
should succeed. Returning `None` from `decode()` itself is correct here because
there is no record batch.
An explicit EOS marker is optional:
- The [Arrow IPC streaming
specification](https://arrow.apache.org/docs/format/Columnar.html#ipc-streaming-format)
marks EOS as optional and allows a writer to end the stream by closing the
stream interface instead of writing the marker.
- The [Schema message
specification](https://arrow.apache.org/docs/format/Columnar.html#schema-message)
specifies that a serialized schema has metadata only, with no body.
- The existing [`MessageReader::read_meta_len()` comment and
implementation](https://github.com/apache/arrow-rs/blob/5db01b1fa7ba6f70d5f63abf31c48fabed40929b/arrow-ipc/src/reader.rs#L1998-L2007)
explicitly handle EOF without `0xFFFFFFFF 0x00000000` and cite the
specification.
-
[`StreamDecoder::finish()`](https://github.com/apache/arrow-rs/blob/5db01b1fa7ba6f70d5f63abf31c48fabed40929b/arrow-ipc/src/reader/stream.rs#L298-L310)
already accepts the initial `Header` state as a valid end, without requiring
`Finished`.
### Additional context
The [decode
loop](https://github.com/apache/arrow-rs/blob/5db01b1fa7ba6f70d5f63abf31c48fabed40929b/arrow-ipc/src/reader/stream.rs#L159-L207)
runs only while `!buffer.is_empty()`. After parsing metadata, the `Message`
arm changes the state to `Body`. If that consumes the last input byte, the loop
exits before the `Body` arm can process the message, even when no body bytes
are needed.
Appending EOS masks this because the extra bytes allow another iteration. At
an intermediate chunk boundary, the same condition delays processing of a
complete zero-body message until another nonempty chunk arrives.
The loop should also advance when a zero-body message is ready to process.
It must still stop for a nonzero body when more input is needed: an empty input
buffer does not imply a zero declared body length.
Related issues and PRs:
- #9837 and #11374: the proposed `BufferStreamReader` wraps `StreamDecoder`;
#11374 explicitly intends to accept streams ending at a message boundary
without EOS. This edge case is relevant to its schema-only input handling.
- #6420 and #7488: exposed `StreamDecoder::schema()`, including the use case
of obtaining a schema when there are zero record batches.
- #5531: introduced `StreamDecoder`; the original implementation already
contains this loop condition and `Message` to `Body` transition, so this
behavior predates the recent buffering changes.
A local fix allowing the zero-body transition makes the schema-only
regression test pass; all six stream-decoder tests, including the new
regression test, pass locally.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
