This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-3247-9a2111a914e9b17fdc039700186056ec14fae606 in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit 615af68da85f2cc6e5139aa180c939214de91d32 Author: Anoop Johnson <[email protected]> AuthorDate: Sat Sep 26 09:28:36 2026 +0000 fix(expr): decode BYTE_ARRAY page-index bounds by field type (#3247) * fix(expr): decode BYTE_ARRAY page-index bounds by field type The page index evaluator decoded every BYTE_ARRAY min/max bound as a UTF-8 string via String::from_utf8(..).unwrap(), panicking during scan pruning whenever a binary (or byte-encoded decimal) column carried a predicate. Convert the bound according to the field's primitive type instead: string, binary, or decimal. * PR feedback * Round2 feedback * Fix trace --- .../src/expr/visitors/page_index_evaluator.rs | 376 +++++++++++++++++++-- 1 file changed, 343 insertions(+), 33 deletions(-) diff --git a/crates/iceberg/src/expr/visitors/page_index_evaluator.rs b/crates/iceberg/src/expr/visitors/page_index_evaluator.rs index bd901eda5..37d9c6a23 100644 --- a/crates/iceberg/src/expr/visitors/page_index_evaluator.rs +++ b/crates/iceberg/src/expr/visitors/page_index_evaluator.rs @@ -191,6 +191,7 @@ impl<'a> PageIndexEvaluator<'a> { }; let Some(page_filter) = Self::apply_predicate_to_column_index( + field_id, predicate, field_type, column_index, @@ -238,6 +239,7 @@ impl<'a> PageIndexEvaluator<'a> { } fn apply_predicate_to_column_index<F>( + field_id: i32, predicate: F, field_type: &PrimitiveType, column_index: &ColumnIndexMetaData, @@ -339,46 +341,96 @@ impl<'a> PageIndexEvaluator<'a> { ) }) .collect(), - ColumnIndexMetaData::BYTE_ARRAY(idx) => idx - .min_values_iter() - .zip(idx.max_values_iter()) - .enumerate() - .zip(row_counts.iter()) - .map(|((i, (min, max)), &row_count)| { - predicate( - min.map(|val| { - Datum::new( - field_type.clone(), - PrimitiveLiteral::String(String::from_utf8(val.to_vec()).unwrap()), - ) - }), - max.map(|val| { - Datum::new( - field_type.clone(), - PrimitiveLiteral::String(String::from_utf8(val.to_vec()).unwrap()), - ) - }), + ColumnIndexMetaData::BYTE_ARRAY(idx) => { + // Parquet stores Iceberg string and binary bounds as BYTE_ARRAY. + // Any other field type on a BYTE_ARRAY column (e.g. a non-spec + // BYTE_ARRAY decimal) can't be decoded safely, so skip page + // pruning for it rather than risk pruning pages that match. + if !matches!(field_type, PrimitiveType::String | PrimitiveType::Binary) { + tracing::debug!( + field_id, + %field_type, + "Skipping page-index pruning: BYTE_ARRAY column index on a non-string/binary field" + ); + 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 + // mid-UTF-8-sequence) means this column's page index can't + // be trusted, so skip pruning for the whole column rather + // than abort the scan. + let (min, max) = match ( + min.map(|val| Self::byte_array_bound_to_datum(field_type, val)) + .transpose(), + max.map(|val| Self::byte_array_bound_to_datum(field_type, val)) + .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 BYTE_ARRAY page bound" + ); + return Ok(None); + } + }; + + page_filter.push(predicate( + min, + max, PageNullCount::from_row_and_null_counts(row_count, idx.null_count(i)), - ) - }) - .collect(), - ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(_) => { - return Err(Error::new( - ErrorKind::FeatureUnsupported, - "unsupported 'FIXED_LEN_BYTE_ARRAY' index type in column_index", - )); + )?); + } + + Ok(page_filter) } - ColumnIndexMetaData::INT96(_) => { - return Err(Error::new( - ErrorKind::FeatureUnsupported, - "unsupported 'INT96' index type in column_index", - )); + // 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(_) => { + tracing::debug!( + field_id, + %field_type, + "Skipping page-index pruning: unsupported FIXED_LEN_BYTE_ARRAY or INT96 column index" + ); + return Ok(None); } }; Ok(Some(result?)) } + /// Converts a `BYTE_ARRAY` page bound into a [`Datum`] according to the + /// field's primitive type. Parquet stores Iceberg `string` and `binary` + /// bounds as `BYTE_ARRAY`. + fn byte_array_bound_to_datum(field_type: &PrimitiveType, bytes: &[u8]) -> Result<Datum> { + match field_type { + PrimitiveType::String => { + let value = std::str::from_utf8(bytes).map_err(|err| { + Error::new(ErrorKind::DataInvalid, "Invalid UTF-8 in string page bound") + .with_source(err) + })?; + Ok(Datum::string(value)) + } + PrimitiveType::Binary => Ok(Datum::binary(bytes.to_vec())), + // Defensive: production callers only invoke this for String/Binary. + _ => Err(Error::new( + ErrorKind::Unexpected, + format!("Unsupported primitive type for BYTE_ARRAY page bound: {field_type}"), + )), + } + } + fn visit_inequality( &mut self, reference: &BoundReference, @@ -779,7 +831,9 @@ mod tests { use std::collections::HashMap; use std::sync::Arc; - use arrow_array::{ArrayRef, Float32Array, RecordBatch, StringArray}; + use arrow_array::{ + ArrayRef, FixedSizeBinaryArray, Float32Array, LargeBinaryArray, RecordBatch, StringArray, + }; use arrow_schema::{DataType, Field, Schema as ArrowSchema}; use parquet::arrow::ArrowWriter; use parquet::arrow::arrow_reader::{ @@ -896,6 +950,92 @@ mod tests { Ok((metadata, temp_file)) } + /// Creates a single-column parquet file whose `col_binary` page bounds hold + /// non-UTF-8 bytes, backing the binary path through `BYTE_ARRAY` column + /// indexes. Page 0 is `[0x01]`, page 1 is `[0xff, 0x00]`. + fn create_binary_parquet_file() -> Result<(Arc<ParquetMetaData>, NamedTempFile)> { + let arrow_schema = Arc::new(ArrowSchema::new(vec![Field::new( + "col_binary", + DataType::LargeBinary, + 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) + .build(); + + let mut writer = ArrowWriter::try_new(file, arrow_schema.clone(), Some(props)).unwrap(); + + // Each page holds 1024 rows of a single value, one page per byte pattern. + let pages: [&[u8]; 2] = [&[0x01], &[0xff, 0x00]]; + + for value in pages { + let array = Arc::new(LargeBinaryArray::from_iter_values(std::iter::repeat_n( + value, 1024, + ))) 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 parquet file with a `FixedSizeBinary(2)` column, + /// producing `FIXED_LEN_BYTE_ARRAY` page bounds. Page 0 is `[0x01, 0x01]`, + /// page 1 is `[0x02, 0x02]`. + fn create_fixed_len_byte_array_parquet_file() -> Result<(Arc<ParquetMetaData>, NamedTempFile)> { + let arrow_schema = Arc::new(ArrowSchema::new(vec![Field::new( + "col_fixed", + DataType::FixedSizeBinary(2), + 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) + .build(); + + let mut writer = ArrowWriter::try_new(file, arrow_schema.clone(), Some(props)).unwrap(); + + // Each page holds 1024 rows of a single value, one page per byte pattern. + let pages: [&[u8]; 2] = [&[0x01, 0x01], &[0x02, 0x02]]; + + 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)) + } + /// Get the test metadata components for testing fn get_test_metadata( metadata: &ParquetMetaData, @@ -1185,6 +1325,176 @@ mod tests { Ok(()) } + #[test] + fn eval_inequality_prunes_binary_pages_with_non_utf8_bounds() -> Result<()> { + let (metadata, _temp_file) = create_binary_parquet_file()?; + let (column_index, offset_index, row_group_metadata) = get_test_metadata(&metadata); + + let iceberg_schema = Arc::new( + Schema::builder() + .with_fields([Arc::new(NestedField::new( + 1, + "col_binary", + Type::Primitive(PrimitiveType::Binary), + true, + ))]) + .build()?, + ); + let field_id_map = HashMap::from_iter([(1, 0)]); + + // Page 0 bounds are [0x01], page 1 bounds are [0xff, 0x00]; only page 1 + // exceeds [0x80]. Decoding [0xff, 0x00] as UTF-8 would panic. + let filter = Reference::new("col_binary") + .greater_than(Datum::binary(vec![0x80u8])) + .bind(iceberg_schema.clone(), false)?; + + let result = PageIndexEvaluator::eval( + &filter, + &column_index, + &offset_index, + row_group_metadata, + &field_id_map, + iceberg_schema.as_ref(), + )?; + + let expected = vec![RowSelector::skip(1024), RowSelector::select(1024)]; + + assert_eq!(result, expected); + + Ok(()) + } + + #[test] + fn eval_skips_pruning_for_byte_array_decimal_bounds() -> Result<()> { + // A non-spec writer can store a decimal column as BYTE_ARRAY. Those + // bounds can't be decoded safely (Parquet may truncate them, and + // truncation only preserves lexicographic byte order, not decimal + // order), so the evaluator skips page pruning rather than prune pages + // that might match. + let (metadata, _temp_file) = create_binary_parquet_file()?; + let (column_index, offset_index, row_group_metadata) = get_test_metadata(&metadata); + + let iceberg_schema = Arc::new( + Schema::builder() + .with_fields([Arc::new(NestedField::new( + 1, + "col_decimal", + Type::Primitive(PrimitiveType::Decimal { + precision: 10, + scale: 2, + }), + true, + ))]) + .build()?, + ); + let field_id_map = HashMap::from_iter([(1, 0)]); + + // A predicate that would prune every page if the bounds were decoded. + let filter = Reference::new("col_decimal") + .greater_than(Datum::decimal_with_precision( + crate::spec::decimal_utils::decimal_from_i128_with_scale(99999, 2), + 10, + )?) + .bind(iceberg_schema.clone(), false)?; + + let result = PageIndexEvaluator::eval( + &filter, + &column_index, + &offset_index, + row_group_metadata, + &field_id_map, + iceberg_schema.as_ref(), + )?; + + // Both 1024-row pages survive: no page pruning is applied. + assert_eq!(result, vec![RowSelector::select(2048)]); + + Ok(()) + } + + #[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. + let (metadata, _temp_file) = create_fixed_len_byte_array_parquet_file()?; + let (column_index, offset_index, row_group_metadata) = get_test_metadata(&metadata); + + let iceberg_schema = Arc::new( + Schema::builder() + .with_fields([Arc::new(NestedField::new( + 1, + "col_fixed", + Type::Primitive(PrimitiveType::Fixed(2)), + true, + ))]) + .build()?, + ); + let field_id_map = HashMap::from_iter([(1, 0)]); + + // A predicate that would prune every page if the bounds were decoded. + let filter = Reference::new("col_fixed") + .greater_than(Datum::fixed(vec![0xffu8, 0xff])) + .bind(iceberg_schema.clone(), false)?; + + let result = PageIndexEvaluator::eval( + &filter, + &column_index, + &offset_index, + row_group_metadata, + &field_id_map, + iceberg_schema.as_ref(), + )?; + + // All rows survive: no page pruning is applied for FIXED_LEN_BYTE_ARRAY. + assert_eq!(result, vec![RowSelector::select(2048)]); + + Ok(()) + } + + #[test] + fn eval_skips_pruning_for_non_utf8_string_bound() -> Result<()> { + // A writer can truncate a string min/max stat mid-UTF-8-sequence, + // leaving a BYTE_ARRAY page bound that isn't valid UTF-8. Decoding it + // as a string fails, so the evaluator skips page pruning for the column + // rather than aborting the whole file read. + let (metadata, _temp_file) = create_binary_parquet_file()?; + let (column_index, offset_index, row_group_metadata) = get_test_metadata(&metadata); + + let iceberg_schema = Arc::new( + Schema::builder() + .with_fields([Arc::new(NestedField::new( + 1, + "col_string", + Type::Primitive(PrimitiveType::String), + true, + ))]) + .build()?, + ); + let field_id_map = HashMap::from_iter([(1, 0)]); + + // Page 1's bound [0xff, 0x00] isn't valid UTF-8. A predicate that would + // otherwise prune every page must not: the undecodable bound forces a + // skip rather than an error. + let filter = Reference::new("col_string") + .greater_than(Datum::string("zzz")) + .bind(iceberg_schema.clone(), false)?; + + let result = PageIndexEvaluator::eval( + &filter, + &column_index, + &offset_index, + row_group_metadata, + &field_id_map, + iceberg_schema.as_ref(), + )?; + + // Both pages survive: no page pruning is applied when a bound won't decode. + assert_eq!(result, vec![RowSelector::select(2048)]); + + Ok(()) + } + #[test] fn eval_starts_with_error_float_col() -> Result<()> { let (metadata, _temp_file) = create_test_parquet_file()?;
