Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
laskoviymishka merged 5 commits into from
Oct 9, 2026
Merged

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

Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
063610b
feat(reader): page-index pruning for FIXED_LEN_BYTE_ARRAY decimals
anoopj Oct 1, 2026
60fc5c1
Merge branch 'main' into page-index-flba-decimal
anoopj Oct 3, 2026
0404238
Merge branch 'main' into page-index-flba-decimal
anoopj Oct 7, 2026
7256559
Merge branch 'main' into page-index-flba-decimal
anoopj Oct 8, 2026
a8383cb
Merge branch 'main' into page-index-flba-decimal
anoopj Oct 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
322 changes: 314 additions & 8 deletions crates/iceberg/src/expr/visitors/page_index_evaluator.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ use parquet::file::page_index::offset_index::OffsetIndexMetaData;

use crate::expr::visitors::bound_predicate_visitor::{BoundPredicateVisitor, visit};
use crate::expr::{BoundPredicate, BoundReference};
use crate::spec::decimal_utils::i128_from_be_bytes;
use crate::spec::{Datum, PrimitiveLiteral, PrimitiveType, Schema};
use crate::{Error, ErrorKind, Result};

Expand Down Expand Up @@ -167,6 +168,17 @@ impl<'a> PageIndexEvaluator<'a> {
return self.select_all_rows();
};

// Declared byte width of a FIXED_LEN_BYTE_ARRAY column, used to detect
// truncated page-index bounds when decoding decimals. `None` when the
// column metadata is absent or reports no fixed width.
let type_length = self
.row_group_metadata
.columns()
.get(parquet_column_index)
.map(|column| column.column_descr().type_length())
.filter(|&len| len > 0)
.map(|len| len as usize);

@laskoviymishka laskoviymishka Oct 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 anoopj Oct 9, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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


