diff --git a/crates/iceberg/src/arrow/caching_delete_file_loader.rs b/crates/iceberg/src/arrow/caching_delete_file_loader.rs index eb5e1ac4b0..87e4acf207 100644 --- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs +++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs @@ -341,10 +341,10 @@ 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?; - let schema = batch.schema(); let columns = batch.columns(); let Some(file_paths) = columns[0].as_any().downcast_ref::() else { @@ -360,6 +360,15 @@ 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; + 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,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, + 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() { + for &pos in positions { + delete_vector.insert(pos); + } + } + } + async fn parse_equality_deletes_record_batch_stream( mut stream: ArrowRecordBatchStream, equality_ids: HashSet, @@ -927,6 +971,66 @@ 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_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]); + 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 /// when only equality_ids columns are evolved, not all table columns. /// 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(