You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "hanke580 (via GitHub)" <gi...@apache.org> on 2026/10/20 03:34:16 UTC
[I] arrow-avro: streaming `Decoder` returns wrong list/map values when a record body is split across two `decode()` calls [arrow-rs]
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: github-unsubscribe@arrow.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] arrow-avro: streaming `Decoder` returns wrong list/map values when a record body is split across two `decode()` calls [arrow-rs]
Posted by "ahlag (via GitHub)" <gi...@apache.org>.
ahlag commented on issue #11736:
URL: https://github.com/apache/arrow-rs/issues/11736#issuecomment-6283994583
I'd like to take this. I reproduced it on main: over every split position of the 81-byte stream in the report, 37 decode correctly, 35 give wrong rows and 8 fail with `bad varint`.
Before writing the PR, I'd like to check the approach, since it touches every field decoder:
1. **Roll back on incomplete data.** `RecordDecoder::decode` remembers `row_count` before each record. If a field returns EOF / need-more-data, it truncates every field decoder back to that many rows before returning the error. That needs a `Decoder::truncate(rows)`: `Vec::truncate` for the primitive arms, offsets plus child values for String/Binary/List/Map, per-branch counts for unions, and `pending_nulls` / validity for `Nullable`. The happy path costs nothing.
2. **Treat a truncated varint as EOF.** Make `AvroCursor::read_vlq` return `EOF` instead of `ParseError("bad varint")` when the buffer ends inside a varint, so `is_incomplete_data` treats it as "need more bytes". This has to land with (1). On its own it would turn those 8 errors into silently wrong rows.
The alternative is to run the existing `Skipper` over the body first and only decode once the whole record is present. That is much less code, but every framed record would be parsed twice, so I'd go with (1).
A few arms have no cheap truncate today: the `Decimal*` and `Duration` builders (`PrimitiveBuilder` has no `truncate`) and `RunEndEncoded`. I'd either switch those to `Vec`-backed storage like the other arms, or rebuild them on rollback.
The test would decode the report's list, map and `{id, tags}` shapes split at every position and compare each result with the one-shot decode.
Does this direction sound right? cc @Jefffrey
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org