JoeryH opened a new issue, #11478:
URL: https://github.com/apache/arrow-rs/issues/11478
**Describe the bug**
When a `RowSelection` is scattered enough for `RowSelectionPolicy::Auto`
(the default, `threshold: 32`) to pick the **Mask** strategy, the reader
decodes every row from the first selected row up to the `batch_size`-th
selected row in one go, and only filters afterwards. The number of decoded rows
per batch therefore has no upper bound: with 1 row kept in every `k`, one batch
decodes about `batch_size * k` rows.
For a `Utf8`/`Binary` column, all of those decoded values go into a single
`i32`-offset buffer before the filter is applied. The read then fails with
`index overflow decoding byte array`, even though the *selected* data per batch
is small (about 50 MB below) and the same read with
`RowSelectionPolicy::Selectors` succeeds.
This is related to #7973, but it is a separate trigger. In #7973 the
selected rows of a batch really hold more than 2 GiB. Here they hold about 50
MB, and the Mask strategy multiplies that by the selection stride. Even if
#7973 is fixed by emitting partial batches, the mask path currently requires
`read_records(chunk_rows)` to return exactly `chunk_rows` (`read_mask_batch` in
`arrow_reader/mod.rs`), so it would need its own handling.
The mask also costs memory before it fails: the repro below peaks at about 2
GB RSS on a 5 KB file.
**To Reproduce**
Cargo.toml:
```toml
[dependencies]
arrow-array = "58.4"
arrow-schema = "58.4"
parquet = { version = "58.4", default-features = false, features = ["arrow",
"zstd"] }
bytes = "1"
```
src/main.rs:
```rust
use std::sync::Arc;
use arrow_array::{RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema};
use bytes::Bytes;
use parquet::arrow::arrow_reader::{
ParquetRecordBatchReaderBuilder, RowSelection, RowSelectionPolicy,
RowSelector,
};
use parquet::arrow::ArrowWriter;
use parquet::basic::{Compression, ZstdLevel};
use parquet::file::properties::WriterProperties;
const ROWS: usize = 500_000;
const VALUE_LEN: usize = 6_000;
const STRIDE: usize = 50; // keep 1 row out of every 50
const BATCH_SIZE: usize = 8192;
fn main() {
// 500k rows of a 6 KB string: ~3 GB of string data, a few hundred KB on
disk.
let schema = Arc::new(Schema::new(vec![Field::new("s", DataType::Utf8,
false)]));
let value = "x".repeat(VALUE_LEN);
let props = WriterProperties::builder()
.set_compression(Compression::ZSTD(ZstdLevel::default()))
.set_max_row_group_row_count(Some(ROWS))
.build();
let mut buf = Vec::new();
let mut writer = ArrowWriter::try_new(&mut buf, schema.clone(),
Some(props)).unwrap();
for _ in 0..ROWS / 10_000 {
let col =
StringArray::from_iter_values(std::iter::repeat_n(value.as_str(), 10_000));
writer
.write(&RecordBatch::try_new(schema.clone(),
vec![Arc::new(col)]).unwrap())
.unwrap();
}
writer.close().unwrap();
let file = Bytes::from(buf);
println!("file size: {} bytes", file.len());
// Select every 50th row: 10k selected rows, i.e. ~60 MB of selected
strings.
let selection: RowSelection = (0..ROWS / STRIDE)
.flat_map(|_| [RowSelector::select(1), RowSelector::skip(STRIDE -
1)])
.collect::<Vec<_>>()
.into();
for policy in [
RowSelectionPolicy::Selectors,
RowSelectionPolicy::default(), // Auto { threshold: 32 } -> picks
Mask here
] {
let reader = ParquetRecordBatchReaderBuilder::try_new(file.clone())
.unwrap()
.with_batch_size(BATCH_SIZE)
.with_row_selection(selection.clone())
.with_row_selection_policy(policy)
.build()
.unwrap();
let mut rows = 0;
let result = reader
.map(|batch| batch.map(|b| rows += b.num_rows()))
.collect::<Result<Vec<_>, _>>();
match result {
Ok(_) => println!("{policy:?}: ok, {rows} rows"),
Err(e) => println!("{policy:?}: error after {rows} rows: {e}"),
}
}
}
```
Output (`cargo run --release`):
```
file size: 4991 bytes
Selectors: ok, 10000 rows
Auto { threshold: 32 }: error after 0 rows: Parquet argument error: Parquet
error: index overflow decoding byte array
```
The same output on `main` (`c07655f8e986834f2aebf9de46ba1c02ac58d32a`).
With 1 row in 50 selected, the selectors alternate `select(1)` / `skip(49)`,
which averages 25 rows per selector. That is below the threshold of 32, so Auto
resolves to Mask. With `batch_size = 8192`, one mask chunk then spans about
409,600 rows × 6,000 bytes ≈ 2.46 GB, which is more than `i32::MAX`.
**Expected behavior**
The choice between Mask and Selectors should affect only performance, never
whether a read succeeds. The `Selectors` read above returns all 10,000 rows,
and `Auto` should too.
**Possible fixes**
- Bound a mask chunk by rows decoded (for example, a small multiple of
`batch_size`), not only by rows selected. This would emit smaller batches when
the selection is sparse.
- Have `Auto` prefer Selectors when the projection contains variable-width
(byte array) leaves and the selection density makes the mask span much larger
than `batch_size`, similar to how `should_force_selectors` already overrides
Mask when pages are skipped.
- Once the byte-array decoders can stop early on offset overflow (#7973),
let the mask path accept a short `read_records` and continue with the rest of
the chunk.
Relevant code: `next_mask_chunk_non_empty` in
[`parquet/src/arrow/arrow_reader/selection/cursor.rs`](https://github.com/apache/arrow-rs/blob/c07655f8e986834f2aebf9de46ba1c02ac58d32a/parquet/src/arrow/arrow_reader/selection/cursor.rs#L239-L270)
(on 58.4.0: `MaskCursor::next_mask_chunk` in `selection.rs`).
**Workaround**
Use `RowSelectionPolicy::Selectors`. In DataFusion, set
`datafusion.execution.parquet.force_filter_selections = true`. This costs CPU
on dense selections with short runs, which is exactly the case the mask
strategy exists for.
**Additional context**
Reproduced with parquet 58.4.0 and with `main` at
`c07655f8e986834f2aebf9de46ba1c02ac58d32a` (macOS aarch64). In practice we hit
it through DataFusion's parquet scan, where the selection comes from
pushed-down predicate evaluation and dynamic filters. The column there was a
list of short strings with many elements per row, rather than one long string.
--
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]
