This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new d8072a13 [core] Build deletion-vector row ranges from roaring runs
(#1148)
d8072a13 is described below
commit d8072a1302d70ab3a0c654f4c01ef5d4c268caff
Author: JeonDaehong <[email protected]>
AuthorDate: Sun Oct 18 17:00:30 2026 +0900
[core] Build deletion-vector row ranges from roaring runs (#1148)
---
crates/paimon/src/deletion_vector/core.rs | 104 +++++++++++++++++++++++
crates/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 99f700f2..e274e301 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<Item = (u32, u32)> +
'_ {
+ 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<Vec<u8>> {
@@ -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::<Vec<_>>(), 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 15c0cdc9..d90f99c0 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -2149,15 +2149,15 @@ fn prefix_row_ranges(
fn non_deleted_local_ranges(row_count: i64, deletion_vector: &DeletionVector)
-> Vec<RowRange> {
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 742771b3..c3edf965 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -1458,15 +1458,15 @@ fn ranges_cover_all_rows(ranges: &[RowRange],
row_count: i64) -> bool {
fn dv_to_non_deleted_ranges(dv: &DeletionVector, row_count: i64) ->
Vec<RowRange> {
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));
@@ -2546,6 +2546,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();
