diff --git a/crates/iceberg/src/expr/visitors/page_index_evaluator.rs b/crates/iceberg/src/expr/visitors/page_index_evaluator.rs index 4e51042336..2fa6fd7653 100644 --- a/crates/iceberg/src/expr/visitors/page_index_evaluator.rs +++ b/crates/iceberg/src/expr/visitors/page_index_evaluator.rs @@ -596,7 +596,9 @@ impl<'a> PageIndexEvaluator<'a> { let bound = if use_lower_bound { min } else { max }; if let Some(bound) = bound { - if cmp_fn(&bound, datum) { + // A bound that can't be compared with the literal (e.g. a + // file column narrower than the field type) might match. + if bound.partial_cmp(datum).is_none() || cmp_fn(&bound, datum) { return Ok(true); } @@ -937,11 +939,13 @@ impl BoundPredicateVisitor for PageIndexEvaluator<'_> { // if all values are outside the bounds, no rows can match return Ok(false); } - (Some(min), _) if !literals.iter().any(|datum| datum.ge(&min)) => { + // `all(lt)` rather than `!any(ge)`, so that a bound that + // can't be compared with the literals keeps the page. + (Some(min), _) if literals.iter().all(|datum| datum.lt(&min)) => { // if no values are within the min bound, no rows can match return Ok(false); } - (_, Some(max)) if !literals.iter().any(|datum| datum.le(&max)) => { + (_, Some(max)) if literals.iter().all(|datum| datum.gt(&max)) => { // if no values are within the max bound, no rows can match return Ok(false); } @@ -973,8 +977,8 @@ mod tests { use std::sync::Arc; use arrow_array::{ - ArrayRef, Decimal128Array, FixedSizeBinaryArray, Float32Array, LargeBinaryArray, - RecordBatch, StringArray, + ArrayRef, Decimal128Array, FixedSizeBinaryArray, Float32Array, Int32Array, + LargeBinaryArray, RecordBatch, StringArray, UInt32Array, }; use arrow_schema::{DataType, Field, Schema as ArrowSchema}; use parquet::arrow::ArrowWriter; @@ -987,7 +991,7 @@ mod tests { use tempfile::NamedTempFile; use super::PageIndexEvaluator; - use crate::expr::{Bind, Reference}; + use crate::expr::{Bind, Predicate, Reference}; use crate::spec::decimal_utils::decimal_from_i128_with_scale; use crate::spec::{Datum, NestedField, PrimitiveType, Schema, Type}; use crate::{ErrorKind, Result}; @@ -1247,10 +1251,35 @@ mod tests { precision: u8, scale: i8, pages: &[Option], + ) -> Result<(Arc, NamedTempFile)> { + let pages: Vec = pages + .iter() + .map(|page| { + Arc::new( + match page { + Some(unscaled) => { + Decimal128Array::from_iter_values(std::iter::repeat_n(*unscaled, 1024)) + } + None => Decimal128Array::from(vec![None; 1024]), + } + .with_precision_and_scale(precision, scale) + .unwrap(), + ) as ArrayRef + }) + .collect(); + create_single_column_parquet_file(&pages) + } + + /// Creates a single-column parquet file named `col` with one page per + /// 1024-row array in `pages`. Used to back an Iceberg field with a + /// narrower physical column, e.g. a `long` field with an `INT32` column + /// written before an `int` -> `long` promotion. + fn create_single_column_parquet_file( + pages: &[ArrayRef], ) -> Result<(Arc, NamedTempFile)> { let arrow_schema = Arc::new(ArrowSchema::new(vec![Field::new( - "col_decimal", - DataType::Decimal128(precision, scale), + "col", + pages[0].data_type().clone(), true, )])); @@ -1265,17 +1294,7 @@ mod tests { let mut writer = ArrowWriter::try_new(file, arrow_schema.clone(), Some(props)).unwrap(); for page in pages { - let array = Arc::new( - match page { - Some(unscaled) => { - Decimal128Array::from_iter_values(std::iter::repeat_n(*unscaled, 1024)) - } - None => Decimal128Array::from(vec![None; 1024]), - } - .with_precision_and_scale(precision, scale) - .unwrap(), - ) as ArrayRef; - let batch = RecordBatch::try_new(arrow_schema.clone(), vec![array]).unwrap(); + let batch = RecordBatch::try_new(arrow_schema.clone(), vec![page.clone()]).unwrap(); // Write rows one at a time so the writer splits into per-value pages. for i in 0..batch.num_rows() { writer.write(&batch.slice(i, 1)).unwrap(); @@ -1292,6 +1311,16 @@ mod tests { Ok((metadata, temp_file)) } + /// Creates a single-column parquet file with one 1024-row page per value, + /// using `to_array` to build each page. + fn create_parquet_file_from_values( + values: &[T], + to_array: impl Fn(Vec) -> ArrayRef, + ) -> Result<(Arc, NamedTempFile)> { + let pages: Vec = values.iter().map(|&v| to_array(vec![v; 1024])).collect(); + create_single_column_parquet_file(&pages) + } + /// Get the test metadata components for testing fn get_test_metadata( metadata: &ParquetMetaData, @@ -2312,6 +2341,142 @@ mod tests { Ok(()) } + #[test] + fn eval_keeps_int32_pages_of_promoted_long_field() -> Result<()> { + // An `int` -> `long` promotion leaves older files with INT32 page + // bounds, which can't be compared with a `long` literal. + let (metadata, _temp_file) = create_parquet_file_from_values(&[100, 200, 300, 400], |v| { + Arc::new(Int32Array::from(v)) + })?; + + assert_keeps_all_pages(&metadata, PrimitiveType::Long, 4096, [ + Reference::new("col").less_than(Datum::long(250)), + Reference::new("col").less_than_or_equal_to(Datum::long(250)), + Reference::new("col").greater_than(Datum::long(250)), + Reference::new("col").greater_than_or_equal_to(Datum::long(250)), + // A literal outside the `int` range. + Reference::new("col").less_than(Datum::long(i64::from(i32::MAX) + 1)), + Reference::new("col").is_in([Datum::long(200), Datum::long(400)]), + ]) + } + + #[test] + fn eval_keeps_float_pages_of_promoted_double_field() -> Result<()> { + // A `float` -> `double` promotion leaves older files with FLOAT page + // bounds, which can't be compared with a `double` literal. + let (metadata, _temp_file) = + create_parquet_file_from_values(&[1.0_f32, 2.0, 3.0, 4.0], |v| { + Arc::new(Float32Array::from(v)) + })?; + + assert_keeps_all_pages(&metadata, PrimitiveType::Double, 4096, [ + Reference::new("col").less_than(Datum::double(2.5)), + Reference::new("col").less_than_or_equal_to(Datum::double(2.5)), + Reference::new("col").greater_than(Datum::double(2.5)), + Reference::new("col").greater_than_or_equal_to(Datum::double(2.5)), + Reference::new("col").is_in([Datum::double(2.0), Datum::double(4.0)]), + ]) + } + + #[test] + fn eval_keeps_uint32_pages_of_long_field() -> Result<()> { + // Parquet `UINT_32` columns map to `long` but have INT32 page bounds. + // The last page mixes both values, so its raw signed max is smaller + // than its min and a sign-extending widening would prune it. + let mixed: Vec = [100_u32, 3_000_000_000].repeat(512); + let (metadata, _temp_file) = create_single_column_parquet_file(&[ + Arc::new(UInt32Array::from(vec![100_u32; 1024])), + Arc::new(UInt32Array::from(vec![3_000_000_000_u32; 1024])), + Arc::new(UInt32Array::from(mixed)), + ])?; + + let column_index = &metadata.column_index().unwrap()[0][0]; + let parquet::file::page_index::column_index::ColumnIndexMetaData::INT32(idx) = column_index + else { + panic!("expected INT32 column index, got {column_index:?}"); + }; + let bounds: Vec<_> = idx.min_values_iter().zip(idx.max_values_iter()).collect(); + assert_eq!(bounds.len(), 3); + assert_eq!(bounds[2], (Some(&100), Some(&(3_000_000_000_u32 as i32)))); + + assert_keeps_all_pages(&metadata, PrimitiveType::Long, 3072, [ + Reference::new("col").equal_to(Datum::long(3_000_000_000_i64)), + Reference::new("col").greater_than(Datum::long(2_000_000_000)), + Reference::new("col").is_in([ + Datum::long(3_000_000_000_i64), + Datum::long(4_000_000_000_i64), + ]), + ]) + } + + #[test] + fn eval_keeps_pages_with_bounds_incomparable_to_predicate() -> Result<()> { + // INT32 bounds on a `timestamp` field (e.g. a file whose column was + // written as `date`) can't be compared with the predicate literal. + let (metadata, _temp_file) = create_parquet_file_from_values(&[100, 200, 300, 400], |v| { + Arc::new(Int32Array::from(v)) + })?; + + assert_keeps_all_pages(&metadata, PrimitiveType::Timestamp, 4096, [ + Reference::new("col").less_than(Datum::timestamp_micros(250)), + Reference::new("col").less_than_or_equal_to(Datum::timestamp_micros(250)), + Reference::new("col").greater_than(Datum::timestamp_micros(250)), + Reference::new("col").greater_than_or_equal_to(Datum::timestamp_micros(250)), + Reference::new("col").equal_to(Datum::timestamp_micros(250)), + Reference::new("col") + .is_in([Datum::timestamp_micros(200), Datum::timestamp_micros(400)]), + ]) + } + + /// Asserts that each of `filters`, bound to a single `field_type` field + /// over the column in `metadata`, selects all `row_count` rows. Pages + /// whose bounds can't be compared with the predicate might match. + fn assert_keeps_all_pages( + metadata: &ParquetMetaData, + field_type: PrimitiveType, + row_count: usize, + filters: impl IntoIterator, + ) -> Result<()> { + let (column_index, offset_index, row_group_metadata) = get_test_metadata(metadata); + let (iceberg_schema, field_id_map) = build_single_field_schema_and_field_map(field_type)?; + + for filter in filters { + let filter = filter.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::select(row_count)], "{filter}"); + } + + Ok(()) + } + + /// Builds a schema holding a single optional field `col` (id 1) of + /// `field_type`, mapped to parquet column 0. + fn build_single_field_schema_and_field_map( + field_type: PrimitiveType, + ) -> Result<(Arc, HashMap)> { + let iceberg_schema = Arc::new( + Schema::builder() + .with_fields([Arc::new(NestedField::new( + 1, + "col", + Type::Primitive(field_type), + false, + ))]) + .build()?, + ); + + Ok((iceberg_schema, HashMap::from_iter([(1, 0)]))) + } + /// 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 {