hanke580 opened a new issue, #11736:
URL: https://github.com/apache/arrow-rs/issues/11736
### Describe the bug
`Decoder::decode` is documented to handle a framed record whose body
straddles the chunk boundary: it returns fewer bytes consumed and the caller
passes the rest again with the next chunk. When the split falls inside a value
of a list (or map) field, the values decoded from the first, incomplete attempt
stay in the child builders, and the body is then decoded again from its start.
For a record whose first field is a list, the result is silently wrong: list
items are duplicated, shifted into the next row, and the last item is lost. No
error is returned.
### To Reproduce
`Cargo.toml` dependencies: `arrow-array`, `arrow-schema`, `arrow-avro`, and
`arrow` with the `prettyprint` feature (60.0.0, or git main).
```rust
use std::sync::Arc;
use arrow::util::pretty::pretty_format_batches;
use arrow_array::builder::{ListBuilder, StringBuilder};
use arrow_array::{ArrayRef, RecordBatch};
use arrow_avro::reader::{Decoder, ReaderBuilder};
use arrow_avro::schema::{AvroSchema, SchemaStore};
use arrow_avro::writer::{format::AvroSoeFormat, WriterBuilder};
use arrow_schema::{DataType, Field, Schema};
fn main() -> Result<(), Box<dyn std::error::Error>> {
// tags: list<string>, two rows
let item = Arc::new(Field::new("item", DataType::Utf8, false));
let schema = Schema::new(vec![Field::new("tags",
DataType::List(item.clone()), false)]);
let mut b = ListBuilder::new(StringBuilder::new()).with_field(item);
for row in [["r0-alpha", "r0-bravo", "r0-charlie"], ["r1-delta",
"r1-echo", "r1-foxtrot"]] {
for s in row {
b.values().append_value(s);
}
b.append(true);
}
let batch = RecordBatch::try_new(Arc::new(schema.clone()),
vec![Arc::new(b.finish()) as ArrayRef])?;
// Encode as a single-object-encoding stream (Rabin fingerprint).
let mut enc =
WriterBuilder::new(schema.clone()).build_encoder::<AvroSoeFormat>()?;
enc.encode(&batch)?;
let mut bytes = Vec::new();
for row in enc.flush().iter() {
bytes.extend_from_slice(&row);
}
let new_decoder = || -> Result<Decoder, Box<dyn std::error::Error>> {
let mut store = SchemaStore::new();
store.register(AvroSchema::try_from(&schema)?)?;
Ok(ReaderBuilder::new().with_writer_schema_store(store).build_decoder()?)
};
// Deliver the same bytes in two chunks, split in the middle of
"r0-bravo".
// Unconsumed bytes are kept and passed again together with the next
chunk.
let split = bytes.windows(8).position(|w| w == b"r0-bravo").unwrap() + 4;
let mut decoder = new_decoder()?;
let mut buf = bytes[..split].to_vec();
let n = decoder.decode(&buf)?;
println!("decode({} bytes) -> consumed {n}", buf.len());
buf.drain(..n);
buf.extend_from_slice(&bytes[split..]);
while !buf.is_empty() {
let n = decoder.decode(&buf)?;
println!("decode({} bytes) -> consumed {n}", buf.len());
if n == 0 {
break;
}
buf.drain(..n);
}
let out = decoder.flush()?.unwrap();
println!("expected:\n{}", pretty_format_batches(&[batch])?);
println!("actual:\n{}", pretty_format_batches(&[out])?);
Ok(())
}
```
Output on main (33d338dd) and on 60.0.0:
```
decode(25 bytes) -> consumed 10
decode(71 bytes) -> consumed 71
expected:
+----------------------------------+
| tags |
+----------------------------------+
| [r0-alpha, r0-bravo, r0-charlie] |
| [r1-delta, r1-echo, r1-foxtrot] |
+----------------------------------+
actual:
+---------------------------------+
| tags |
+---------------------------------+
| [r0-alpha, r0-alpha, r0-bravo] |
| [r0-charlie, r1-delta, r1-echo] |
+---------------------------------+
```
Decoding the whole buffer in one call gives the expected table.
### Expected behavior
The decoded rows do not depend on where the input is split into chunks:
either the partial attempt leaves no trace, or the decoder resumes where it
stopped.
### Additional context
Tested: main 33d338ddf661b6e85364a9b0bd4d367c3468257d (2026-10-18), 60.0.0
and 59.3.1.
Trying each of the 80 split positions of the stream above (81 bytes) on
main: 37 give the right result, 35 give silently wrong rows, and 8 fail with
`bad varint` (see below). Other shapes:
- `map<string,string>` as the only field: wrong rows at some positions (e.g.
`{k1: v1-alpha, k1: v1-alpha}`), at others `Map keys length (5) != map values
length (4)` (or similar) at flush.
- When the list is not the first field (`{id: int, tags: list<string>}`), or
with two string fields split inside the second, the values appended by the
failed attempt make the columns uneven, and `flush()` fails with `all columns
in a record batch must have the specified row count`. That is an error rather
than wrong data, but the input is valid.
- When the split falls just before or inside a varint (a length, block count
or int), `decode()` fails with `ParseError("bad varint")`.
`AvroCursor::read_vlq`
([`reader/cursor.rs:61-66`](https://github.com/apache/arrow-rs/blob/33d338ddf661b6e85364a9b0bd4d367c3468257d/arrow-avro/src/reader/cursor.rs#L61-L66))
reports a truncated varint as a `ParseError`, not `EOF`, so
`is_incomplete_data`
([`reader/mod.rs:520`](https://github.com/apache/arrow-rs/blob/33d338ddf661b6e85364a9b0bd4d367c3468257d/arrow-avro/src/reader/mod.rs#L520-L525))
does not treat it as "need more bytes".
Likely cause: `Decoder::decode_framed`
([`arrow-avro/src/reader/mod.rs:736-766`](https://github.com/apache/arrow-rs/blob/33d338ddf661b6e85364a9b0bd4d367c3468257d/arrow-avro/src/reader/mod.rs#L736-L766))
handles an incomplete-data error from `active_decoder.decode(.., 1)` by
keeping `awaiting_body = true` and retrying the whole body on the next call.
`RecordDecoder::decode`
([`reader/record.rs:155-173`](https://github.com/apache/arrow-rs/blob/33d338ddf661b6e85364a9b0bd4d367c3468257d/arrow-avro/src/reader/record.rs#L155-L173))
writes into the field decoders as it goes, and nothing truncates the field
decoders back to the last complete row when it returns an error. For lists, the
leftover child values remain in the child decoder while
`off.push_length(total_items)`
([`reader/record.rs:1360-1362`](https://github.com/apache/arrow-rs/blob/33d338ddf661b6e85364a9b0bd4d367c3468257d/arrow-avro/src/reader/record.rs#L1360-L1362))
records only the retried row's item count, so every later offse
t points one value too early. Single-object and Confluent framing both go
through `decode_framed`. The existing streaming tests
(`test_row_writer_roundtrip_decoder_soe_streaming_chunks` in `writer/mod.rs`)
feed only frame-aligned chunks.
Related but different: #8212 (panic at flush after a failed decode of a
nullable field, closed), #10494 / #10497 (OCF sync marker split across
`decode()` calls, fixed), #8100 (the refactor that added partial decoding).
--
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]
