brgr-s commented on code in PR #2961:
URL: https://github.com/apache/iceberg-rust/pull/2961#discussion_r3726801770
##########
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:
I followed your trace and you are correct, this is a spec divergence.
Digging deeper: the old code was also spec-divergent, in two different ways:
- required missing columns lead to all rows beeing dropped for that data file
- optional missing columns on the other hand "do nothing"
This PR keeps all rows in both cases, but but scoped to that specific layout
group. All rows survive, but if other delete files apply (eq deletes with
different layouts, pos deletes), they "can" delete rows... I'd argue that this
a slight improvement (resurecting some rows vs. dropping all), but the fact you
pointed out remains: it is spec divergent, it needs to be made obvious, and it
needs to be fixed (but not on this PR). I think this would be worth filing an
issue, even if this PR does not get accepted.
--
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]
