feat(reader): page-index pruning for FIXED_LEN_BYTE_ARRAY decimals

[anoopj]

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

Throwaway benchmark

Run with:

cargo test -p iceberg --release --lib \
  expr::visitors::page_index_evaluator::tests::bench_flba_decimal_page_pruning \
  -- --ignored --nocapture

#[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(())
}

Are these changes tested?

Unit tests in page_index_evaluator

[laskoviymishka]

The core of this is right — decoding FLBA decimals as big-endian two's-complement i128 with sign extension matches the row-group stats path, the decode is panic-free (i128_from_be_bytes returns Option, no unchecked indexing), and skipping pruning for the whole column on any undecodable bound is the correct conservative failure mode rather than aborting a file that's valid for other readers. The width-equality guard is sound too: a valid full-width FLBA bound always has exactly type_length bytes.

A few things I'd tighten, none blocking:

• The width-guard rationale (doc + test comment) oversells how reachable it is — parquet-rs and parquet-mr don't truncate DECIMAL-annotated bounds, so for a real Iceberg decimal column this guard never fires. Worth reframing as defence-in-depth against a non-conforming writer so it doesn't read as "Iceberg tables hit this."
• The min/max decode is the same closure written twice to log one error; hoisting a single Copy closure and keeping the transpose() flow collapses it. While there, DataInvalid reads as corruption for what's really legitimate truncation.
• Test coverage leans entirely on precision-30 (13-byte) non-null data — a null-only page and the width extremes (precision 38 at 16 bytes, and a narrow-width-widened-precision decode) would pin down the None handling and the sign extension.

One observation, not for this PR: the row-group stats path (get_parquet_stat_max_as_datum) still calls i128_from_be_bytes with no width check, so it's now less strict than this new page-index path. Worth a note or a follow-up for symmetry.

Specifics are in the inline comments.

[laskoviymishka]

The guard itself is right — a valid full-width FLBA bound always has exactly type_length bytes, so a short bound can only mean truncation or a broken writer, and skipping is the safe response. The wording oversells how reachable it is, though: parquet-rs and parquet-mr's BinaryTruncator don't truncate DECIMAL-annotated bounds, so for a real Iceberg decimal column this never fires. I'd reframe both this doc comment and the test comment as defence-in-depth against a non-conforming writer that stores a decimal as plain FLBA, so nobody reads it as "Iceberg tables hit this path."

[anoopj]

In the followup, I reworded the helper doc and the inline arm comment to frame the guard as defence against a non-conforming writer.

[laskoviymishka]

The min and max arms are the same closure written out twice, and we build two Results just to log one %err and bail. decode captures only field_type and type_length (both Copy), so the closure is Copy and you can hoist it and reuse it on both sides:

let decode = |b: &[u8]| Self::fixed_len_byte_array_decimal_bound_to_datum(field_type, b, type_length);
let (min, max) = match (min.map(decode).transpose(), max.map(decode).transpose()) {
    (Ok(min), Ok(max)) => (min, max),
    (Err(err), _) | (_, Err(err)) => {
        tracing::debug!(field_id, %field_type, page_index = i, %err, "...");
        return Ok(None);
    }
};

While you're in here: the helper's DataInvalid reads as corruption, but a short bound is legitimate column_index_truncate_length behavior that we only ever log and skip on — I'd soften the kind (or have the helper return Option<Datum>) so a future ?-caller isn't misled.

[anoopj]

Thanks. Hoisted the decode into a single Copy closure. Kept Result<Datum> and DataInvalid: a short/undecodable bound isn't normal behavior for a DECIMAL-annotated FLBA, so DataInvalid is the right signal. Happy to move both to Option in a follow-up if you'd prefer.

[laskoviymishka]

len as usize is safe here (positive i32), but usize::try_from(len).ok() keeps it lossless-by-construction. Bigger picture: this runs for every predicate column though only the FLBA arm reads it, and it widens the shared helper signature by a positional arg — I'd lean toward deriving type_length lazily inside the arm (or passing the ColumnDescriptor) so the next metadata-hungry arm doesn't add another. Not blocking either way.

[anoopj]

Kept len as usize: it's guarded by .filter(|&len| len > 0), so the cast is on a positive i32 and lossless

[laskoviymishka]

All four new tests use non-null data, so the None-bound / null-count path on this arm (the idx.null_count(i) + None, None into the predicate) isn't exercised. I'd add one test with a null-only page — an IS NULL, or a > x that should keep the null page — so a bad refactor of the None handling here gets caught.

[anoopj]

Added eval_is_null_selects_all_null_fixed_len_byte_array_decimal_page to cover this.

[laskoviymishka]

The new tests all land on precision 30 (13-byte bounds), so the width extremes aren't covered. Two worth adding: a precision-38 column (16-byte, hits the type_length == 16 boundary and the full-width negative extreme), and a narrow-width-widened-precision case — a column physically stored below the field precision's minimum width (e.g. a 9-byte FLBA read as decimal(30,2)) — which is where the sign extension in i128_from_be_bytes actually earns its keep. Not blocking, but they'd pin the decode down across the width range.

[anoopj]

Added both.