let row_counts = {
// Caches row count calculations for columns that appear multiple times in
// the predicate
Expand Down Expand Up @@ -194,6 +206,7 @@ impl<'a> PageIndexEvaluator<'a> {
field_id,
predicate,
field_type,
type_length,
column_index,
&row_counts,
)?
Expand Down Expand Up @@ -242,6 +255,7 @@ impl<'a> PageIndexEvaluator<'a> {
field_id: i32,
predicate: F,
field_type: &PrimitiveType,
type_length: Option<usize>,
column_index: &ColumnIndexMetaData,
row_counts: &[usize],
) -> Result<Option<Vec<bool>>>
Expand Down Expand Up @@ -394,14 +408,89 @@ impl<'a> PageIndexEvaluator<'a> {

Ok(page_filter)
}
ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(idx) => {
// Parquet stores Iceberg decimals with precision > 18 as
// FIXED_LEN_BYTE_ARRAY. Other field types on such a column
// (fixed, uuid) aren't decoded here, so skip page pruning for
// them rather than risk pruning pages that match.
if !matches!(field_type, PrimitiveType::Decimal { .. }) {
tracing::debug!(
field_id,
%field_type,
"Skipping page-index pruning: FIXED_LEN_BYTE_ARRAY column index on a non-decimal field"
);
return Ok(None);
}

let Some(type_length) = type_length else {
tracing::debug!(
field_id,
%field_type,
"Skipping page-index pruning: unknown FIXED_LEN_BYTE_ARRAY decimal column width"
);
return Ok(None);
};

let mut page_filter = Vec::with_capacity(row_counts.len());
for ((i, (min, max)), &row_count) in idx
.min_values_iter()
.zip(idx.max_values_iter())
.enumerate()
.zip(row_counts.iter())
{
// A bound that won't decode (e.g. a min/max stat truncated
// by column_index_truncate_length, whose byte prefix no
// longer preserves the two's-complement decimal ordering)
// means this column's page index can't be trusted, so skip
// pruning for the whole column rather than abort the scan.

@laskoviymishka laskoviymishka Oct 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 anoopj Oct 9, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

let (min, max) = match (
min.map(|val| {
Self::fixed_len_byte_array_decimal_bound_to_datum(
field_type,
val,
type_length,
)
})
.transpose(),
max.map(|val| {
Self::fixed_len_byte_array_decimal_bound_to_datum(
field_type,
val,
type_length,
)
})
.transpose(),
) {
(Ok(min), Ok(max)) => (min, max),
(Err(err), _) | (_, Err(err)) => {
tracing::debug!(
field_id,
%field_type,
page_index = i,
%err,
"Skipping page-index pruning: undecodable FIXED_LEN_BYTE_ARRAY decimal page bound"
);
return Ok(None);
}
};

page_filter.push(predicate(
min,
max,
PageNullCount::from_row_and_null_counts(row_count, idx.null_count(i)),
)?);
}

Ok(page_filter)
}
// Column index types we can't interpret: skip page pruning rather
// than abort the scan. Row-group filtering and the Arrow row filter
// still apply the predicate.
ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(_) | ColumnIndexMetaData::INT96(_) => {
ColumnIndexMetaData::INT96(_) => {
tracing::debug!(
field_id,
%field_type,
"Skipping page-index pruning: unsupported FIXED_LEN_BYTE_ARRAY or INT96 column index"
"Skipping page-index pruning: unsupported INT96 column index"
);
return Ok(None);
}
Expand Down Expand Up @@ -455,6 +544,40 @@ impl<'a> PageIndexEvaluator<'a> {
}
}

/// Converts a `FIXED_LEN_BYTE_ARRAY` decimal page bound into a [`Datum`].
/// Parquet stores Iceberg decimals with precision > 18 as big-endian
/// two's-complement `FIXED_LEN_BYTE_ARRAY` of width `type_length`.
///
/// Returns an error for a bound whose width differs from `type_length`: that

@laskoviymishka laskoviymishka Oct 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 anoopj Oct 9, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

/// signals a `column_index_truncate_length` truncation, and a truncated byte
/// prefix decodes to the wrong number under two's-complement.
fn fixed_len_byte_array_decimal_bound_to_datum(
field_type: &PrimitiveType,
bytes: &[u8],
type_length: usize,
) -> Result<Datum> {
if bytes.len() != type_length {
return Err(Error::new(
ErrorKind::DataInvalid,
format!(
"Truncated FIXED_LEN_BYTE_ARRAY decimal page bound: {} bytes (expected {type_length})",
bytes.len()
),
));
}

let value = i128_from_be_bytes(bytes).ok_or_else(|| {
Error::new(
ErrorKind::DataInvalid,
format!("Can't convert bytes to i128: {bytes:?}"),
)
})?;
Ok(Datum::new(
field_type.clone(),
PrimitiveLiteral::Int128(value),
))
}

fn visit_inequality(
&mut self,
reference: &BoundReference,
Expand Down Expand Up @@ -1062,11 +1185,58 @@ mod tests {
Ok((metadata, temp_file))
}

/// Creates a single-column `FixedSizeBinary(width)` parquet file whose column
/// index is truncated to `truncate_length` bytes (`< width`), producing
/// `FIXED_LEN_BYTE_ARRAY` page bounds narrower than the declared `type_length`.
/// Writes one 1024-row page per value in `pages`; each value must be `width`
/// bytes.
fn create_truncated_flba_file(
width: i32,
truncate_length: usize,
pages: &[&[u8]],
) -> Result<(Arc<ParquetMetaData>, NamedTempFile)> {
let arrow_schema = Arc::new(ArrowSchema::new(vec![Field::new(
"col_fixed",
DataType::FixedSizeBinary(width),
true,
)]));

let temp_file = NamedTempFile::new().unwrap();
let file = temp_file.reopen().unwrap();

let props = WriterProperties::builder()
.set_data_page_row_count_limit(1024)
.set_write_batch_size(512)
.set_column_index_truncate_length(Some(truncate_length))
.build();

let mut writer = ArrowWriter::try_new(file, arrow_schema.clone(), Some(props)).unwrap();

for value in pages {
let array = Arc::new(
FixedSizeBinaryArray::try_from_iter(std::iter::repeat_n(*value, 1024)).unwrap(),
) as ArrayRef;
let batch = RecordBatch::try_new(arrow_schema.clone(), vec![array]).unwrap();
for i in 0..batch.num_rows() {
writer.write(&batch.slice(i, 1)).unwrap();
}
}

writer.close().unwrap();

let file = temp_file.reopen().unwrap();
let options = ArrowReaderOptions::new().with_page_index_policy(PageIndexPolicy::Required);
let reader = ParquetRecordBatchReaderBuilder::try_new_with_options(file, options).unwrap();
let metadata = reader.metadata().clone();

Ok((metadata, temp_file))
}

/// Creates a single-column `Decimal128(precision, scale)` parquet file.
/// Parquet encodes precision <= 9 as `INT32` and precision 10..=18 as
/// `INT64`, so `precision` selects which page-index encoding is exercised.
/// Writes one 1024-row page per value in `unscaled`, so page bounds
/// partition the value range.
/// Parquet encodes precision <= 9 as `INT32`, precision 10..=18 as `INT64`,
/// and precision > 18 as `FIXED_LEN_BYTE_ARRAY`, so `precision` selects which
/// page-index encoding is exercised. Writes one 1024-row page per value in
/// `unscaled`, so page bounds partition the value range.
fn create_decimal_parquet_file(
precision: u8,
scale: i8,
Expand Down Expand Up @@ -1490,8 +1660,9 @@ mod tests {
#[test]
fn eval_skips_pruning_for_fixed_len_byte_array() -> Result<()> {
// FIXED_LEN_BYTE_ARRAY page indexes back spec-conforming fixed, uuid,
// and decimal(P > 18) columns. The evaluator can't interpret those
// bounds, so it skips page pruning rather than abort the scan.
// and decimal(P > 18) columns. Only decimals are decoded; fixed and uuid
// bounds aren't, so the evaluator skips page pruning for them rather than
// abort the scan.
let (metadata, _temp_file) = create_fixed_len_byte_array_parquet_file()?;
let (column_index, offset_index, row_group_metadata) = get_test_metadata(&metadata);

Expand Down Expand Up @@ -1880,6 +2051,141 @@ mod tests {
Ok(())
}

#[test]
fn eval_inequality_prunes_fixed_len_byte_array_decimal_pages() -> Result<()> {

@laskoviymishka laskoviymishka Oct 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 anoopj Oct 9, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added eval_is_null_selects_all_null_fixed_len_byte_array_decimal_page to cover this.

// precision 30 -> Parquet FIXED_LEN_BYTE_ARRAY page bounds.
let (metadata, _temp_file) = create_decimal_parquet_file(30, 2, &[100, 200, 300, 400])?;
let (column_index, offset_index, row_group_metadata) = get_test_metadata(&metadata);
let (iceberg_schema, field_id_map) = build_decimal_schema_and_field_map(30, 2)?;

// Pages hold 1.00, 2.00, 3.00, 4.00. `> 2.50` keeps the pages whose
// upper bound exceeds 2.50 (3.00 and 4.00).
let filter = Reference::new("col_decimal")
.greater_than(decimal_datum(250, 2, 30)?)
.bind(iceberg_schema.clone(), false)?;

let result = PageIndexEvaluator::eval(
&filter,
&column_index,
&offset_index,
row_group_metadata,
&field_id_map,
iceberg_schema.as_ref(),
)?;

assert_eq!(result, vec![
RowSelector::skip(2048),
RowSelector::select(2048)
]);

Ok(())
}

#[test]
fn eval_in_prunes_fixed_len_byte_array_decimal_pages() -> Result<()> {
// precision 30 -> Parquet FIXED_LEN_BYTE_ARRAY page bounds.
let (metadata, _temp_file) = create_decimal_parquet_file(30, 2, &[100, 200, 300, 400])?;
let (column_index, offset_index, row_group_metadata) = get_test_metadata(&metadata);
let (iceberg_schema, field_id_map) = build_decimal_schema_and_field_map(30, 2)?;

// Pages hold 1.00, 2.00, 3.00, 4.00. IN (2.00, 4.00) keeps only the
// pages whose single value is one of the literals.
let filter = Reference::new("col_decimal")
.is_in([decimal_datum(200, 2, 30)?, decimal_datum(400, 2, 30)?])
.bind(iceberg_schema.clone(), false)?;

let result = PageIndexEvaluator::eval(
&filter,
&column_index,
&offset_index,
row_group_metadata,
&field_id_map,
iceberg_schema.as_ref(),
)?;

assert_eq!(result, vec![
RowSelector::skip(1024),
RowSelector::select(1024),
RowSelector::skip(1024),
RowSelector::select(1024),
]);

Ok(())
}

#[test]
fn eval_inequality_prunes_negative_fixed_len_byte_array_decimal_pages() -> Result<()> {
// precision 30 -> Parquet FIXED_LEN_BYTE_ARRAY page bounds, spanning
// negative values. Negative decimals have a 0xff high byte in two's
// complement, so this only prunes correctly if the bounds are compared
// numerically rather than by raw byte order.
let (metadata, _temp_file) = create_decimal_parquet_file(30, 2, &[-400, -200, 100, 300])?;
let (column_index, offset_index, row_group_metadata) = get_test_metadata(&metadata);
let (iceberg_schema, field_id_map) = build_decimal_schema_and_field_map(30, 2)?;

// Pages hold -4.00, -2.00, 1.00, 3.00. `>= -2.00` keeps the pages whose
// upper bound is at least -2.00, including page 1 whose bound equals it.
let filter = Reference::new("col_decimal")
.greater_than_or_equal_to(decimal_datum(-200, 2, 30)?)
.bind(iceberg_schema.clone(), false)?;

let result = PageIndexEvaluator::eval(
&filter,
&column_index,
&offset_index,
row_group_metadata,
&field_id_map,
iceberg_schema.as_ref(),
)?;

assert_eq!(result, vec![
RowSelector::skip(1024),
RowSelector::select(3072)
]);

Ok(())
}

#[test]
fn eval_skips_pruning_for_truncated_fixed_len_byte_array_decimal_bound() -> Result<()> {

@laskoviymishka laskoviymishka Oct 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 anoopj Oct 9, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added both.

// A non-Iceberg writer can store a decimal as a plain FIXED_LEN_BYTE_ARRAY
// and truncate its column index (parquet-rs truncates FLBA bounds but not
// decimal/float16 ones). A bound narrower than the column's declared
// type_length can't be decoded as an i128 without changing its value, so
// the evaluator skips page pruning for the column rather than aborting the
// file read.
//
// Page 0 is 0.00, page 1 is 2.55; with full-width bounds `> 1.00` would
// prune page 0. The 13-byte values are truncated to 4-byte column-index
// bounds (!= type_length 13), so neither page is pruned instead.
let page0 = [0u8; 13]; // unscaled 0 -> 0.00
let mut page1 = [0u8; 13];
page1[12] = 0xff; // unscaled 255 -> 2.55
let (metadata, _temp_file) = create_truncated_flba_file(13, 4, &[&page0, &page1])?;
let (column_index, offset_index, row_group_metadata) = get_test_metadata(&metadata);

// The Iceberg schema calls the 13-byte column a Decimal(30,2).
let (iceberg_schema, field_id_map) = build_decimal_schema_and_field_map(30, 2)?;

let filter = Reference::new("col_decimal")
.greater_than(decimal_datum(100, 2, 30)?) // > 1.00
.bind(iceberg_schema.clone(), false)?;

let result = PageIndexEvaluator::eval(
&filter,
&column_index,
&offset_index,
row_group_metadata,
&field_id_map,
iceberg_schema.as_ref(),
)?;

// Truncated bounds -> no page pruning: every row survives.
assert_eq!(result, vec![RowSelector::select(2048)]);

Ok(())
}

/// Builds a decimal query [`Datum`] from an `unscaled` mantissa, matching
/// the field's `scale` and `precision`.
fn decimal_datum(unscaled: i128, scale: u32, precision: u32) -> Result<Datum> {
Expand Down
Loading