You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by gi...@apache.org on 2026/10/13 17:43:22 UTC
(iceberg-rust) 01/01: fix(reader): rewrite NOT out of the read predicate before pruning (#3416)
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-b154603a710d730d2f43f8af4db257cc137237ef
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit 9192b5ee1fa9386876b0dd2b7defe6d20a4dcef8
Author: Anoop Johnson <an...@apache.org>
AuthorDate: Tue Oct 13 09:53:05 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(