From 2511f21380bba11468d156f3c03c416ddbc617ba Mon Sep 17 00:00:00 2001 From: JeonDaehong Date: Sat, 17 Oct 2026 19:00:40 +0900 Subject: [PATCH] [core] Build deletion-vector row ranges from roaring runs Walk the deletion vector run by run instead of row by row when turning it into non-deleted row ranges. Short runs still use next, since next_range is slower on scattered deletes. Closes #1146 --- crates/paimon/src/deletion_vector/core.rs | 104 ++++++++++++++++++ .../paimon/src/table/data_evolution_reader.rs | 12 +- crates/paimon/src/table/data_file_reader.rs | 36 +++++- 3 files changed, 140 insertions(+), 12 deletions(-) diff --git a/crates/paimon/src/deletion_vector/core.rs b/crates/paimon/src/deletion_vector/core.rs index 99f700f2c..e274e3017 100644 --- a/crates/paimon/src/deletion_vector/core.rs +++ b/crates/paimon/src/deletion_vector/core.rs @@ -36,6 +36,9 @@ const MAGIC_NUMBER: u32 = 1581511376; /// unlike the v1 magic, so the two are told apart by byte order as well as value. const MAGIC_NUMBER_64: u32 = 1681511377; const MAGIC_NUMBER_SIZE_BYTES: u64 = 4; +/// Adjacent deleted positions walked one by one before switching to +/// `next_range` in [`DeletionVector::deleted_ranges`]. +const RUN_PROBE_LEN: u32 = 4; /// Size of the `bitmapLength` prefix and of the trailing CRC. Mirrors Java /// `Bitmap64DeletionVector.LENGTH_SIZE_BYTES` / `CRC_SIZE_BYTES`; a v2 entry /// counts both in `DeletionFile.length()` while a v1 entry counts neither. @@ -93,6 +96,50 @@ impl DeletionVector { self.bitmap.is_empty() } + /// Deleted positions as ascending inclusive `(start, end)` runs. + /// + /// Short runs are walked with `next`; once a run is longer than + /// [`RUN_PROBE_LEN`], the rest of it is taken with `next_range`, one roaring + /// container at a time. `next_range` alone is slower than `next` on + /// scattered deletes. + pub(crate) fn deleted_ranges(&self) -> impl Iterator + '_ { + let mut iter = self.bitmap.iter(); + let mut pending: Option<(u32, u32)> = None; + std::iter::from_fn(move || { + let (start, mut end) = match pending.take() { + Some(run) => run, + None => { + let pos = iter.next()?; + (pos, pos) + } + }; + let mut walked = end - start; + while walked < RUN_PROBE_LEN { + match iter.next() { + Some(pos) if end.checked_add(1) == Some(pos) => { + end = pos; + walked += 1; + } + Some(pos) => { + pending = Some((pos, pos)); + return Some((start, end)); + } + None => return Some((start, end)), + } + } + // A run split at a container boundary comes back in pieces; join them. + while let Some(run) = iter.next_range() { + if end.checked_add(1) == Some(*run.start()) { + end = *run.end(); + } else { + pending = Some((*run.start(), *run.end())); + break; + } + } + Some((start, end)) + }) + } + /// Serialize using Java `BitmapDeletionVector` format: /// `i32 bitmapLength | i32 magic | roaring bitmap bytes | i32 crc`. pub(crate) fn serialize_to_bytes(&self) -> crate::Result> { @@ -588,4 +635,61 @@ mod tests { // Positions above u32::MAX cannot exist in a roaring32 bitmap -> not deleted. assert!(!dv.is_deleted(u64::from(u32::MAX) + 1)); } + + /// Runs of `bitmap.iter()`, the obvious way. + fn model_ranges(bitmap: &RoaringBitmap) -> Vec<(u32, u32)> { + let mut ranges: Vec<(u32, u32)> = Vec::new(); + for pos in bitmap { + match ranges.last_mut() { + Some((_, end)) if end.checked_add(1) == Some(pos) => *end = pos, + _ => ranges.push((pos, pos)), + } + } + ranges + } + + fn assert_ranges(bitmap: RoaringBitmap) { + let expected = model_ranges(&bitmap); + let dv = DeletionVector::from_bitmap(bitmap); + assert_eq!(dv.deleted_ranges().collect::>(), expected); + } + + #[test] + fn test_deleted_ranges_edges() { + assert_ranges(RoaringBitmap::new()); + assert_ranges(RoaringBitmap::from([7])); + // Runs right at the probe length, and one past it. + assert_ranges((10..14).chain(20..25).chain(30..31).collect()); + // One run across several roaring containers, then a short one. + assert_ranges((65_530..200_000).chain([200_002]).collect()); + // A long run followed by an adjacent-container run that does not touch it. + assert_ranges((0..65_535).chain(65_537..70_000).collect()); + // Runs ending at u32::MAX. + assert_ranges([u32::MAX].into_iter().collect()); + assert_ranges((u32::MAX - 70_000..=u32::MAX).collect()); + } + + #[test] + fn test_deleted_ranges_match_model_on_random_bitmaps() { + let mut state = 0x2545_f491_4f6c_dd1du64; + let mut next = move || { + state ^= state << 13; + state ^= state >> 7; + state ^= state << 17; + state + }; + for round in 0..128 { + let max_gap = [2u64, 30, 3_000, 100_000][round % 4]; + let max_len = [3u64, 50, 3_000, 70_000][(round / 4) % 4]; + let mut bitmap = RoaringBitmap::new(); + let mut pos = (next() % 200_000) as u32; + for _ in 0..40 { + let len = 1 + (next() % max_len) as u32; + let end = pos.saturating_add(len); + bitmap.insert_range(pos..end); + pos = end.saturating_add(1 + (next() % max_gap) as u32); + } + assert_ranges(bitmap); + } + } } diff --git a/crates/paimon/src/table/data_evolution_reader.rs b/crates/paimon/src/table/data_evolution_reader.rs index 6fac24c65..7750d8a86 100644 --- a/crates/paimon/src/table/data_evolution_reader.rs +++ b/crates/paimon/src/table/data_evolution_reader.rs @@ -2167,15 +2167,15 @@ fn prefix_row_ranges( fn non_deleted_local_ranges(row_count: i64, deletion_vector: &DeletionVector) -> Vec { let mut ranges = Vec::new(); let mut cursor = 0i64; - for deleted in deletion_vector.iter() { - let deleted = deleted as i64; - if deleted >= row_count { + for (start, end) in deletion_vector.deleted_ranges() { + let (start, end) = (i64::from(start), i64::from(end)); + if start >= row_count { break; } - if deleted > cursor { - ranges.push(RowRange::new(cursor, deleted - 1)); + if start > cursor { + ranges.push(RowRange::new(cursor, start - 1)); } - cursor = deleted + 1; + cursor = end + 1; } if cursor < row_count { ranges.push(RowRange::new(cursor, row_count - 1)); diff --git a/crates/paimon/src/table/data_file_reader.rs b/crates/paimon/src/table/data_file_reader.rs index b97f051aa..be0fe1494 100644 --- a/crates/paimon/src/table/data_file_reader.rs +++ b/crates/paimon/src/table/data_file_reader.rs @@ -1420,15 +1420,15 @@ fn ranges_cover_all_rows(ranges: &[RowRange], row_count: i64) -> bool { fn dv_to_non_deleted_ranges(dv: &DeletionVector, row_count: i64) -> Vec { let mut result = Vec::new(); let mut cursor: i64 = 0; - for deleted in dv.iter() { - let del = deleted as i64; - if del >= row_count { + for (start, end) in dv.deleted_ranges() { + let (start, end) = (i64::from(start), i64::from(end)); + if start >= row_count { break; } - if del > cursor { - result.push(RowRange::new(cursor, del - 1)); + if start > cursor { + result.push(RowRange::new(cursor, start - 1)); } - cursor = del + 1; + cursor = end + 1; } if cursor < row_count { result.push(RowRange::new(cursor, row_count - 1)); @@ -2356,6 +2356,30 @@ mod tests { ); } + #[test] + fn dv_to_non_deleted_ranges_handles_long_runs() { + let dv = DeletionVector::from_bitmap; + // A run crossing roaring containers. + assert_eq!( + dv_to_non_deleted_ranges(&dv((70_000..140_000).collect()), 200_000), + vec![RowRange::new(0, 69_999), RowRange::new(140_000, 199_999)] + ); + // A run that goes past the end of the file. + assert_eq!( + dv_to_non_deleted_ranges(&dv((3..100_000).collect()), 50_000), + vec![RowRange::new(0, 2)] + ); + // Positions at or past row_count are ignored. + assert_eq!( + dv_to_non_deleted_ranges(&dv((0..10).chain(60_000..60_010).collect()), 50_000), + vec![RowRange::new(10, 49_999)] + ); + assert_eq!( + dv_to_non_deleted_ranges(&dv((0..50_000).collect()), 50_000), + vec![] + ); + } + #[test] fn file_index_selection_coalesces_positions_and_rejects_out_of_range_values() { let selection = [0_u32, 1, 3].into_iter().collect();