mbutrovich commented on code in PR #2961:
URL: https://github.com/apache/iceberg-rust/pull/2961#discussion_r3722140834
##########
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:
When an equality-delete key column is absent from a data file's parquet
schema (schema evolution: the column was added after this file was written),
the probe treats it as null for every row: `columns.push(vec![None; num_rows]);
continue;`. Per spec.md:898, "the column value is read for older data files
using normal projection rules" for exactly this case, and iceberg-rust already
has a mechanism for "normal projection rules" for a missing column:
`RecordBatchTransformer` resolves `initial_default` (see
`record_batch_transformer.rs`, "Rule #3"), used for the data file's own output
columns at pipeline.rs:494. The eq-delete probe runs earlier, inside the
`RowFilter`, on raw decoded batches, before `RecordBatchTransformer` ever sees
them, so it never consults `initial_default` and always substitutes null
instead. Java's `DeleteFilter` avoids this by construction: it compares against
`record`, which is built by the regular row reader against `requiredSchema`
(`DeleteFilter.java:99-101
, 202-209`), i.e. defaults are already resolved before the delete-set lookup
runs at all. Concretely: a required column added later with a non-null
`initial_default`, later used as an equality-delete key, will never delete
matching rows in data files written before that column existed, because every
row in those files probes as null instead of the default value. Low likelihood
in practice, but it's a silent, spec-observable divergence, not a crash, so
worth a test either way.
##########
crates/iceberg/src/arrow/delete_filter.rs:
##########
@@ -163,68 +162,74 @@ impl DeleteFilter {
}
}
- /// Retrieve the equality delete predicate for a given eq delete file path
- pub(crate) async fn get_equality_delete_predicate_for_delete_file_path(
+ /// Retrieve the equality delete set for a given eq delete file path
+ pub(crate) async fn get_equality_delete_set_for_delete_file_path(
&self,
file_path: &str,
- ) -> Option<Predicate> {
+ ) -> Option<Arc<EqDeleteSet>> {
let notifier = {
match self.state.read().unwrap().equality_deletes.get(file_path) {
None => return None,
Some(EqDelState::Loading(notifier)) => notifier.clone(),
- Some(EqDelState::Loaded(predicate)) => {
- return Some(predicate.clone());
+ Some(EqDelState::Loaded(set)) => {
+ return Some(set.clone());
}
}
};
notifier.notified().await;
match self.state.read().unwrap().equality_deletes.get(file_path) {
- Some(EqDelState::Loaded(predicate)) => Some(predicate.clone()),
+ Some(EqDelState::Loaded(set)) => Some(set.clone()),
_ => unreachable!("Cannot be any other state than loaded"),
}
}
- /// Builds eq delete predicate for the provided task.
- pub(crate) async fn build_equality_delete_predicate(
+ /// Builds the equality-delete sets applicable to the given task, one per
distinct
+ /// equality-column layout.
+ pub(crate) async fn build_equality_delete_sets(
&self,
file_scan_task: &FileScanTask,
- ) -> Result<Option<BoundPredicate>> {
- // * Filter the task's deletes into just the Equality deletes
- // * Retrieve the unbound predicate for each from
self.state.equality_deletes
- // * Logical-AND them all together to get a single combined `Predicate`
- // * Bind the predicate to the task's schema to get a `BoundPredicate`
-
- let mut combined_predicate = AlwaysTrue;
+ ) -> Result<Vec<Arc<EqDeleteSet>>> {
+ let mut groups: HashMap<Vec<i32>, Vec<Arc<EqDeleteSet>>> =
HashMap::new();
for delete in &file_scan_task.deletes {
if !is_equality_delete(delete) {
continue;
}
- let Some(predicate) = self
-
.get_equality_delete_predicate_for_delete_file_path(&delete.file_path)
+ let Some(set) = self
+
.get_equality_delete_set_for_delete_file_path(&delete.file_path)
.await
else {
return Err(Error::new(
ErrorKind::Unexpected,
format!(
- "Missing predicate for equality delete file '{}'",
+ "Missing equality delete set for delete file '{}'",
delete.file_path
),
));
};
- combined_predicate = combined_predicate.and(predicate);
+ let layout = set.fields.iter().map(|(_, id, _)| *id).collect();
Review Comment:
Grouping key is `set.fields.iter().map(|(_, id, _)| *id).collect()`, field
id only, no type. In isolation this looks like it could let two type-mismatched
equality-delete sets reach `union()` and hard-fail the read. It can't in
practice: `CachingDeleteFileLoader` evolves every equality-delete file's batch
stream against `task.schema` before parsing it into an `EqDeleteSet`
(`caching_delete_file_loader.rs:333-343`, via
`BasicDeleteFileLoader::evolve_schema` / `RecordBatchTransformer`, which casts
present-but-differently-typed columns, `record_batch_transformer.rs:810`). So
every set sharing the same scan's `task.schema` already carries the same
recorded type for the same field id by construction, and grouping by id only
mirrors Java's own grouping (`Sets.newHashSet(delete.equalityFieldIds())`,
`DeleteFilter.java:194`), which also doesn't carry type in the key, for the
same reason: Java projects every delete file into `requiredSchema` up front
instead of comparing delete-file-record
ed types against each other. Given that, the `union()` type check and its
comment are effectively dead code / defense-in-depth for an invariant that
already holds structurally, not the live bug I'd have guessed from reading this
function alone. Worth a comment saying why the invariant holds (ties to
`evolve_schema` in the other file) instead of "should a change break this,"
since the current comment reads as a guess rather than a traced guarantee. Not
worth blocking on.
##########
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;
+ };
+ let array = batch.column(pos);
+ let source_type = arrow_type_to_type(array.data_type())
+ .map_err(|e|
ArrowError::ComputeError(e.to_string()))?;
+ let source_primitive = source_type
+ .as_primitive_type()
+ .ok_or_else(|| {
+ ArrowError::ComputeError(
+ "equality delete key column is not a
primitive type"
+ .to_string(),
+ )
+ })?
+ .clone();
+ let needs_promotion = source_type != *target_type;
+ let literals = arrow_primitive_to_literal(array,
&source_type)
+ .map_err(|e|
ArrowError::ComputeError(e.to_string()))?;
+
+ let mut column = Vec::with_capacity(num_rows);
+ for literal in literals {
+ let datum = match literal {
+ Some(literal) => {
+ let primitive =
+
literal.as_primitive_literal().ok_or_else(|| {
+ ArrowError::ComputeError(
+ "failed to convert to
primitive literal"
+ .to_string(),
+ )
+ })?;
+ let datum =
Datum::new(source_primitive.clone(), primitive);
+ let datum = if needs_promotion {
+ datum
+ .to(target_type)
+ .map_err(|e|
ArrowError::ComputeError(e.to_string()))?
+ } else {
+ datum
+ };
+ Some(datum)
+ }
+ None => None,
+ };
+ column.push(datum);
+ }
+ columns.push(column);
+ }
+
+ // One hash lookup per row.
+ let mut keep = Vec::with_capacity(num_rows);
+ let mut probe = EqDeleteKey(vec![None; num_cols]);
+ for row in 0..num_rows {
+ for (i, column) in columns.iter_mut().enumerate() {
+ // we can `take` because each cell is probed once.
+ probe.0[i] = std::mem::take(&mut column[row]);
+ }
+ keep.push(!set.keys.contains(&probe));
+ }
+ Ok(BooleanArray::from(keep))
+ };
Review Comment:
Per batch, per key column: `arrow_primitive_to_literal` (row_filter.rs:144)
builds an owned `Vec<Option<Literal>>` for the whole column, then
row_filter.rs:148-171 walks it again to build a `Vec<Option<Datum>>`. Two full
passes and, for string/binary columns, two rounds of per-cell heap ownership,
on the exact hot path this PR is trying to make cheap. Separately,
row_filter.rs:158 does `source_primitive.clone()` inside the per-row loop even
though `source_primitive` is fixed per column (computed once at
row_filter.rs:134-142); cheap since `PrimitiveType` doesn't allocate, but still
pointless per-row work. `caching_delete_file_loader.rs:541-556` has the same
double-pass/box-per-cell pattern already, unmodified by this PR, but it runs
once per delete-file load; this PR adds a second copy of it on the probe side,
which runs once per batch per applicable data file, i.e. far more often.
`arrow::row::RowConverter` (crate `arrow-row`, already in Cargo.lock at 59.1.0,
not yet a direct dep
of `crates/iceberg`) builds comparable/hashable rows straight from Arrow
arrays without this boxing, and is what DataFusion uses for multi-column
join/group keys. For the common single-column case, DataFusion's `InListExpr`
static filters
(`datafusion/physical-expr/src/expressions/in_list/{primitive_filter,static_filter}.rs`)
use a native-typed `hashbrown`/`ahash` set (`datafusion_common::HashSet`, see
`datafusion/common/src/lib.rs:119`) instead of boxed values in
`std::collections::HashSet` with the default SipHash, which is what
`EqDeleteSet::keys` uses here. Worth noting Java's generic `DeleteFilter`
doesn't push equality-delete filtering into the file decoder at all, it applies
a row-by-row `Predicate<T>` after full materialization; pushing it into
parquet-rs's `RowFilter` the way this PR does is a real advantage over that, so
this finding is about the representation used inside the predicate, not the
decision to use a `RowFilter` predicate.
##########
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)?;
Review Comment:
`resolve_field_id_map` recomputes the parquet-schema-to-field-id map
whenever `eq_delete_sets` is non-empty, even though pipeline.rs already
computed an equivalent map for the scan predicate a few lines earlier via
`ArrowReader::build_field_id_set_and_map` (projection.rs:49-62, which itself
calls `resolve_field_id_map`). Same schema both times; minor duplicated
per-file work, not per-row, so low severity, but easy to thread through instead.
##########
crates/iceberg/src/arrow/reader/pipeline.rs:
##########
@@ -400,13 +395,12 @@ impl FileScanTaskReader {
use_position_fallback,
)?;
- let row_filter = ArrowReader::get_row_filter(
+ row_filter_predicates.push(ArrowReader::build_scan_predicate(
Review Comment:
Also line 440.
Scan predicate is always pushed before the eq-delete predicates,
unconditionally. parquet-rs applies `RowFilter` predicates in list order,
threading a shrinking `RowSelection` between them (arrow-rs
`parquet/src/arrow/arrow_reader/filter.rs:139-149`), so a cheap/selective
predicate run first prunes work for everything after it. DataFusion's
`ParquetSource` reorders predicates by estimated column byte-size before
building the `RowFilter` for exactly this reason
(`datafusion/datasource-parquet/src/row_filter.rs:33-52`). A hash-probe over
one or two narrow key columns is often cheaper and more selective than an
arbitrary scan predicate; this PR never reorders to take advantage of that.
##########
crates/iceberg/src/arrow/delete_filter.rs:
##########
@@ -163,68 +162,74 @@ impl DeleteFilter {
}
}
- /// Retrieve the equality delete predicate for a given eq delete file path
- pub(crate) async fn get_equality_delete_predicate_for_delete_file_path(
+ /// Retrieve the equality delete set for a given eq delete file path
+ pub(crate) async fn get_equality_delete_set_for_delete_file_path(
&self,
file_path: &str,
- ) -> Option<Predicate> {
+ ) -> Option<Arc<EqDeleteSet>> {
let notifier = {
match self.state.read().unwrap().equality_deletes.get(file_path) {
None => return None,
Some(EqDelState::Loading(notifier)) => notifier.clone(),
- Some(EqDelState::Loaded(predicate)) => {
- return Some(predicate.clone());
+ Some(EqDelState::Loaded(set)) => {
+ return Some(set.clone());
}
}
};
notifier.notified().await;
match self.state.read().unwrap().equality_deletes.get(file_path) {
- Some(EqDelState::Loaded(predicate)) => Some(predicate.clone()),
+ Some(EqDelState::Loaded(set)) => Some(set.clone()),
_ => unreachable!("Cannot be any other state than loaded"),
}
}
- /// Builds eq delete predicate for the provided task.
- pub(crate) async fn build_equality_delete_predicate(
+ /// Builds the equality-delete sets applicable to the given task, one per
distinct
+ /// equality-column layout.
+ pub(crate) async fn build_equality_delete_sets(
&self,
file_scan_task: &FileScanTask,
- ) -> Result<Option<BoundPredicate>> {
- // * Filter the task's deletes into just the Equality deletes
- // * Retrieve the unbound predicate for each from
self.state.equality_deletes
- // * Logical-AND them all together to get a single combined `Predicate`
- // * Bind the predicate to the task's schema to get a `BoundPredicate`
-
- let mut combined_predicate = AlwaysTrue;
+ ) -> Result<Vec<Arc<EqDeleteSet>>> {
+ let mut groups: HashMap<Vec<i32>, Vec<Arc<EqDeleteSet>>> =
HashMap::new();
for delete in &file_scan_task.deletes {
if !is_equality_delete(delete) {
continue;
}
- let Some(predicate) = self
-
.get_equality_delete_predicate_for_delete_file_path(&delete.file_path)
+ let Some(set) = self
+
.get_equality_delete_set_for_delete_file_path(&delete.file_path)
.await
else {
return Err(Error::new(
ErrorKind::Unexpected,
format!(
- "Missing predicate for equality delete file '{}'",
+ "Missing equality delete set for delete file '{}'",
delete.file_path
),
));
};
- combined_predicate = combined_predicate.and(predicate);
+ let layout = set.fields.iter().map(|(_, id, _)| *id).collect();
+ groups.entry(layout).or_default().push(set);
}
- if combined_predicate == AlwaysTrue {
- return Ok(None);
+ let mut result = Vec::with_capacity(groups.len());
+ for mut sets in groups.into_values() {
+ if sets.len() == 1 {
+ result.push(sets.pop().unwrap());
+ } else {
+ let mut combined = (*sets[0]).clone();
+ for other in &sets[1..] {
+ // `union` checks if `other`s' layout matches `combined`,
+ // which is currently always the case. This fails should a
change
+ // break this current invariant.
+ combined.union(other)?;
+ }
+ result.push(Arc::new(combined));
+ }
}
Review Comment:
`build_equality_delete_sets` runs once per data-file scan task, and whenever
a data file has more than one delete file sharing a layout, it clones the first
set's entire `HashSet<EqDeleteKey>` and unions in the rest, from scratch, every
time. The PR's own benchmark numbers (7,140 and 124,750 "delete refs") are
exactly the count of (data-file, delete-file) edges this repeats over, so the
real cost is closer to `O(data_rows + delete_refs)` than the claimed
`O(data_rows + delete_keys)` once delete files span many data files, which is
the compaction case motivating the PR. Java doesn't have this cost at all: it
builds one `StructLikeSet` per equality-id group once per `DeleteFilter`
(`DeleteFilter.java:191-211`), not per data file. Fix: pass
`Vec<Arc<EqDeleteSet>>` per layout to the row filter and probe against all of
them (row is deleted if it matches any), instead of merging into one owned set
per task. `Arc::clone` is a refcount bump; the `HashSet` clone is not.
--
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]
