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-3416-a853455db9ed693ceae74f5d2edc1aa346169542 in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit a269dad56bd90d6b5f8cda026f25274642b78baf Author: Anoop Johnson <[email protected]> AuthorDate: Wed Oct 14 22:12:51 2026 +0000 fix(reader): rewrite NOT out of the read predicate before pruning (#3416) * fix(reader): rewrite NOT out of the read predicate before pruning The reader fans a FileScanTask's predicate to the row-group metrics evaluator and the page-index evaluator, both conservative (inclusive) filters that cannot soundly negate a NOT node. The row-group evaluator's naive `!inner` flips a MIGHT_MATCH bound to CANT_MATCH. For NOT(x > 5) over a row group spanning [0, 10] that prunes a group which does contain matching rows (silent data loss). The page-index evaluator rejects NOT outright, aborting the read. A FileScanTask built or deserialized outside TableScan carries an un-normalized predicate, so a raw NOT reaches both. Rewrite NOT out of the predicate once, where it is first bound in the read pipeline, so every consumer sees a NOT-free predicate. * PR feedback --- crates/iceberg/src/arrow/reader/pipeline.rs | 8 ++ crates/iceberg/src/arrow/reader/row_filter.rs | 155 +++++++++++++++++++++ .../expr/visitors/row_group_metrics_evaluator.rs | 38 ++++- 3 files changed, 198 insertions(+), 3 deletions(-) diff --git a/crates/iceberg/src/arrow/reader/pipeline.rs b/crates/iceberg/src/arrow/reader/pipeline.rs index 84b4467d3..afc93e1df 100644 --- a/crates/iceberg/src/arrow/reader/pipeline.rs +++ b/crates/iceberg/src/arrow/reader/pipeline.rs @@ -521,6 +521,14 @@ impl FileScanTaskReader { } if let Some(predicate) = final_predicate { + // Normalize NOT out of the predicate once for every consumer below. The + // conservative row-group and page-index evaluators cannot soundly negate + // a NOT node (negating a "might match" bound can prune matching rows, and + // the page-index evaluator rejects NOT outright). A FileScanTask built or + // deserialized outside TableScan carries an un-normalized predicate, so + // rewrite it here. Matches TableScan::with_filter. + let predicate = predicate.rewrite_not(); + let (iceberg_field_ids, field_id_map) = ArrowReader::build_field_id_set_and_map( record_batch_stream_builder.parquet_schema(), record_batch_stream_builder.schema(), diff --git a/crates/iceberg/src/arrow/reader/row_filter.rs b/crates/iceberg/src/arrow/reader/row_filter.rs index 4a99914a3..331e7f965 100644 --- a/crates/iceberg/src/arrow/reader/row_filter.rs +++ b/crates/iceberg/src/arrow/reader/row_filter.rs @@ -975,6 +975,161 @@ mod tests { ) } + /// A `FileScanTask` built or deserialized outside `TableScan` carries an + /// un-normalized predicate, so a raw `Not(...)` can reach the reader. + #[tokio::test] + async fn test_not_predicate_read_returns_matching_rows() { + use arrow_array::Int32Array; + + let schema = int_schema(); + let arrow_schema = Arc::new(ArrowSchema::new(vec![ + Field::new("x", DataType::Int32, false).with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + "1".to_string(), + )])), + ])); + + let tmp_dir = TempDir::new().unwrap(); + let table_location = tmp_dir.path().to_str().unwrap().to_string(); + let file_path = format!("{table_location}/not_pred.parquet"); + + let file = File::create(&file_path).unwrap(); + let mut writer = ArrowWriter::try_new(file, arrow_schema.clone(), None).unwrap(); + let batch = RecordBatch::try_new(arrow_schema.clone(), vec![Arc::new(Int32Array::from( + (0..=10).collect::<Vec<i32>>(), + ))]) + .unwrap(); + writer.write(&batch).expect("Writing batch"); + writer.close().unwrap(); + + // NOT(x > 5) == x <= 5, bound without rewriting, as a non-TableScan caller + // would supply it. + let predicate = (!Reference::new("x").greater_than(Datum::int(5))) + .bind(schema.clone(), false) + .unwrap(); + assert!( + matches!(predicate, BoundPredicate::Not(_)), + "test must drive a raw Not(...) node" + ); + + let reader = ArrowReaderBuilder::new(FileIO::new_with_fs(), Runtime::current()) + .with_row_group_filtering_enabled(true) + .build(); + + let task = FileScanTask::builder() + .with_file_size_in_bytes(std::fs::metadata(&file_path).unwrap().len()) + .with_start(0) + .with_length(0) + .with_data_file_path(file_path) + .with_data_file_format(DataFileFormat::Parquet) + .with_schema(schema.clone()) + .with_project_field_ids(vec![1]) + .with_predicate(Some(predicate)) + .with_case_sensitive(false) + .build() + .unwrap(); + + let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream; + let batches = reader + .read(tasks) + .unwrap() + .stream() + .try_collect::<Vec<RecordBatch>>() + .await + .unwrap(); + + let mut values = Vec::new(); + for batch in &batches { + let col = batch + .column(0) + .as_primitive::<arrow_array::types::Int32Type>(); + values.extend(col.values().iter().copied()); + } + + assert_eq!(values, (0..=5).collect::<Vec<i32>>()); + } + + #[tokio::test] + async fn test_not_predicate_read_through_page_index() { + use arrow_array::Int32Array; + use parquet::file::reader::{FileReader, SerializedFileReader}; + use parquet::file::serialized_reader::ReadOptionsBuilder; + + let schema = int_schema(); + let arrow_schema = Arc::new(ArrowSchema::new(vec![ + Field::new("x", DataType::Int32, false).with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + "1".to_string(), + )])), + ])); + + let tmp_dir = TempDir::new().unwrap(); + let table_location = tmp_dir.path().to_str().unwrap().to_string(); + let file_path = format!("{table_location}/not_pred_pages.parquet"); + + let file = File::create(&file_path).unwrap(); + let mut writer = ArrowWriter::try_new(file, arrow_schema.clone(), None).unwrap(); + let batch = RecordBatch::try_new(arrow_schema.clone(), vec![Arc::new(Int32Array::from( + (0..100).collect::<Vec<i32>>(), + ))]) + .unwrap(); + writer.write(&batch).expect("Writing batch"); + writer.close().unwrap(); + + { + let f = File::open(&file_path).unwrap(); + let opts = ReadOptionsBuilder::new().with_page_index().build(); + let rdr = SerializedFileReader::new_with_options(f, opts).unwrap(); + assert!( + rdr.metadata().column_index().is_some(), + "fixture must carry a column index so the page-index path runs" + ); + } + + // NOT(x >= 50) == x < 50, as a non-TableScan caller would supply it. + let predicate = (!Reference::new("x").greater_than_or_equal_to(Datum::int(50))) + .bind(schema.clone(), false) + .unwrap(); + assert!(matches!(predicate, BoundPredicate::Not(_))); + + let reader = ArrowReaderBuilder::new(FileIO::new_with_fs(), Runtime::current()) + .with_row_group_filtering_enabled(false) + .with_row_selection_enabled(true) + .build(); + + let task = FileScanTask::builder() + .with_file_size_in_bytes(std::fs::metadata(&file_path).unwrap().len()) + .with_start(0) + .with_length(0) + .with_data_file_path(file_path) + .with_data_file_format(DataFileFormat::Parquet) + .with_schema(schema.clone()) + .with_project_field_ids(vec![1]) + .with_predicate(Some(predicate)) + .with_case_sensitive(false) + .build() + .unwrap(); + + let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream; + let batches = reader + .read(tasks) + .unwrap() + .stream() + .try_collect::<Vec<RecordBatch>>() + .await + .unwrap(); + + let mut values = Vec::new(); + for batch in &batches { + let col = batch + .column(0) + .as_primitive::<arrow_array::types::Int32Type>(); + values.extend(col.values().iter().copied()); + } + + assert_eq!(values, (0..50).collect::<Vec<i32>>()); + } + #[test] fn test_row_group_pruning_rejects_invalid_candidate() { let schema = int_schema(); diff --git a/crates/iceberg/src/expr/visitors/row_group_metrics_evaluator.rs b/crates/iceberg/src/expr/visitors/row_group_metrics_evaluator.rs index ab8d81ce2..3a1170d4a 100644 --- a/crates/iceberg/src/expr/visitors/row_group_metrics_evaluator.rs +++ b/crates/iceberg/src/expr/visitors/row_group_metrics_evaluator.rs @@ -207,8 +207,14 @@ impl BoundPredicateVisitor for RowGroupMetricsEvaluator<'_> { Ok(lhs || rhs) } - fn not(&mut self, inner: bool) -> Result<bool> { - Ok(!inner) + fn not(&mut self, _: bool) -> Result<bool> { + // Negating a conservative "might match" bound is unsound (it can turn into + // "can't match" and prune row groups that do match), so NOT must be + // rewritten out of the predicate before it reaches this evaluator. + Err(Error::new( + ErrorKind::Unexpected, + "NOT unsupported at this point. NOT-rewrite should be performed first", + )) } fn is_null(&mut self, reference: &BoundReference, _predicate: &BoundPredicate) -> Result<bool> { @@ -543,9 +549,9 @@ mod tests { use rand::Rng; use super::RowGroupMetricsEvaluator; - use crate::Result; use crate::expr::{Bind, Reference}; use crate::spec::{Datum, NestedField, PrimitiveType, Schema, Type}; + use crate::{ErrorKind, Result}; #[test] fn eval_matches_no_rows_for_empty_row_group() -> Result<()> { @@ -591,6 +597,32 @@ mod tests { Ok(()) } + #[test] + fn eval_error_for_not_predicate_requires_rewrite() -> Result<()> { + // The evaluator cannot soundly negate a conservative bound, so a NOT node + // that reaches it (one that was not rewritten away first) is rejected + // rather than silently mispruning a matching row group. + let row_group_metadata = create_row_group_metadata(1, 1, None, 1, None)?; + + let (iceberg_schema_ref, field_id_map) = build_iceberg_schema_and_field_map()?; + + let filter = (!Reference::new("col_float").greater_than(Datum::float(5.0_f32))) + .bind(iceberg_schema_ref.clone(), false)?; + + let err = RowGroupMetricsEvaluator::eval( + &filter, + &row_group_metadata, + &field_id_map, + iceberg_schema_ref.as_ref(), + ) + .unwrap_err(); + + assert_eq!(err.kind(), ErrorKind::Unexpected); + assert!(err.message().contains("NOT")); + + Ok(()) + } + #[test] fn eval_false_for_meta_all_null_filter_not_null() -> Result<()> { let row_group_metadata = create_row_group_metadata(
