brgr-s commented on code in PR #2961:
URL: https://github.com/apache/iceberg-rust/pull/2961#discussion_r3727726866
##########
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:
You are absolutely correct about the `O(data_rows + delete_keys`) claim, it
is incorrect.
The correct bound ist `O(data_rows + applicable_delete_refs)`.
Or, if we set m = applicable delete files sharing a layout, K = keys per
delete file, D = rows in the data file
```
Old = D*m*K
New = D + m*K
```
My intuition on your suggestion `Vec<Arc<EqDeleteSet>>` is (with c_prope
cost for lookup, c_clone cost for `Clone`):
```
merge: m*K*c_clone (build, once) + D*c_probe (one lookup per row)
N-set: 0 (Arc bumps) + D*m*c_probe (m lookups per row)
```
Merging wins when `m*K*c_copy < D*(m-1)*c_probe`, or, `D/K >
(m/(m-1))*(c_clone/c_probe).
When m grows, `(m/(m-1)) -> 1`, so the threshold gets smaller (assuming
relativ constant `c_clone` vs `c_probe`, but I think thats Ok), and merging
gets "relativly better" in the "fanout case" (for m=1, we never merge). N-set
probing wins when `D/K ~ small`, but I'd argue that delete file application is
not the dominant cost for reading in that case.
Regarding Java, I took a look at
[`DeleteFilter.java:191-211`](https://github.com/apache/iceberg/blob/main/data/src/main/java/org/apache/iceberg/data/DeleteFilter.java#L191),
and I think you have `applyEqDeletes` and `isInDeleteSets` in mind.
`isInDeleteSets` lives in a `DeleteFilter` instance, which is constructed per
data file. I might be looking at the wrong thing, however?
--
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]
