Skip to content

Commit a269dad

Browse files
authored
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
1 parent a853455 commit a269dad

3 files changed

Lines changed: 198 additions & 3 deletions

File tree

‎crates/iceberg/src/arrow/reader/pipeline.rs‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -521,6 +521,14 @@ impl FileScanTaskReader {
521521
}
522522

523523
if let Some(predicate) = final_predicate {
524+
// Normalize NOT out of the predicate once for every consumer below. The
525+
// conservative row-group and page-index evaluators cannot soundly negate
526+
// a NOT node (negating a "might match" bound can prune matching rows, and
527+
// the page-index evaluator rejects NOT outright). A FileScanTask built or
528+
// deserialized outside TableScan carries an un-normalized predicate, so
529+
// rewrite it here. Matches TableScan::with_filter.
530+
let predicate = predicate.rewrite_not();
531+
524532
let (iceberg_field_ids, field_id_map) = ArrowReader::build_field_id_set_and_map(
525533
record_batch_stream_builder.parquet_schema(),
526534
record_batch_stream_builder.schema(),

‎crates/iceberg/src/arrow/reader/row_filter.rs‎

Lines changed: 155 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -975,6 +975,161 @@ mod tests {
975975
)
976976
}
977977

978+
/// A `FileScanTask` built or deserialized outside `TableScan` carries an
979+
/// un-normalized predicate, so a raw `Not(...)` can reach the reader.
980+
#[tokio::test]
981+
async fn test_not_predicate_read_returns_matching_rows() {
982+
use arrow_array::Int32Array;
983+
984+
let schema = int_schema();
985+
let arrow_schema = Arc::new(ArrowSchema::new(vec![
986+
Field::new("x", DataType::Int32, false).with_metadata(HashMap::from([(
987+
PARQUET_FIELD_ID_META_KEY.to_string(),
988+
"1".to_string(),
989+
)])),
990+
]));
991+
992+
let tmp_dir = TempDir::new().unwrap();
993+
let table_location = tmp_dir.path().to_str().unwrap().to_string();
994+
let file_path = format!("{table_location}/not_pred.parquet");
995+
996+
let file = File::create(&file_path).unwrap();
997+
let mut writer = ArrowWriter::try_new(file, arrow_schema.clone(), None).unwrap();
998+
let batch = RecordBatch::try_new(arrow_schema.clone(), vec![Arc::new(Int32Array::from(
999+
(0..=10).collect::<Vec<i32>>(),
1000+
))])
1001+
.unwrap();
1002+
writer.write(&batch).expect("Writing batch");
1003+
writer.close().unwrap();
1004+
1005+
// NOT(x > 5) == x <= 5, bound without rewriting, as a non-TableScan caller
1006+
// would supply it.
1007+
let predicate = (!Reference::new("x").greater_than(Datum::int(5)))
1008+
.bind(schema.clone(), false)
1009+
.unwrap();
1010+
assert!(
1011+
matches!(predicate, BoundPredicate::Not(_)),
1012+
"test must drive a raw Not(...) node"
1013+
);
1014+
1015+
let reader = ArrowReaderBuilder::new(FileIO::new_with_fs(), Runtime::current())
1016+
.with_row_group_filtering_enabled(true)
1017+
.build();
1018+
1019+
let task = FileScanTask::builder()
1020+
.with_file_size_in_bytes(std::fs::metadata(&file_path).unwrap().len())
1021+
.with_start(0)
1022+
.with_length(0)
1023+
.with_data_file_path(file_path)
1024+
.with_data_file_format(DataFileFormat::Parquet)
1025+
.with_schema(schema.clone())
1026+
.with_project_field_ids(vec![1])
1027+
.with_predicate(Some(predicate))
1028+
.with_case_sensitive(false)
1029+
.build()
1030+
.unwrap();
1031+
1032+
let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream;
1033+
let batches = reader
1034+
.read(tasks)
1035+
.unwrap()
1036+
.stream()
1037+
.try_collect::<Vec<RecordBatch>>()
1038+
.await
1039+
.unwrap();
1040+
1041+
let mut values = Vec::new();
1042+
for batch in &batches {
1043+
let col = batch
1044+
.column(0)
1045+
.as_primitive::<arrow_array::types::Int32Type>();
1046+
values.extend(col.values().iter().copied());
1047+
}
1048+
1049+
assert_eq!(values, (0..=5).collect::<Vec<i32>>());
1050+
}
1051+
1052+
#[tokio::test]
1053+
async fn test_not_predicate_read_through_page_index() {
1054+
use arrow_array::Int32Array;
1055+
use parquet::file::reader::{FileReader, SerializedFileReader};
1056+
use parquet::file::serialized_reader::ReadOptionsBuilder;
1057+
1058+
let schema = int_schema();
1059+
let arrow_schema = Arc::new(ArrowSchema::new(vec![
1060+
Field::new("x", DataType::Int32, false).with_metadata(HashMap::from([(
1061+
PARQUET_FIELD_ID_META_KEY.to_string(),
1062+
"1".to_string(),
1063+
)])),
1064+
]));
1065+
1066+
let tmp_dir = TempDir::new().unwrap();
1067+
let table_location = tmp_dir.path().to_str().unwrap().to_string();
1068+
let file_path = format!("{table_location}/not_pred_pages.parquet");
1069+
1070+
let file = File::create(&file_path).unwrap();
1071+
let mut writer = ArrowWriter::try_new(file, arrow_schema.clone(), None).unwrap();
1072+
let batch = RecordBatch::try_new(arrow_schema.clone(), vec![Arc::new(Int32Array::from(
1073+
(0..100).collect::<Vec<i32>>(),
1074+
))])
1075+
.unwrap();
1076+
writer.write(&batch).expect("Writing batch");
1077+
writer.close().unwrap();
1078+
1079+
{
1080+
let f = File::open(&file_path).unwrap();
1081+
let opts = ReadOptionsBuilder::new().with_page_index().build();
1082+
let rdr = SerializedFileReader::new_with_options(f, opts).unwrap();
1083+
assert!(
1084+
rdr.metadata().column_index().is_some(),
1085+
"fixture must carry a column index so the page-index path runs"
1086+
);
1087+
}
1088+
1089+
// NOT(x >= 50) == x < 50, as a non-TableScan caller would supply it.
1090+
let predicate = (!Reference::new("x").greater_than_or_equal_to(Datum::int(50)))
1091+
.bind(schema.clone(), false)
1092+
.unwrap();
1093+
assert!(matches!(predicate, BoundPredicate::Not(_)));
1094+
1095+
let reader = ArrowReaderBuilder::new(FileIO::new_with_fs(), Runtime::current())
1096+
.with_row_group_filtering_enabled(false)
1097+
.with_row_selection_enabled(true)
1098+
.build();
1099+
1100+
let task = FileScanTask::builder()
1101+
.with_file_size_in_bytes(std::fs::metadata(&file_path).unwrap().len())
1102+
.with_start(0)
1103+
.with_length(0)
1104+
.with_data_file_path(file_path)
1105+
.with_data_file_format(DataFileFormat::Parquet)
1106+
.with_schema(schema.clone())
1107+
.with_project_field_ids(vec![1])
1108+
.with_predicate(Some(predicate))
1109+
.with_case_sensitive(false)
1110+
.build()
1111+
.unwrap();
1112+
1113+
let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream;
1114+
let batches = reader
1115+
.read(tasks)
1116+
.unwrap()
1117+
.stream()
1118+
.try_collect::<Vec<RecordBatch>>()
1119+
.await
1120+
.unwrap();
1121+
1122+
let mut values = Vec::new();
1123+
for batch in &batches {
1124+
let col = batch
1125+
.column(0)
1126+
.as_primitive::<arrow_array::types::Int32Type>();
1127+
values.extend(col.values().iter().copied());
1128+
}
1129+
1130+
assert_eq!(values, (0..50).collect::<Vec<i32>>());
1131+
}
1132+
9781133
#[test]
9791134
fn test_row_group_pruning_rejects_invalid_candidate() {
9801135
let schema = int_schema();

‎crates/iceberg/src/expr/visitors/row_group_metrics_evaluator.rs‎

Lines changed: 35 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -207,8 +207,14 @@ impl BoundPredicateVisitor for RowGroupMetricsEvaluator<'_> {
207207
Ok(lhs || rhs)
208208
}
209209

210-
fn not(&mut self, inner: bool) -> Result<bool> {
211-
Ok(!inner)
210+
fn not(&mut self, _: bool) -> Result<bool> {
211+
// Negating a conservative "might match" bound is unsound (it can turn into
212+
// "can't match" and prune row groups that do match), so NOT must be
213+
// rewritten out of the predicate before it reaches this evaluator.
214+
Err(Error::new(
215+
ErrorKind::Unexpected,
216+
"NOT unsupported at this point. NOT-rewrite should be performed first",
217+
))
212218
}
213219

214220
fn is_null(&mut self, reference: &BoundReference, _predicate: &BoundPredicate) -> Result<bool> {
@@ -543,9 +549,9 @@ mod tests {
543549
use rand::Rng;
544550

545551
use super::RowGroupMetricsEvaluator;
546-
use crate::Result;
547552
use crate::expr::{Bind, Reference};
548553
use crate::spec::{Datum, NestedField, PrimitiveType, Schema, Type};
554+
use crate::{ErrorKind, Result};
549555

550556
#[test]
551557
fn eval_matches_no_rows_for_empty_row_group() -> Result<()> {
@@ -591,6 +597,32 @@ mod tests {
591597
Ok(())
592598
}
593599

600+
#[test]
601+
fn eval_error_for_not_predicate_requires_rewrite() -> Result<()> {
602+
// The evaluator cannot soundly negate a conservative bound, so a NOT node
603+
// that reaches it (one that was not rewritten away first) is rejected
604+
// rather than silently mispruning a matching row group.
605+
let row_group_metadata = create_row_group_metadata(1, 1, None, 1, None)?;
606+
607+
let (iceberg_schema_ref, field_id_map) = build_iceberg_schema_and_field_map()?;
608+
609+
let filter = (!Reference::new("col_float").greater_than(Datum::float(5.0_f32)))
610+
.bind(iceberg_schema_ref.clone(), false)?;
611+
612+
let err = RowGroupMetricsEvaluator::eval(
613+
&filter,
614+
&row_group_metadata,
615+
&field_id_map,
616+
iceberg_schema_ref.as_ref(),
617+
)
618+
.unwrap_err();
619+
620+
assert_eq!(err.kind(), ErrorKind::Unexpected);
621+
assert!(err.message().contains("NOT"));
622+
623+
Ok(())
624+
}
625+
594626
#[test]
595627
fn eval_false_for_meta_all_null_filter_not_null() -> Result<()> {
596628
let row_group_metadata = create_row_group_metadata(

0 commit comments

Comments
 (0)