linliu-code opened a new issue, #10876:
URL: https://github.com/apache/arrow-rs/issues/10876
## Describe the bug
`arrow-avro`'s `Decoder` produces a corrupt second batch when the schema
contains an Avro union with two or more non-null branches. The first batch
decodes correctly; the next one fails:
```
Parser error: Invalid argument error: Offsets must be non-negative and
within the length of the Array
```
Unions of the form `["null", T]` are unaffected, since those become a
nullable `T` rather than a `UnionArray`.
## To Reproduce
Schema with a two-branch union, every value null, more records than one
batch holds:
```rust
const SCHEMA: &str = r#"{"type":"record","name":"R","fields":[
{"name":"id","type":"long"},
{"name":"v","type":["null","int","string"]}]}"#;
let mut store = SchemaStore::new();
let fingerprint = store.register(AvroSchema::new(SCHEMA.to_string()))?;
let mut decoder = ReaderBuilder::new()
.with_writer_schema_store(store)
.with_active_fingerprint(fingerprint)
.with_reader_schema(AvroSchema::new(SCHEMA.to_string()))
.with_batch_size(16)
.build_decoder()?;
// Feed 100 single-object-encoded records whose `v` is null, flushing
whenever
// batch_is_full(). The first flush returns 16 rows; the next fails.
```
The data is valid: `apache_avro::from_avro_datum` reads all 100 records
against the same schema.
Varying only the batch size, with 5000 records:
| batch_size | result |
| --- | --- |
| 1024 | fails after 1 batch |
| 4096 | fails after 1 batch |
| 8192 | ok, 5000 rows in one batch |
| 16384 | ok, 5000 rows in one batch |
So the boundary is the batch, not the record count or the data.
## Expected behavior
A decoder should produce as many batches as the input requires.
## Additional context
Cause appears to be in `UnionDecoder` (`arrow-avro/src/reader/record.rs`).
Each offset is derived from a per-branch counter:
```rust
self.offsets.push(self.counts[reader_idx]); // :1513
self.counts[reader_idx] += 1; // :1514
```
but `flush` drains `type_ids` and `offsets` and flushes the children while
leaving `counts` at the previous batch's totals:
```rust
let children = self.branches.iter_mut().map(|d| d.flush(None))...;
let arr = UnionArray::try_new(
self.fields.clone(),
flush_values(&mut self.type_ids).into_iter().collect(),
Some(flush_values(&mut self.offsets).into_iter().collect()),
children,
)
```
So the second batch emits offsets starting where the first left off, against
children that were just emptied, and `UnionArray::try_new` rejects them.
Zeroing `counts` in `flush` looks like it would be sufficient.
This also explains why all-null data still triggers it: the null branch's
counter grows like any other.
Found in 57.3.1. Reproduced against a schema with a nested union too (union
inside a record field), and with `["null","int","long"]`, so it is not specific
to string offsets.
--
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]
