mbutrovich commented on code in PR #3015:
URL: https://github.com/apache/iceberg-rust/pull/3015#discussion_r3797507420
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -374,16 +383,43 @@ impl CachingDeleteFileLoader {
));
}
- result
- .entry(file_path.to_string())
- .or_default()
- .insert(pos as u64);
+ if run_path != Some(file_path) {
+ if let Some(run_path) = run_path {
+ Self::merge_delete_positions(&mut result, run_path,
&run_positions);
+ run_positions.clear();
+ }
+
+ run_path = Some(file_path);
+ }
+
+ run_positions.push(pos as u64);
+ }
+
+ if let Some(run_path) = run_path {
+ Self::merge_delete_positions(&mut result, run_path,
&run_positions);
}
}
Ok(result)
}
+ /// Marks every position in `positions` as deleted for `file_path`, merging
+ /// into any delete vector already recorded for that file.
+ fn merge_delete_positions(
+ result: &mut HashMap<String, DeleteVector>,
+ file_path: &str,
+ positions: &[u64],
+ ) {
+ if positions.is_empty() {
+ return;
+ }
+
+ let delete_vector = result.entry(file_path.to_string()).or_default();
+ for &pos in positions {
+ delete_vector.insert(pos);
+ }
Review Comment:
`DeleteVector::insert_positions` (delete_vector.rs:56) already bulk-appends
a sorted slice into the underlying `RoaringTreemap` in one call, and it's
currently unused (`#[allow(dead_code)]`). Since a run here is ascending in the
common spec-compliant case, calling `insert_positions` first and falling back
to this per-element loop only on `Err` would get the append-based speedup for
the case your benchmark targets, while keeping the non-compliant-run handling
your test at line 1007 covers.
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -374,16 +383,43 @@ impl CachingDeleteFileLoader {
));
}
- result
- .entry(file_path.to_string())
- .or_default()
- .insert(pos as u64);
+ if run_path != Some(file_path) {
+ if let Some(run_path) = run_path {
+ Self::merge_delete_positions(&mut result, run_path,
&run_positions);
+ run_positions.clear();
+ }
+
+ run_path = Some(file_path);
+ }
+
+ run_positions.push(pos as u64);
+ }
+
+ if let Some(run_path) = run_path {
+ Self::merge_delete_positions(&mut result, run_path,
&run_positions);
}
}
Ok(result)
}
+ /// Marks every position in `positions` as deleted for `file_path`, merging
+ /// into any delete vector already recorded for that file.
+ fn merge_delete_positions(
+ result: &mut HashMap<String, DeleteVector>,
+ file_path: &str,
+ positions: &[u64],
+ ) {
+ if positions.is_empty() {
Review Comment:
This guard can't be hit: both call sites only run after `run_path` was set
to `Some`, which always happens in the same iteration as a push onto
`run_positions`. Since this is a caller-guaranteed invariant rather than a real
precondition, `debug_assert!(!positions.is_empty())` would document it instead
of silently swallowing a case that can't occur.
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -360,6 +359,16 @@ impl CachingDeleteFileLoader {
));
};
+ // Within a batch, positional deletes are sorted by (file_path,
pos),
+ // so the rows for one data file form a contiguous run. Buffer each
+ // run and merge it with a single map lookup, allocating and
hashing
+ // the key once per run instead of once per row. Grouping is per
+ // batch, not across the whole stream: a run never spans batch
+ // boundaries, so a path that also appears in another batch merges
+ // into its existing delete vector (order does not affect the
result).
+ let mut run_path: Option<&str> = None;
+ let mut run_positions: Vec<u64> = Vec::new();
Review Comment:
`run_positions` is declared inside the `while` loop, so it drops and
rebuilds from empty every batch, discarding the capacity it grew to from the
previous batch's runs. `run_path` needs to stay batch-scoped since it borrows
the batch's `StringArray`, but `run_positions` is owned and could be hoisted
above the loop, with a `.clear()` added after the final flush at line 399 to
match.
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -374,16 +383,43 @@ impl CachingDeleteFileLoader {
));
}
- result
- .entry(file_path.to_string())
- .or_default()
- .insert(pos as u64);
+ if run_path != Some(file_path) {
+ if let Some(run_path) = run_path {
Review Comment:
This binding shadows the outer `run_path`, which is reassigned two lines
down at line 392, two different bindings sharing one name. Renaming this one to
`prev_path` would make the reassignment read as updating the loop state rather
than the just-matched value.
--
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]
