From 7b0fed38e50d6f83e8927deb858c3523574479af Mon Sep 17 00:00:00 2001 From: Anoop Johnson Date: Sun, 16 Aug 2026 09:51:36 -0700 Subject: [PATCH 1/3] perf(arrow): buffer positional-delete runs to drop per-row key allocation parse_positional_deletes_record_batch_stream called result.entry(file_path.to_string()) for every deleted row, allocating and hashing a fresh path string per row even though a positional delete file repeats the same data-file path across all of its rows (the records are sorted by (file_path, pos)). Within each Arrow batch, buffer the contiguous run of positions for one path and merge it into the map in a single entry lookup, so the key is allocated and hashed once per run instead of once per row. Grouping is per batch, not across the whole stream (the run buffer is scoped to the batch), so a path spanning batches is flushed once per batch it appears in. Behavior is unchanged: positions are still inserted one-by-one (no ordering precondition), and a path that recurs merges into its existing delete vector, so results are identical regardless of ordering. Also drops an unused schema binding. Microbenchmark filling HashMap from 500k rows in 8192-row batches: one data file 32.9 -> 8.5 ns/row (3.9x); 50 data files 36.8 -> 10.0 ns/row (3.7x). --- .../src/arrow/caching_delete_file_loader.rs | 79 +++++++++++++++++-- 1 file changed, 74 insertions(+), 5 deletions(-) diff --git a/crates/iceberg/src/arrow/caching_delete_file_loader.rs b/crates/iceberg/src/arrow/caching_delete_file_loader.rs index eb5e1ac4b0..6d142df628 100644 --- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs +++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs @@ -344,7 +344,6 @@ impl CachingDeleteFileLoader { while let Some(batch) = stream.next().await { let batch = batch?; - let schema = batch.schema(); let columns = batch.columns(); let Some(file_paths) = columns[0].as_any().downcast_ref::() else { @@ -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 = Vec::new(); + for (file_path, pos) in file_paths.iter().zip(positions.iter()) { let (Some(file_path), Some(pos)) = (file_path, pos) else { return Err(Error::new( @@ -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, + 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); + } + } + async fn parse_equality_deletes_record_batch_stream( mut stream: ArrowRecordBatchStream, equality_ids: HashSet, @@ -927,6 +963,39 @@ mod tests { assert!(err.message().contains("negative position")); } + #[tokio::test] + async fn test_parse_positional_deletes_groups_and_merges_paths() { + let schema = crate::arrow::delete_filter::tests::create_pos_del_schema(); + + // "a" appears in two non-contiguous runs (must merge), "b" spans the + // two batches, and each file accumulates every one of its positions. + let batch1 = RecordBatch::try_new(schema.clone(), vec![ + Arc::new(StringArray::from_iter_values(vec!["a", "a", "b", "a"])), + Arc::new(Int64Array::from_iter_values(vec![1i64, 3, 2, 5])), + ]) + .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(); + + let sorted = |dv: &DeleteVector| { + let mut v: Vec = dv.iter().collect(); + v.sort_unstable(); + v + }; + assert_eq!(result.len(), 3); + assert_eq!(sorted(&result["a"]), vec![1, 3, 5]); + assert_eq!(sorted(&result["b"]), vec![2, 4]); + assert_eq!(sorted(&result["c"]), vec![0]); + } + /// Verifies that evolve_schema on partial-schema equality deletes works correctly /// when only equality_ids columns are evolved, not all table columns. /// From 721b42eec862fa9a5f984cf1b3e113b146cc56e6 Mon Sep 17 00:00:00 2001 From: Anoop Johnson Date: Sun, 16 Aug 2026 13:48:47 -0700 Subject: [PATCH 2/3] test(arrow): split positional-delete parse test into sorted and unsorted cases Replace the single grouping test with two: one spec-compliant case (rows sorted by (file_path, pos), covering multi-position runs, several files in a batch, and a run spanning the batch boundary), and one deliberately spec-noncompliant unsorted case that proves the reader does not depend on the spec sort order. A path split into non-contiguous runs still merges into one delete vector rather than dropping positions. The non-compliance is called out in the test name and doc comment so the input is not mistaken for a realistic one. --- .../src/arrow/caching_delete_file_loader.rs | 53 ++++++++++++++----- 1 file changed, 40 insertions(+), 13 deletions(-) diff --git a/crates/iceberg/src/arrow/caching_delete_file_loader.rs b/crates/iceberg/src/arrow/caching_delete_file_loader.rs index 6d142df628..9f39fa5b9f 100644 --- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs +++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs @@ -963,15 +963,22 @@ mod tests { assert!(err.message().contains("negative position")); } + fn sorted_positions(dv: &DeleteVector) -> Vec { + let mut positions: Vec = 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_groups_and_merges_paths() { + async fn test_parse_positional_deletes_merges_sorted_runs() { let schema = crate::arrow::delete_filter::tests::create_pos_del_schema(); - // "a" appears in two non-contiguous runs (must merge), "b" spans the - // two batches, and each file accumulates every one of its positions. let batch1 = RecordBatch::try_new(schema.clone(), vec![ - Arc::new(StringArray::from_iter_values(vec!["a", "a", "b", "a"])), - Arc::new(Int64Array::from_iter_values(vec![1i64, 3, 2, 5])), + 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![ @@ -985,15 +992,35 @@ mod tests { .await .unwrap(); - let sorted = |dv: &DeleteVector| { - let mut v: Vec = dv.iter().collect(); - v.sort_unstable(); - v - }; assert_eq!(result.len(), 3); - assert_eq!(sorted(&result["a"]), vec![1, 3, 5]); - assert_eq!(sorted(&result["b"]), vec![2, 4]); - assert_eq!(sorted(&result["c"]), vec![0]); + assert_eq!(sorted_positions(&result["a"]), vec![1, 3, 5]); + assert_eq!(sorted_positions(&result["b"]), vec![2, 4]); + assert_eq!(sorted_positions(&result["c"]), vec![0]); + } + + /// Deliberately unsorted input. The spec requires position delete rows to be + /// sorted by (file_path, pos), but the reader must not depend on it: run + /// buffering only groups *contiguous* rows, so a path split into + /// non-contiguous runs (here "a" before and after "b") must still merge into + /// a single delete vector rather than silently dropping positions. + #[tokio::test] + async fn test_parse_positional_deletes_merges_spec_noncompliant_unsorted_runs() { + let schema = crate::arrow::delete_filter::tests::create_pos_del_schema(); + + let batch = RecordBatch::try_new(schema, vec![ + Arc::new(StringArray::from_iter_values(vec!["a", "b", "a"])), + Arc::new(Int64Array::from_iter_values(vec![3i64, 2, 1])), + ]) + .unwrap(); + let stream = futures::stream::iter(vec![Ok(batch)]).boxed(); + + let result = CachingDeleteFileLoader::parse_positional_deletes_record_batch_stream(stream) + .await + .unwrap(); + + assert_eq!(result.len(), 2); + assert_eq!(sorted_positions(&result["a"]), vec![1, 3]); + assert_eq!(sorted_positions(&result["b"]), vec![2]); } /// Verifies that evolve_schema on partial-schema equality deletes works correctly From 68e4692eefb2d60d562fba41ed7b343bb4404432 Mon Sep 17 00:00:00 2001 From: Anoop Johnson Date: Mon, 17 Aug 2026 18:06:56 -0700 Subject: [PATCH 3/3] PR feedback --- .../src/arrow/caching_delete_file_loader.rs | 28 ++++++++++++------- crates/iceberg/src/delete_vector.rs | 1 - 2 files changed, 18 insertions(+), 11 deletions(-) diff --git a/crates/iceberg/src/arrow/caching_delete_file_loader.rs b/crates/iceberg/src/arrow/caching_delete_file_loader.rs index 9f39fa5b9f..87e4acf207 100644 --- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs +++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs @@ -341,6 +341,7 @@ impl CachingDeleteFileLoader { mut stream: ArrowRecordBatchStream, ) -> Result> { let mut result: HashMap = HashMap::default(); + let mut run_positions: Vec = Vec::new(); while let Some(batch) = stream.next().await { let batch = batch?; @@ -367,7 +368,6 @@ impl CachingDeleteFileLoader { // 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 = Vec::new(); for (file_path, pos) in file_paths.iter().zip(positions.iter()) { let (Some(file_path), Some(pos)) = (file_path, pos) else { @@ -384,8 +384,8 @@ impl CachingDeleteFileLoader { } if run_path != Some(file_path) { - if let Some(run_path) = run_path { - Self::merge_delete_positions(&mut result, run_path, &run_positions); + if let Some(prev_path) = run_path { + Self::merge_delete_positions(&mut result, prev_path, &run_positions); run_positions.clear(); } @@ -395,8 +395,9 @@ impl CachingDeleteFileLoader { run_positions.push(pos as u64); } - if let Some(run_path) = run_path { - Self::merge_delete_positions(&mut result, run_path, &run_positions); + if let Some(prev_path) = run_path { + Self::merge_delete_positions(&mut result, prev_path, &run_positions); + run_positions.clear(); } } @@ -410,13 +411,20 @@ impl CachingDeleteFileLoader { file_path: &str, positions: &[u64], ) { - if positions.is_empty() { - return; - } + // 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(); - for &pos in positions { - delete_vector.insert(pos); + // 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() { + for &pos in positions { + delete_vector.insert(pos); + } } } diff --git a/crates/iceberg/src/delete_vector.rs b/crates/iceberg/src/delete_vector.rs index df8a10193c..c3c764c02c 100644 --- a/crates/iceberg/src/delete_vector.rs +++ b/crates/iceberg/src/delete_vector.rs @@ -52,7 +52,6 @@ impl DeleteVector { /// # Errors /// /// Returns an error if the precondition is not met. - #[allow(dead_code)] pub fn insert_positions(&mut self, positions: &[u64]) -> Result { if let Err(err) = self.inner.append(positions.iter().copied()) { return Err(Error::new(