mbutrovich commented on code in PR #2961:
URL: https://github.com/apache/iceberg-rust/pull/2961#discussion_r4298017220
##########
crates/iceberg/src/arrow/reader/row_filter.rs:
##########
@@ -62,8 +67,131 @@ impl ArrowReader {
// creates the projection mask for the Arrow predicates.
let projection_mask = ProjectionMask::leaves(parquet_schema,
column_indices.clone());
let predicate_func = visit(&mut converter, predicates)?;
- let arrow_predicate = ArrowPredicateFn::new(projection_mask,
predicate_func);
- Ok(RowFilter::new(vec![Box::new(arrow_predicate)]))
+ Ok(Box::new(ArrowPredicateFn::new(
+ projection_mask,
+ predicate_func,
+ )))
+ }
+
+ /// Builds one Arrow row-filter predicate per equality-delete set. The
predicate is based
+ /// on a hash-set lookup (see `EqDeleteSet`). It keeps a row unless its
key tuple is present
+ /// in that set. A row is deleted when it matches any set (the predicates
are AND-ed by the `RowFilter`).
+ pub(super) fn build_equality_delete_predicates(
+ sets: &[Arc<EqDeleteSet>],
+ parquet_schema: &SchemaDescriptor,
+ arrow_schema: &ArrowSchemaRef,
+ use_position_fallback: bool,
+ ) -> Result<Vec<Box<dyn ArrowPredicate>>> {
+ let field_id_map =
+ Self::resolve_field_id_map(parquet_schema, arrow_schema,
use_position_fallback)?;
+
+ let mut predicates: Vec<Box<dyn ArrowPredicate>> = Vec::new();
+ for set in sets {
+ if set.is_empty() {
+ continue;
+ }
+
+ // Parquet leaf index for each key column, in `fields` order; a
column dropped
+ // from this file by schema evolution has no entry.
+ let leaf_indices: Vec<Option<usize>> = set
+ .fields
+ .iter()
+ .map(|(_, id, _)| field_id_map.get(id).copied())
+ .collect();
+
+ let mut column_indices: Vec<usize> =
leaf_indices.iter().flatten().copied().collect();
+ column_indices.sort_unstable();
+ column_indices.dedup();
+ let projection_mask = ProjectionMask::leaves(parquet_schema,
column_indices.clone());
+
+ // Position of each key column within the projected batch
(parquet-rs presents the
+ // masked leaves in ascending leaf-index order).
+ let batch_positions: Vec<Option<usize>> = leaf_indices
+ .iter()
+ .map(|leaf| leaf.and_then(|idx|
column_indices.binary_search(&idx).ok()))
+ .collect();
+
+ let target_types: Vec<Type> = set.fields.iter().map(|(_, _, ty)|
ty.clone()).collect();
+ let num_cols = set.fields.len();
+ let set = set.clone();
+
+ let predicate_func =
+ move |batch: RecordBatch| -> std::result::Result<BooleanArray,
ArrowError> {
+ let num_rows = batch.num_rows();
+
+ // Change each key column into `Datum`s once, promoting to
the
+ // table type so the keys match the parsed delete keys
under schema
+ // evolution. A column absent from this file reads as
all-null.
+ let mut columns: Vec<Vec<Option<Datum>>> =
Vec::with_capacity(num_cols);
+ for (i, target_type) in target_types.iter().enumerate() {
+ let Some(pos) = batch_positions[i] else {
+ columns.push(vec![None; num_rows]);
+ continue;
+ };
Review Comment:
> it is spec divergent, it needs to be made obvious, and it needs to be
fixed (but maybe not on this PR).
#3302 fixes case B of #3369 on the predicate path in `main`. Before the row
filter is built, it evaluates each leaf on a column missing from the data file
against that column's identity partition value or `initial-default`. The keep
predicate built from equality deletes goes through the same step.
[`assert_absent_b_reads_as_7`](https://github.com/apache/iceberg-rust/blob/006d4b9a648b49e7ffbed59a2d2da9c8f425bd70/crates/iceberg/src/arrow/reader/row_filter.rs#L2075-L2126)
checks that an equality delete on `b = 7` removes every row of a file that
doesn't store `b`, when `b` reads as 7 through `initial-default` or an identity
partition value.
The set probe here doesn't go through that predicate, and [a missing key
column becomes
`None`](https://github.com/apache/iceberg-rust/blob/12c2bd3f3b719d9ce80bf0d0dc7f1c62e1620f81/crates/iceberg/src/arrow/reader/row_filter.rs#L136-L140).
If #3302 merges first, this PR would bring case B back as a regression against
`main`.
What do you think about resolving a missing key column with the [normal
projection
rules](https://github.com/apache/iceberg/blob/5e7169168db3d34e29354c6f59ec4d6e420b8d2d/format/spec.md?plain=1#L1466)
in the probe: the identity partition value, then `initial-default`, then null?
That value is the same for every row of the file, so the probe can use it as a
constant column. When every key column of a set is missing, the set reduces to
one lookup per file that keeps or drops all rows. `residual_for_missing_fields`
in #3302 resolves the same values from `constants_map` and `initial_default`,
so the two paths could share that lookup.
--
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]
