laskoviymishka commented on code in PR #3015:
URL: https://github.com/apache/iceberg-rust/pull/3015#discussion_r3842878464
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -374,16 +383,51 @@ impl CachingDeleteFileLoader {
));
}
- result
- .entry(file_path.to_string())
- .or_default()
- .insert(pos as u64);
+ if run_path != Some(file_path) {
+ if let Some(prev_path) = run_path {
+ Self::merge_delete_positions(&mut result, prev_path,
&run_positions);
+ run_positions.clear();
+ }
+
+ run_path = Some(file_path);
+ }
+
+ run_positions.push(pos as u64);
+ }
+
+ if let Some(prev_path) = run_path {
+ Self::merge_delete_positions(&mut result, prev_path,
&run_positions);
+ run_positions.clear();
}
}
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],
+ ) {
+ // Callers only flush a run after pushing at least one position onto
it.
+ debug_assert!(!positions.is_empty());
+
+ let delete_vector = result.entry(file_path.to_string()).or_default();
+ // A run is a strictly ascending slice in the spec-compliant case,
which
+ // `insert_positions` bulk-appends in one pass. Fall back to
per-position
+ // inserts when the append precondition doesn't hold (unsorted rows,
or a
+ // run that overlaps positions already recorded from an earlier batch).
+ // `insert` is idempotent, so re-inserting any prefix the failed append
+ // already added is harmless.
+ if delete_vector.insert_positions(positions).is_err() {
Review Comment:
This is the one thing I'd like in before it merges. The whole value of the
change is staying on the fast path, but `.is_err()` swallows the error with no
signal — so if a malformed file or a future bug routes a run through the
per-position fallback, the win just evaporates with nothing in the logs to
explain the slowdown.
Could we `tracing::warn!` (or `debug!`) before the fallback loop? If we want
a tighter contract we could match specifically on
`ErrorKind::PreconditionFailed` and propagate anything else, but even a log
line turns a silent regression into something diagnosable. Everything else here
is a nit — happy to approve once this one's in.
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -927,6 +971,66 @@ mod tests {
assert!(err.message().contains("negative position"));
}
+ fn sorted_positions(dv: &DeleteVector) -> Vec<u64> {
+ let mut positions: Vec<u64> = dv.iter().collect();
+ positions.sort_unstable();
+ positions
+ }
+
+ /// Spec-compliant input: rows sorted by (file_path, pos). Exercises the
+ /// common shape: multi-position runs, several files in one batch, and a
+ /// run for "b" that continues across the batch boundary.
+ #[tokio::test]
+ async fn test_parse_positional_deletes_merges_sorted_runs() {
+ let schema =
crate::arrow::delete_filter::tests::create_pos_del_schema();
+
+ let batch1 = RecordBatch::try_new(schema.clone(), vec![
+ Arc::new(StringArray::from_iter_values(vec!["a", "a", "a", "b"])),
+ Arc::new(Int64Array::from_iter_values(vec![1i64, 3, 5, 2])),
+ ])
+ .unwrap();
+ let batch2 = RecordBatch::try_new(schema, vec![
+ Arc::new(StringArray::from_iter_values(vec!["b", "c"])),
+ Arc::new(Int64Array::from_iter_values(vec![4i64, 0])),
+ ])
+ .unwrap();
+ let stream = futures::stream::iter(vec![Ok(batch1),
Ok(batch2)]).boxed();
+
+ let result =
CachingDeleteFileLoader::parse_positional_deletes_record_batch_stream(stream)
+ .await
+ .unwrap();
+
+ assert_eq!(result.len(), 3);
+ assert_eq!(sorted_positions(&result["a"]), vec![1, 3, 5]);
+ assert_eq!(sorted_positions(&result["b"]), vec![2, 4]);
Review Comment:
This cross-batch "b" always satisfies `append` (4 > 2), so it never actually
exercises the overlap fallback the helper comment calls out ("a run that
overlaps positions already recorded from an earlier batch"). Could we add a
case where batch2 overlaps batch1 — e.g. batch1 `("a", [5, 10])`, batch2 `("a",
[3, 7])`, asserting `{3, 5, 7, 10}`? A single-batch duplicate like `("a", [3,
3])` would cover the other fallback trigger.
That'd give us real coverage of the slow path instead of just the fast one —
which matters more here since the fallback is the part with no observability
today.
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -927,6 +971,66 @@ mod tests {
assert!(err.message().contains("negative position"));
}
+ fn sorted_positions(dv: &DeleteVector) -> Vec<u64> {
+ let mut positions: Vec<u64> = dv.iter().collect();
+ positions.sort_unstable();
+ positions
+ }
+
+ /// Spec-compliant input: rows sorted by (file_path, pos). Exercises the
+ /// common shape: multi-position runs, several files in one batch, and a
+ /// run for "b" that continues across the batch boundary.
Review Comment:
Small thing, but this inverts what actually happens — `run_path` resets per
batch, so the "b" run in batch1 is flushed at the end of batch1 and a fresh "b"
run starts in batch2. Runs don't continue across batch boundaries; what
continues is the accumulation into "b"'s delete vector. I'd reword to something
like `path "b" whose positions span two batches, so both runs merge into one
delete vector`.
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -374,16 +383,51 @@ impl CachingDeleteFileLoader {
));
}
- result
- .entry(file_path.to_string())
- .or_default()
- .insert(pos as u64);
+ if run_path != Some(file_path) {
+ if let Some(prev_path) = run_path {
+ Self::merge_delete_positions(&mut result, prev_path,
&run_positions);
+ run_positions.clear();
+ }
+
+ run_path = Some(file_path);
+ }
+
+ run_positions.push(pos as u64);
+ }
+
+ if let Some(prev_path) = run_path {
+ Self::merge_delete_positions(&mut result, prev_path,
&run_positions);
+ run_positions.clear();
}
}
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],
+ ) {
+ // Callers only flush a run after pushing at least one position onto
it.
+ debug_assert!(!positions.is_empty());
+
+ let delete_vector = result.entry(file_path.to_string()).or_default();
+ // A run is a strictly ascending slice in the spec-compliant case,
which
Review Comment:
"strictly ascending" is a bit stronger than the spec actually promises —
position deletes are required to be sorted by (file_path, pos), which is
non-decreasing, so ties are allowed. A spec-compliant file can legitimately
carry duplicates like `[3, 3, 7]` in one run, and `append` rejects that (3
isn't > 3), so we'd drop to the slow path for perfectly valid input.
Same reading applies to the batch comment up at line 363 — both present the
sort as a guaranteed invariant, which is the riskier part: a future maintainer
could conclude the fallback is dead code and remove it. I'd soften both to "in
spec-compliant files … sorted, and the fallback covers duplicates /
out-of-order rows."
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -341,10 +341,10 @@ impl CachingDeleteFileLoader {
mut stream: ArrowRecordBatchStream,
) -> Result<HashMap<String, DeleteVector>> {
let mut result: HashMap<String, DeleteVector> = HashMap::default();
+ let mut run_positions: Vec<u64> = Vec::new();
Review Comment:
Not blocking, but hoisting `run_positions` out here for reuse means
correctness now leans on the end-of-batch flush always running to clear it — a
future early `continue`/`return` in the loop would quietly carry stale
positions into the next batch's run. A
`debug_assert!(run_positions.is_empty())` at the top of the batch loop would
pin that invariant down cheaply. Just while we're here.
--
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]
