anoopj opened a new pull request, #3328:
URL: https://github.com/apache/iceberg-rust/pull/3328
## Which issue does this PR close?
No separate issue. This is the `FIXED_LEN_BYTE_ARRAY` follow-up called out
in the description of #3314
## What changes are included in this PR?
Decimals with precision > 18 are stored as `FIXED_LEN_BYTE_ARRAY`, which the
page-index evaluator skipped entirely, so inequality and `IN` predicates on
them got no page pruning.
- Decode `FIXED_LEN_BYTE_ARRAY` decimal page bounds with
`i128_from_be_bytes`, matching the row-group statistics path
(`get_parquet_stat_{min,max}_as_datum`), so the two pruning layers agree on the
P > 18 encoding.
- Guard against truncated `ColumnIndex` bounds: a bound whose width differs
from the column's declared `type_length` signals a
`column_index_truncate_length` truncation, and a truncated two's-complement
byte prefix decodes to the wrong number. Reject it so the column skips pruning
rather than decode a wrong bound. parquet-rs never truncates decimal/float16
column indexes (`can_truncate_value` excludes them, since their sort order
differs from raw byte order), so this only defends against files written by
other engines.
- Non-decimal `FIXED_LEN_BYTE_ARRAY` fields (fixed, uuid) are still not
decoded here, so they skip page pruning rather than risk pruning matching pages.
### Benchmark
A throwaway benchmark on a 1,000,000-row single-column `decimal(30,2)` file
(precision > 18 → `FIXED_LEN_BYTE_ARRAY`), 100 pages in one row group, read
fully in memory so timing is decode-only. A `col > threshold` predicate sweeps
selectivity. "read full file" is today's behavior (FLBA decimals skip page
pruning); "read ms" applies the page-index `RowSelection` this change produces.
```
1,000,000 rows in 100 pages, 1 row group
read full file : 3.431 ms
predicate (keep) | rows read | eval us | read ms | speedup
1% | 10,000 | 17.5 | 0.038 | 90.0x
10% | 100,000 | 17.1 | 0.330 | 10.4x
50% | 500,000 | 16.6 | 1.629 | 2.1x
100% | 1,000,000 | 16.3 | 3.397 | 1.0x
```
<details>
<summary>Throwaway benchmark</summary>
Run with:
```
cargo test -p iceberg --release --lib \
expr::visitors::page_index_evaluator::tests::bench_flba_decimal_page_pruning \
-- --ignored --nocapture
```
```rust
#[test]
#[ignore = "throwaway benchmark"]
fn bench_flba_decimal_page_pruning() -> Result<()> {
use std::time::Instant;
let precision: u8 = 30; // > 18 -> FIXED_LEN_BYTE_ARRAY
let scale: i8 = 2;
let num_rows: usize = 1_000_000;
let page_rows: usize = 10_000;
// Single-column FLBA decimal file with monotonically increasing values,
// so page bounds partition the value range and a `>` predicate prunes
// every page below the threshold.
let arrow_schema = Arc::new(ArrowSchema::new(vec![Field::new(
"col_decimal",
DataType::Decimal128(precision, scale),
false,
)]));
let temp_file = NamedTempFile::new().unwrap();
let file = temp_file.reopen().unwrap();
let props = WriterProperties::builder()
.set_data_page_row_count_limit(page_rows)
.set_write_batch_size(page_rows)
.build(); // default max row group size (1,048,576) keeps all rows
in one group
let mut writer = ArrowWriter::try_new(file, arrow_schema.clone(),
Some(props)).unwrap();
let mut start = 0usize;
while start < num_rows {
let end = (start + page_rows).min(num_rows);
let values: Vec<i128> = (start as i128..end as i128).collect();
let array = Arc::new(
Decimal128Array::from_iter_values(values)
.with_precision_and_scale(precision, scale)
.unwrap(),
) as ArrayRef;
let batch = RecordBatch::try_new(arrow_schema.clone(),
vec![array]).unwrap();
writer.write(&batch).unwrap();
start = end;
}
writer.close().unwrap();
// Read the file into memory so timing measures decode, not disk I/O.
let data = bytes::Bytes::from(std::fs::read(temp_file.path()).unwrap());
let options =
ArrowReaderOptions::new().with_page_index_policy(PageIndexPolicy::Required);
let metadata = ParquetRecordBatchReaderBuilder::try_new_with_options(
data.clone(),
options.clone(),
)
.unwrap()
.metadata()
.clone();
let (iceberg_schema, field_id_map) =
build_decimal_schema_and_field_map(precision as u32, scale as u32)?;
let (column_index, offset_index, row_group_metadata) =
get_test_metadata(&metadata);
let num_pages = offset_index[0].page_locations().len();
let read_all = || -> usize {
ParquetRecordBatchReaderBuilder::try_new_with_options(data.clone(),
options.clone())
.unwrap()
.build()
.unwrap()
.map(|b| b.unwrap().num_rows())
.sum()
};
let read_pruned = |selection: &Vec<RowSelector>| -> usize {
ParquetRecordBatchReaderBuilder::try_new_with_options(data.clone(),
options.clone())
.unwrap()
.with_row_selection(selection.clone().into())
.build()
.unwrap()
.map(|b| b.unwrap().num_rows())
.sum()
};
let runs = 30;
assert_eq!(read_all(), num_rows); // warmup + sanity
let t = Instant::now();
for _ in 0..runs {
std::hint::black_box(read_all());
}
let all_ms = t.elapsed().as_secs_f64() * 1000.0 / runs as f64;
println!("\n=== FLBA decimal page-pruning benchmark (precision 30, scale
2) ===");
println!("{num_rows} rows in {num_pages} pages, 1 row group");
println!(
"read full file : {all_ms:.3} ms (today's behavior: FLBA decimals
skip pruning)\n"
);
println!(
"{:>18} | {:>10} | {:>9} | {:>10} | {:>8}",
"predicate (keep)", "rows read", "eval us", "read ms", "speedup"
);
// Sweep selectivity: a `>` predicate keeping the top k% of values.
for keep_pct in [1u32, 10, 50, 100] {
let threshold = (num_rows as i128) * (100 - keep_pct as i128) / 100;
let filter = Reference::new("col_decimal")
.greater_than(decimal_datum(threshold, scale as u32, precision
as u32)?)
.bind(iceberg_schema.clone(), false)?;
// Time eval() -- the work this change adds per row group per scan.
let eval_iters = 2000;
let t = Instant::now();
let mut selection = Vec::new();
for _ in 0..eval_iters {
selection = PageIndexEvaluator::eval(
&filter,
&column_index,
&offset_index,
row_group_metadata,
&field_id_map,
iceberg_schema.as_ref(),
)?;
}
let eval_us = t.elapsed().as_secs_f64() * 1e6 / eval_iters as f64;
let pruned_rows = read_pruned(&selection);
let t = Instant::now();
for _ in 0..runs {
std::hint::black_box(read_pruned(&selection));
}
let pruned_ms = t.elapsed().as_secs_f64() * 1000.0 / runs as f64;
println!(
"{:>15}% | {:>10} | {:>9.1} | {:>10.3} | {:>7.1}x",
keep_pct, pruned_rows, eval_us, pruned_ms, all_ms / pruned_ms
);
}
println!();
Ok(())
}
```
</details>
## Are these changes tested?
Unit tests in `page_index_evaluator`
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
