comphead commented on code in PR #3302:
URL: https://github.com/apache/iceberg-rust/pull/3302#discussion_r4287799308
##########
crates/iceberg/src/arrow/reader/predicate_visitor.rs:
##########
@@ -195,6 +197,236 @@ impl BoundPredicateVisitor for CollectFieldIdVisitor {
}
}
+/// Returns the residual of `predicate` for one data file: each leaf on a
top-level field that is
+/// missing from the file becomes `AlwaysTrue` or `AlwaysFalse`, based on the
value projection
+/// returns for that field. That value is the identity partition value,
otherwise the field's
+/// `initial-default`, and every row of the file holds it. Leaves on missing
fields with neither
+/// keep the null handling of the row filter and the page index evaluator.
+///
+/// Leaves on nested fields are kept, because a nested field also reads as
null in any row where
+/// an ancestor struct is null.
+pub(super) fn residual_for_missing_fields(
+ predicate: BoundPredicate,
+ predicate_field_ids: &HashSet<i32>,
+ field_id_map: &HashMap<i32, usize>,
+ schema: &Schema,
+ partition_spec: Option<&PartitionSpec>,
+ partition: Option<&Struct>,
+) -> Result<BoundPredicate> {
+ if predicate_field_ids
+ .iter()
+ .all(|id| field_id_map.contains_key(id))
+ {
+ return Ok(predicate);
+ }
+
+ let partition_constants = match (partition_spec, partition) {
+ (Some(spec), Some(data)) => constants_map(spec, data, schema)?,
+ _ => HashMap::new(),
+ };
+
+ let mut constant_field_ids = HashSet::new();
Review Comment:
Nit: this set holds at most `predicate_field_ids.len()` entries, so
`HashSet::with_capacity(predicate_field_ids.len())` bounds it.
##########
crates/iceberg/src/scan/mod.rs:
##########
@@ -3109,4 +3120,145 @@ mod tests {
.collect();
assert_eq!(y, (1100..1200).collect::<Vec<i64>>());
}
+
+ /// Appends one data file holding `columns`, written with the table's
current schema.
+ async fn append_data_file(
+ catalog: &impl Catalog,
+ table: Table,
+ columns: Vec<ArrayRef>,
+ ) -> Table {
+ let schema = table.metadata().current_schema().clone();
+ let batch =
+
RecordBatch::try_new(Arc::new(schema_to_arrow_schema(&schema).unwrap()),
columns)
+ .unwrap();
+ let rolling_file_writer_builder =
RollingFileWriterBuilder::new_with_default_file_size(
+ ParquetWriterBuilder::new(WriterProperties::default(), schema),
+ table.file_io().clone(),
+ DefaultLocationGenerator::new(table.metadata()).unwrap(),
+ DefaultFileNameGenerator::new(
+ Uuid::new_v4().to_string(),
+ None,
+ DataFileFormat::Parquet,
+ ),
+ );
+ let mut writer =
DataFileWriterBuilder::new(rolling_file_writer_builder)
+ .build(None)
+ .await
+ .unwrap();
+ writer.write(batch).await.unwrap();
+ let data_files = writer.close().await.unwrap();
+
+ let tx = Transaction::new(&table);
+ let tx = tx
+ .fast_append()
+ .add_data_files(data_files)
+ .apply(tx)
+ .unwrap();
+ tx.commit(catalog).await.unwrap()
+ }
+
+ /// Mirrors Java's
`TestFilterPushDown.testFilterPushdownOnInitialDefaultColumnAbsentFromFile`.
+ /// The first file is written before `c` is added with `initial-default`
'US', so its row
+ /// reads `c` as 'US'. The second file stores `c`.
+ #[tokio::test]
+ async fn test_scan_filter_on_initial_default_column_absent_from_file() {
Review Comment:
Removing the `residual_for_missing_fields` call from `pipeline.rs` fails
this test, but it also fails the two reader tests in `row_filter.rs`, and no
other mutation I tried fails only this one. It repeats the reader-level
initial-default matrix (`c = 'US'`, `IS NOT NULL`, `IS NULL`, row selection off
and on). What it adds is the `update_schema` plumbing, which this PR does not
change. Consider trimming it to one or two predicates, or dropping it and
keeping the Java mirror in the reader test.
##########
crates/iceberg/src/arrow/reader/predicate_visitor.rs:
##########
@@ -836,4 +1073,248 @@ mod tests {
);
assert!(!result.is_null(2));
}
+
+ /// Returns the residual of `predicate` for a data file that stores only
field 1.
+ fn residual_for_file_with_field_1(
+ schema: &SchemaRef,
+ predicate: Predicate,
+ partition: Option<(&PartitionSpec, &Struct)>,
+ ) -> BoundPredicate {
+ let predicate = predicate.bind(schema.clone(), true).unwrap();
+ let mut collector = CollectFieldIdVisitor {
+ field_ids: HashSet::default(),
+ };
+ visit(&mut collector, &predicate).unwrap();
+ residual_for_missing_fields(
+ predicate,
+ &collector.field_ids(),
+ &HashMap::from([(1, 0)]),
+ schema,
+ partition.map(|(spec, _)| spec),
+ partition.map(|(_, data)| data),
+ )
+ .unwrap()
+ }
+
+ /// Asserts that each predicate in `always_true` has the residual
`AlwaysTrue` and each in
+ /// `always_false` has `AlwaysFalse`, for a file that doesn't store the
field `field`.
+ fn assert_residuals_for_missing_field_with_initial_default(
+ field: NestedField,
+ always_true: Vec<Predicate>,
+ always_false: Vec<Predicate>,
+ ) {
+ let schema = Arc::new(
+ Schema::builder()
+ .with_fields(vec![
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
+ field.into(),
+ ])
+ .build()
+ .unwrap(),
+ );
+
+ let cases = always_true
+ .into_iter()
+ .map(|predicate| (predicate, BoundPredicate::AlwaysTrue))
+ .chain(
+ always_false
+ .into_iter()
+ .map(|predicate| (predicate, BoundPredicate::AlwaysFalse)),
+ );
+ for (predicate, expected) in cases {
+ let message = predicate.to_string();
+ let residual = residual_for_file_with_field_1(&schema, predicate,
None);
+ assert_eq!(residual, expected, "{message}");
+ }
+ }
+
+ /// Mirrors Java's
`TestMetricsRowGroupFilter.testColumnNotInFileWithInitialDefault`.
+ #[test]
+ fn test_residual_for_missing_string_field_with_initial_default() {
+ let country = || Reference::new("country");
+ let string = Datum::string;
+ assert_residuals_for_missing_field_with_initial_default(
+ NestedField::optional(99, "country",
Type::Primitive(PrimitiveType::String))
+ .with_initial_default(Literal::string("US")),
+ vec![
+ country().equal_to(string("US")),
+ country().is_not_null(),
+ country().not_equal_to(string("CA")),
+ country().is_in([string("US"), string("MX")]),
+ country().is_not_in([string("CA"), string("MX")]),
+ country().greater_than_or_equal_to(string("US")),
+ country().less_than(string("ZW")),
+ country().starts_with(string("U")),
+ country().not_starts_with(string("X")),
+ ],
+ vec![
+ country().equal_to(string("CA")),
+ country().is_null(),
+ country().not_equal_to(string("US")),
+ country().is_in([string("CA"), string("MX")]),
+ country().is_not_in([string("US"), string("MX")]),
+ country().less_than(string("AD")),
+ country().greater_than(string("US")),
+ country().starts_with(string("X")),
+ country().not_starts_with(string("U")),
+ ],
+ );
+ }
+
+ /// Mirrors Java's
`TestMetricsRowGroupFilter.testDateColumnNotInFileWithInitialDefault`.
+ #[test]
+ fn test_residual_for_missing_date_field_with_initial_default() {
Review Comment:
This test looks redundant. The two mutations that failed it (`less_than` and
`greater_than` returning the leaf unchanged) also fail
`test_residual_for_missing_string_field_with_initial_default`. The date literal
conversion happens in `ExpressionEvaluatorVisitor`, not in the new code, so the
test adds no path. Consider dropping it. The double test is worth keeping,
since only it catches `is_nan`. This is a spot check on seven mutations, not a
coverage proof.
##########
crates/iceberg/src/expr/visitors/expression_evaluator.rs:
##########
@@ -54,14 +54,14 @@ impl ExpressionEvaluator {
/// evaluation logic to different parts of a data structure,
/// specifically for data file partitions.
#[derive(Debug)]
-struct ExpressionEvaluatorVisitor<'a> {
+pub(crate) struct ExpressionEvaluatorVisitor<'a> {
/// Reference to a [`DataFile`]'s partition [`Struct`].
partition: &'a Struct,
Review Comment:
Nit: the struct doc and this field name still say "partition", but the
residual now passes a row of top-level field values. Reusing the visitor is the
right call. Updating the doc and renaming the field to `row` (or `values`)
would keep the name accurate.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
