This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-3454-aa9986fe83d4bdb0d883968b9a105c6e0a9ef745 in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit d2679a029f50a25f00850731650d7fbab1b03ca8 Author: JeonDaehong <[email protected]> AuthorDate: Fri Oct 23 19:47:48 2026 +0000 perf(arrow): build the delete row selection from deleted ranges (#3454) * perf(arrow): build the delete row selection from deleted ranges build_deletes_row_selection walked the delete vector one position at a time, so a run of deleted rows cost one iteration per row. It now reads the deleted positions as ranges, which makes a run one step. Short runs are still walked with `next`, because roaring's `next_range` binary-searches the rest of an array container on every call and is slower than `next` for single positions. Row groups that are not selected are skipped with `skip_to` instead of walking their deletes. The new range iterator is crate-private; the public `DeleteVector::iter` and `DeleteVectorIterator::advance_to` are unchanged. * Address review: fix advance_to (#3400), randomized reader test - DeleteVectorIterator::advance_to now handles a target in a missing later bitmap, a target below the current bitmap, and a call before the first next (the cases in #3400), with the same three-way check as skip_to. Dropped the stale comments about roaring-rs#314 and the git dependency. - Randomized test for build_deletes_row_selection: random row group layouts and selections over mixed_delete_vectors, against a selection built one deleted position at a time. - full_range no longer overflows at u64::MAX. - One Buffered enum instead of pending/pending_range. - Model tests also skip backwards, mixed_delete_vectors has gaps between bitmaps, a model test for advance_to, the #3400 cases split per test, and doc updates for ranges, skip_to and RUN_PROBE_LEN. * Address review: deletes past the file end, runs across 2^16, ranges() docs * Use a seeded StdRng in the model tests and cover empty row groups --- .../iceberg/src/arrow/reader/positional_deletes.rs | 289 +++++++++++--- crates/iceberg/src/delete_vector.rs | 443 ++++++++++++++++++++- 2 files changed, 653 insertions(+), 79 deletions(-) diff --git a/crates/iceberg/src/arrow/reader/positional_deletes.rs b/crates/iceberg/src/arrow/reader/positional_deletes.rs index a261bd536..3cb3da98f 100644 --- a/crates/iceberg/src/arrow/reader/positional_deletes.rs +++ b/crates/iceberg/src/arrow/reader/positional_deletes.rs @@ -40,8 +40,10 @@ impl ArrowReader { let mut results: Vec<RowSelector> = Vec::new(); let mut selected_row_groups_idx = 0; let mut current_row_group_base_idx: u64 = 0; - let mut delete_vector_iter = positional_deletes.iter(); - let mut next_deleted_row_idx_opt = delete_vector_iter.next(); + // Deleted positions as half-open ranges, so a run of deletes costs one step rather than + // one per row. + let mut deleted_ranges = positional_deletes.ranges(); + let mut next_deleted_range = deleted_ranges.next(); for (idx, row_group_metadata) in row_group_metadata_list.iter().enumerate() { let row_group_num_rows = row_group_metadata.num_rows() as u64; @@ -60,16 +62,17 @@ impl ArrowReader { // selected row group selected_row_groups_idx += 1; } else { - // Advance iterator past all deletes in the skipped row group. - // advance_to() positions the iterator to the first delete >= next_row_group_base_idx. - // However, if our cached next_deleted_row_idx_opt is in the skipped range, - // we need to call next() to update the cache with the newly positioned value. - delete_vector_iter.advance_to(next_row_group_base_idx); - // Only update the cache if the cached value is stale (in the skipped range) - if let Some(cached_idx) = next_deleted_row_idx_opt - && cached_idx < next_row_group_base_idx - { - next_deleted_row_idx_opt = delete_vector_iter.next(); + // Skip the deletes in this row group without walking them. A range that + // continues into the next row group is cut to start there. + match next_deleted_range { + Some((_, end)) if end <= next_row_group_base_idx => { + deleted_ranges.skip_to(next_row_group_base_idx); + next_deleted_range = deleted_ranges.next(); + } + Some((start, end)) if start < next_row_group_base_idx => { + next_deleted_range = Some((next_row_group_base_idx, end)); + } + _ => {} } // still increment the current page base index but then skip to the next row group @@ -79,59 +82,29 @@ impl ArrowReader { } } - let mut next_deleted_row_idx = match next_deleted_row_idx_opt { - Some(next_deleted_row_idx) => { - // if the index of the next deleted row is beyond this row group, add a selection for - // the remainder of this row group and skip to the next row group - if next_deleted_row_idx >= next_row_group_base_idx { - results.push(RowSelector::select(row_group_num_rows as usize)); - current_row_group_base_idx += row_group_num_rows; - continue; - } - - next_deleted_row_idx + let mut current_idx = current_row_group_base_idx; + while let Some((start, end)) = next_deleted_range { + if start >= next_row_group_base_idx { + break; } - // If there are no more pos deletes, add a selector for the entirety of this row group. - _ => { - results.push(RowSelector::select(row_group_num_rows as usize)); - current_row_group_base_idx += row_group_num_rows; - continue; + // `select` the rows before the deleted range, then `skip` the part of the range + // that lies in this row group + let skip_start = start.max(current_idx); + let skip_end = end.min(next_row_group_base_idx); + if skip_start > current_idx { + results.push(RowSelector::select((skip_start - current_idx) as usize)); } - }; - - let mut current_idx = current_row_group_base_idx; - 'chunks: while next_deleted_row_idx < next_row_group_base_idx { - // `select` all rows that precede the next delete index - if current_idx < next_deleted_row_idx { - let run_length = next_deleted_row_idx - current_idx; - results.push(RowSelector::select(run_length as usize)); - current_idx += run_length; + if skip_end > skip_start { + results.push(RowSelector::skip((skip_end - skip_start) as usize)); } + current_idx = current_idx.max(skip_end); - // `skip` all consecutive deleted rows in the current row group - let mut run_length = 0; - while next_deleted_row_idx == current_idx - && next_deleted_row_idx < next_row_group_base_idx - { - run_length += 1; - current_idx += 1; - - next_deleted_row_idx_opt = delete_vector_iter.next(); - next_deleted_row_idx = match next_deleted_row_idx_opt { - Some(next_deleted_row_idx) => next_deleted_row_idx, - _ => { - // We've processed the final positional delete. - // Conclude the skip and then break so that we select the remaining - // rows in the row group and move on to the next row group - results.push(RowSelector::skip(run_length)); - break 'chunks; - } - }; - } - if run_length > 0 { - results.push(RowSelector::skip(run_length)); + if end > next_row_group_base_idx { + // the range continues into the next row group + break; } + next_deleted_range = deleted_ranges.next(); } if current_idx < next_row_group_base_idx { @@ -168,7 +141,7 @@ mod tests { use crate::Runtime; use crate::arrow::{ArrowReader, ArrowReaderBuilder}; - use crate::delete_vector::DeleteVector; + use crate::delete_vector::{DeleteVector, mixed_delete_vectors}; use crate::io::FileIO; use crate::scan::{FileScanTask, FileScanTaskDeleteFile, FileScanTaskStream}; use crate::spec::{DataContentType, DataFileFormat, NestedField, PrimitiveType, Schema, Type}; @@ -211,6 +184,202 @@ mod tests { Arc::new(SchemaDescriptor::new(Arc::new(schema))) } + /// Row groups and deleted runs that cross the 2^32 position boundary, with and without row + /// group selection. + #[test] + fn test_build_deletes_row_selection_across_2_pow_32() { + let schema_descr = get_test_schema_descr(); + let columns: Vec<_> = schema_descr + .columns() + .iter() + .map(|ptr| ColumnChunkMetaData::builder(ptr.clone()).build().unwrap()) + .collect(); + let row_groups = |sizes: &[i64]| -> Vec<RowGroupMetaData> { + sizes + .iter() + .enumerate() + .map(|(i, &n)| { + build_test_row_group_meta(schema_descr.clone(), columns.clone(), n, i as i16) + }) + .collect() + }; + let select = RowSelector::select; + let skip = RowSelector::skip; + let high = 1u64 << 32; + let big = high as usize; + + // Row group 1 starts at 2^32 - 2. The run 2^32 - 3 ..= 2^32 crosses both boundaries. + let meta = row_groups(&[(1 << 32) - 2, 8]); + let deletes = DeleteVector::new(RoaringTreemap::from_iter([ + 1, + high - 3, + high - 2, + high - 1, + high, + high + 3, + ])); + let build = |selected: Option<Vec<usize>>| { + ArrowReader::build_deletes_row_selection(&meta, &selected, &deletes).unwrap() + }; + assert_eq!( + build(None), + RowSelection::from(vec![ + select(1), + skip(1), + select(big - 5), + skip(4), + select(2), + skip(1), + select(2), + ]) + ); + assert_eq!( + build(Some(vec![0])), + RowSelection::from(vec![select(1), skip(1), select(big - 5), skip(1)]) + ); + assert_eq!( + build(Some(vec![1])), + RowSelection::from(vec![skip(3), select(2), skip(1), select(2)]) + ); + + // Skipping row group 0 lands past 2^32, in the next bitmap. + let meta = row_groups(&[(1 << 32) + 1, 4]); + let deletes = DeleteVector::new(RoaringTreemap::from_iter([5, high - 1, high, high + 2])); + assert_eq!( + ArrowReader::build_deletes_row_selection(&meta, &Some(vec![1]), &deletes).unwrap(), + RowSelection::from(vec![select(1), skip(1), select(2)]) + ); + } + + /// The selection for `deleted` built one deleted position at a time. + fn expected_deletes_row_selection( + row_group_sizes: &[u64], + selected_row_groups: &Option<Vec<usize>>, + deleted: &[u64], + ) -> RowSelection { + let mut selectors = Vec::new(); + let mut base = 0; + for (idx, &num_rows) in row_group_sizes.iter().enumerate() { + let end = base + num_rows; + if selected_row_groups + .as_ref() + .is_none_or(|selected| selected.contains(&idx)) + { + let mut row = base; + let first = deleted.partition_point(|&p| p < base); + for &pos in deleted[first..].iter().take_while(|&&p| p < end) { + selectors.push(RowSelector::select((pos - row) as usize)); + selectors.push(RowSelector::skip(1)); + row = pos + 1; + } + selectors.push(RowSelector::select((end - row) as usize)); + } + base = end; + } + selectors.into() + } + + /// An empty row group inside a deleted run, once selected and once skipped. + #[test] + fn test_build_deletes_row_selection_empty_row_group_inside_run() { + let schema_descr = get_test_schema_descr(); + let columns: Vec<_> = schema_descr + .columns() + .iter() + .map(|ptr| ColumnChunkMetaData::builder(ptr.clone()).build().unwrap()) + .collect(); + let row_groups: Vec<RowGroupMetaData> = [3, 0, 3] + .iter() + .enumerate() + .map(|(i, &n)| { + build_test_row_group_meta(schema_descr.clone(), columns.clone(), n, i as i16) + }) + .collect(); + // Rows 1..5 are deleted: two in row group 0 and two in row group 2. + let deletes = DeleteVector::new(RoaringTreemap::from_iter(1..5)); + let select = RowSelector::select; + let skip = RowSelector::skip; + for (selected, expected) in [ + (None, vec![select(1), skip(4), select(1)]), + (Some(vec![0, 1, 2]), vec![select(1), skip(4), select(1)]), + (Some(vec![0, 2]), vec![select(1), skip(4), select(1)]), + (Some(vec![1, 2]), vec![skip(2), select(1)]), + ] { + assert_eq!( + ArrowReader::build_deletes_row_selection(&row_groups, &selected, &deletes).unwrap(), + RowSelection::from(expected), + "selected {selected:?}" + ); + } + } + + /// Random row group layouts and selections over the delete vectors the iterator tests use, + /// compared with a selection built one deleted position at a time. + #[test] + fn test_build_deletes_row_selection_matches_model() { + use rand::rngs::StdRng; + use rand::{Rng, SeedableRng}; + + let mut rng = StdRng::seed_from_u64(3); + let schema_descr = get_test_schema_descr(); + let columns: Vec<_> = schema_descr + .columns() + .iter() + .map(|ptr| ColumnChunkMetaData::builder(ptr.clone()).build().unwrap()) + .collect(); + + for deletes in mixed_delete_vectors() { + let deleted: Vec<u64> = deletes.iter().collect(); + for _ in 0..8 { + // Half the time the file ends at or before a deleted position, so the deletes + // past its end must be ignored. + let num_rows = match deleted.len() { + n if n > 0 && rng.random_bool(0.5) => { + deleted[rng.random_range(0..n)].saturating_add(rng.random_range(0..2)) + } + _ => deleted.last().map_or(0, |&last| last + 1) + rng.random_range(0..100), + } + .max(1); + // Row group boundaries, half of them next to a deleted position so that they + // fall inside runs. Repeated cuts leave empty row groups. + let mut cuts: Vec<u64> = (0..rng.random_range(0..8)) + .map(|_| match deleted.len() { + n if n > 0 && rng.random_bool(0.5) => (deleted[rng.random_range(0..n)] + + rng.random_range(0..3)) + .saturating_sub(1), + _ => rng.random_range(0..num_rows), + }) + .filter(|&cut| cut > 0 && cut < num_rows) + .collect(); + cuts.extend([0, num_rows]); + cuts.sort_unstable(); + let sizes: Vec<u64> = cuts.windows(2).map(|w| w[1] - w[0]).collect(); + let row_groups: Vec<RowGroupMetaData> = sizes + .iter() + .enumerate() + .map(|(i, &n)| { + build_test_row_group_meta( + schema_descr.clone(), + columns.clone(), + n as i64, + i as i16, + ) + }) + .collect(); + let selected = rng + .random_bool(0.75) + .then(|| (0..sizes.len()).filter(|_| rng.random_bool(0.5)).collect()); + + assert_eq!( + ArrowReader::build_deletes_row_selection(&row_groups, &selected, &deletes) + .unwrap(), + expected_deletes_row_selection(&sizes, &selected, &deleted), + "row groups {sizes:?}, selected {selected:?}" + ); + } + } + } + #[test] fn test_build_deletes_row_selection() { let schema_descr = get_test_schema_descr(); diff --git a/crates/iceberg/src/delete_vector.rs b/crates/iceberg/src/delete_vector.rs index dc76eeadb..f39140ccc 100644 --- a/crates/iceberg/src/delete_vector.rs +++ b/crates/iceberg/src/delete_vector.rs @@ -92,6 +92,20 @@ impl DeleteVector { self.inner.is_empty() } + /// Returns the deleted positions as ascending, non-overlapping half-open ranges. + /// + /// Adjacent ranges may not be merged (a run is split at each 2^32 boundary), so callers + /// shouldn't assume a range is maximal. A range holding only `u64::MAX` is dropped, because a + /// half-open `u64` range can't hold it and no Parquet row can be there. + pub(crate) fn ranges(&self) -> DeleteVectorRanges<'_> { + DeleteVectorRanges { + bitmaps: self.inner.bitmaps(), + high_bits: 0, + bitmap_iter: None, + buffered: None, + } + } + /// Parses a `deletion-vector-v1` Puffin blob into a `DeleteVector`. /// /// The layout, defined by the Iceberg Puffin spec and matching Iceberg-Java's @@ -225,17 +239,8 @@ fn decode_roaring_directory(mut reader: &[u8]) -> Result<RoaringTreemap> { Ok(RoaringTreemap::from_bitmaps(bitmaps)) } -// Ideally, we'd just wrap `roaring::RoaringTreemap`'s iterator, `roaring::treemap::Iter` here. -// But right now, it does not have a corresponding implementation of `roaring::bitmap::Iter::advance_to`, -// which is very handy in ArrowReader::build_deletes_row_selection. -// There is a PR open on roaring to add this (https://github.com/RoaringBitmap/roaring-rs/pull/314) -// and if that gets merged then we can simplify `DeleteVectorIterator` here, refactoring `advance_to` -// to just a wrapper around the underlying iterator's method. /// Iterates the deleted positions of a [`DeleteVector`] in ascending order. pub struct DeleteVectorIterator<'a> { - // NB: `BitMapIter` was only exposed publicly in https://github.com/RoaringBitmap/roaring-rs/pull/316 - // which is not yet released. As a consequence our Cargo.toml temporarily uses a git reference for - // the roaring dependency. outer: BitmapIter<'a>, inner: Option<DeleteVectorIteratorInner<'a>>, } @@ -270,27 +275,166 @@ impl Iterator for DeleteVectorIterator<'_> { impl DeleteVectorIterator<'_> { /// Skips the positions below `pos`, so the next one yielded is the first - /// at or after it. Does nothing before the first position is yielded. + /// at or after it. A `pos` below the next position does nothing. pub fn advance_to(&mut self, pos: u64) { let hi = (pos >> 32) as u32; let lo = pos as u32; - let Some(ref mut inner) = self.inner else { + if let Some(inner) = &mut self.inner { + if inner.high_bits > hi { + return; + } + if inner.high_bits == hi { + inner.bitmap_iter.advance_to(lo); + return; + } + } + + // `pos` is in a later bitmap, or nothing has been yielded yet. + self.inner = None; + for (high_bits, bitmap) in self.outer.by_ref() { + if high_bits < hi { + continue; + } + let mut bitmap_iter = bitmap.iter(); + if high_bits == hi { + bitmap_iter.advance_to(lo); + } + self.inner = Some(DeleteVectorIteratorInner { + high_bits, + bitmap_iter, + }); return; - }; + } + } +} - while inner.high_bits < hi { - let Some((next_hi, next_bitmap)) = self.outer.next() else { - return; +/// Number of consecutive positions [`DeleteVectorRanges`] walks one by one before it asks the +/// bitmap for the rest of the run. +/// +/// `roaring`'s `next_range` binary-searches the rest of an array container on every call, which +/// costs more than `next` when most runs are a single position, so it only pays off for long +/// runs. Tuned with `roaring` 0.11.4 by reading a 20M-row file with deletes in runs of 20, 100 +/// and longer, and as scattered single rows and pairs: 4 was best on runs, compared with 8, 16 +/// and 32, and no slower on scattered deletes. +const RUN_PROBE_LEN: u32 = 4; + +/// What [`DeleteVectorRanges`] has taken from its bitmap iterator but not returned yet. +enum Buffered { + /// Start of the next range. + Pos(u32), + /// A complete range, inclusive end. + Range(u32, u32), +} + +/// Iterator over a [`DeleteVector`]'s deleted positions as half-open ranges, see +/// [`DeleteVector::ranges`]. +pub(crate) struct DeleteVectorRanges<'a> { + bitmaps: BitmapIter<'a>, + /// High 32 bits of the positions in `bitmap_iter` and `buffered`. + high_bits: u32, + bitmap_iter: Option<Iter<'a>>, + buffered: Option<Buffered>, +} + +impl DeleteVectorRanges<'_> { + /// Returns `None` only for a range holding nothing but `u64::MAX`, see [`DeleteVector::ranges`]. + fn full_range(&self, start: u32, end_inclusive: u32) -> Option<(u64, u64)> { + let base = u64::from(self.high_bits) << 32; + let start = base | u64::from(start); + let end = (base | u64::from(end_inclusive)).saturating_add(1); + (start < end).then_some((start, end)) + } + + /// Skips every deleted position below `pos`. A range still buffered here that starts below + /// `pos` and ends after it is cut to start at `pos`; a range already returned by `next` is + /// the caller's to cut. + pub(crate) fn skip_to(&mut self, pos: u64) { + let hi = (pos >> 32) as u32; + let lo = pos as u32; + + if self.bitmap_iter.is_some() && self.high_bits > hi { + return; + } + if let Some(iter) = self.bitmap_iter.as_mut() + && self.high_bits == hi + { + self.buffered = match self.buffered.take() { + Some(Buffered::Pos(start)) if start < lo => None, + Some(Buffered::Range(_, end)) if end < lo => None, + Some(Buffered::Range(start, end)) if start < lo => Some(Buffered::Range(lo, end)), + buffered => buffered, }; + iter.advance_to(lo); + return; + } - *inner = DeleteVectorIteratorInner { - high_bits: next_hi, - bitmap_iter: next_bitmap.iter(), + // `pos` is in a later bitmap: drop everything buffered from the current one. + self.bitmap_iter = None; + self.buffered = None; + for (high_bits, bitmap) in self.bitmaps.by_ref() { + if high_bits < hi { + continue; + } + let mut iter = bitmap.iter(); + if high_bits == hi { + iter.advance_to(lo); } + self.high_bits = high_bits; + self.bitmap_iter = Some(iter); + return; } + } +} - inner.bitmap_iter.advance_to(lo); +impl Iterator for DeleteVectorRanges<'_> { + type Item = (u64, u64); + + fn next(&mut self) -> Option<Self::Item> { + loop { + let pending = match self.buffered.take() { + Some(Buffered::Range(start, end)) => return self.full_range(start, end), + Some(Buffered::Pos(start)) => Some(start), + None => None, + }; + + if let Some(iter) = self.bitmap_iter.as_mut() + && let Some(start) = pending.or_else(|| iter.next()) + { + // Walk short runs with `next`, and ask for the rest of a long run at once. + let mut end = start; + let mut walked = 0; + loop { + if walked == RUN_PROBE_LEN { + match iter.next_range() { + Some(run) if Some(*run.start()) == end.checked_add(1) => { + end = *run.end(); + } + Some(run) => { + self.buffered = Some(Buffered::Range(*run.start(), *run.end())) + } + None => {} + } + break; + } + match iter.next() { + Some(next) if Some(next) == end.checked_add(1) => { + end = next; + walked += 1; + } + next => { + self.buffered = next.map(Buffered::Pos); + break; + } + } + } + return self.full_range(start, end); + } + + let (high_bits, bitmap) = self.bitmaps.next()?; + self.high_bits = high_bits; + self.bitmap_iter = Some(bitmap.iter()); + } } } @@ -316,10 +460,271 @@ pub(crate) fn frame_dv_blob(vector: &[u8]) -> Vec<u8> { blob } +/// Deterministic pseudo-random sets mixing single positions, short and long runs, runs across +/// 2^16 and 2^32 boundaries, and empty bitmaps between used ones. +#[cfg(test)] +pub(crate) fn mixed_delete_vectors() -> Vec<DeleteVector> { + use rand::rngs::StdRng; + use rand::{Rng, SeedableRng}; + + let mut rng = StdRng::seed_from_u64(7); + let mut vectors = vec![DeleteVector::default()]; + for case in 0..40u64 { + let mut dv = DeleteVector::default(); + let base = if case % 4 == 0 { (1u64 << 32) - 40 } else { 0 }; + let mut pos = base + rng.random_range(0..50); + for _ in 0..rng.random_range(0..60) { + let run = match rng.random_range(0..5) { + 0 => 1, + 1 => 2, + // one past the probe length: the rest comes from `next_range` + 2 => u64::from(RUN_PROBE_LEN) + 1, + 3 => rng.random_range(1..=40), + _ => rng.random_range(1..=70_000), + }; + dv.inner.insert_range(pos..pos + run); + pos += run + rng.random_range(1..=30); + } + vectors.push(dv); + } + let mut boundaries = DeleteVector::default(); + boundaries.inner.insert_range((1 << 16) - 3..(1 << 16) + 3); + boundaries + .inner + .insert_range((1 << 32) - 20..(1 << 32) + 20); + boundaries.inner.insert((2 << 32) + 7); + vectors.push(boundaries); + // High bits 0, 3 and 7, so skipping walks past empty bitmaps. + let mut gaps = DeleteVector::default(); + gaps.inner.insert_range(2..12); + gaps.inner.insert_range((3 << 32) + 5..(3 << 32) + 70_000); + gaps.inner.insert((7 << 32) + 1); + vectors.push(gaps); + vectors +} + #[cfg(test)] mod tests { + use rand::rngs::StdRng; + use rand::{Rng, SeedableRng}; + use super::*; + #[test] + fn test_ranges_cover_exactly_the_deleted_positions() { + for dv in mixed_delete_vectors() { + let positions: Vec<u64> = dv.iter().collect(); + let mut expanded = Vec::new(); + let mut last_end = None; + for (start, end) in dv.ranges() { + assert!(start < end, "empty range {start}..{end}"); + assert!( + last_end.is_none_or(|last| last <= start), + "ranges out of order" + ); + last_end = Some(end); + expanded.extend(start..end); + } + assert_eq!(expanded, positions); + } + } + + #[test] + fn test_ranges_skip_to_matches_model() { + let mut rng = StdRng::seed_from_u64(7); + + for dv in mixed_delete_vectors() { + let deleted: Vec<u64> = dv.iter().collect(); + let max = deleted.last().copied().unwrap_or(0) + 10; + // `deleted[next..]` has been neither returned nor skipped, and every position below + // `done` has. + let mut next = 0; + let mut done = 0u64; + let mut ranges = dv.ranges(); + loop { + match rng.random_range(0..6) { + 0 | 1 => { + let target = done.max(deleted.get(next).copied().unwrap_or(max)) + + rng.random_range(0..200); + ranges.skip_to(target); + next = deleted.partition_point(|&p| p < target); + done = target; + continue; + } + // A target at or below `done`, near it or anywhere below: does nothing. + 2 => { + let back = if rng.random_bool(0.5) { + done.saturating_sub(rng.random_range(0..200)) + } else { + rng.random_range(0..=done) + }; + ranges.skip_to(back); + continue; + } + _ => {} + } + match ranges.next() { + Some((start, end)) => { + assert!(start < end, "empty range {start}..{end}"); + let len = (end - start) as usize; + assert_eq!( + deleted + .get(next..next + len) + .map(|run| (run[0], run[len - 1])), + Some((start, end - 1)), + "returned {start}..{end}" + ); + next += len; + done = end; + } + None => { + assert_eq!(next, deleted.len(), "missed {:?}", &deleted[next..]); + break; + } + } + } + } + } + + #[test] + fn test_advance_to_matches_model() { + let mut rng = StdRng::seed_from_u64(11); + + for dv in mixed_delete_vectors() { + let deleted: Vec<u64> = dv.iter().collect(); + let max = deleted.last().copied().unwrap_or(0) + 10; + let mut next = 0; + let mut done = 0u64; + let mut iter = dv.iter(); + loop { + match rng.random_range(0..6) { + 0 => { + let target = done.max(deleted.get(next).copied().unwrap_or(max)) + + rng.random_range(0..200); + iter.advance_to(target); + next = deleted.partition_point(|&p| p < target); + done = target; + continue; + } + 1 => { + let back = if rng.random_bool(0.5) { + done.saturating_sub(rng.random_range(0..200)) + } else { + rng.random_range(0..=done) + }; + iter.advance_to(back); + continue; + } + _ => {} + } + let pos = iter.next(); + assert_eq!(pos, deleted.get(next).copied()); + match pos { + Some(pos) => { + next += 1; + done = pos + 1; + } + None => break, + } + } + } + } + + #[test] + fn test_ranges_skip_to_inside_buffered_run() { + // After a run of exactly RUN_PROBE_LEN + 1, the following run is already buffered. + let probe = u64::from(RUN_PROBE_LEN); + let mut dv = DeleteVector::default(); + dv.inner.insert_range(0..probe + 1); + dv.inner.insert_range(probe + 3..probe + 13); + let mut ranges = dv.ranges(); + assert_eq!(ranges.next(), Some((0, probe + 1))); + ranges.skip_to(probe + 8); + assert_eq!(ranges.next(), Some((probe + 8, probe + 13))); + assert_eq!(ranges.next(), None); + } + + #[test] + fn test_ranges_join_runs_across_2_16() { + // `roaring`'s `next_range` joins a run across containers, so these come back whole. + // Starting near the boundary also covers the positions walked before `next_range`. + for start in [(1 << 16) - 100, (1 << 16) - 2] { + let mut dv = DeleteVector::default(); + dv.inner.insert_range(start..(3 << 16) + 5); + assert_eq!(dv.ranges().collect::<Vec<_>>(), vec![( + start, + (3 << 16) + 5 + )]); + } + } + + #[test] + fn test_ranges_end_at_u64_max() { + let dv = DeleteVector::new(RoaringTreemap::from_iter([5, u64::MAX - 2, u64::MAX - 1])); + assert_eq!(dv.ranges().collect::<Vec<_>>(), vec![ + (5, 6), + (u64::MAX - 2, u64::MAX) + ]); + + // `u64::MAX` itself can't end a half-open range, and no row can be there. + let dv = DeleteVector::new(RoaringTreemap::from_iter([5, u64::MAX - 1, u64::MAX])); + assert_eq!(dv.ranges().collect::<Vec<_>>(), vec![ + (5, 6), + (u64::MAX - 1, u64::MAX) + ]); + let dv = DeleteVector::new(RoaringTreemap::from_iter([u64::MAX])); + assert_eq!(dv.ranges().next(), None); + } + + // The cases from https://github.com/apache/iceberg-rust/issues/3400, for `skip_to` and for + // `advance_to`. + #[test] + fn test_ranges_skip_to_missing_later_bitmap() { + let dv = DeleteVector::new(RoaringTreemap::from_iter([1u64, 5, 9])); + let mut ranges = dv.ranges(); + assert_eq!(ranges.next(), Some((1, 2))); + ranges.skip_to((1 << 32) + 5); + assert_eq!(ranges.next(), None); + } + + #[test] + fn test_ranges_skip_to_below_current_bitmap() { + let high = 1u64 << 32; + let dv = DeleteVector::new(RoaringTreemap::from_iter([high + 1, high + 2, high + 9])); + let mut ranges = dv.ranges(); + assert_eq!(ranges.next(), Some((high + 1, high + 3))); + ranges.skip_to(10); + assert_eq!(ranges.next(), Some((high + 9, high + 10))); + assert_eq!(ranges.next(), None); + } + + #[test] + fn test_advance_to_missing_later_bitmap() { + let dv = DeleteVector::new(RoaringTreemap::from_iter([5u64, 7])); + let mut iter = dv.iter(); + assert_eq!(iter.next(), Some(5)); + iter.advance_to((1 << 32) + 10); + assert_eq!(iter.next(), None); + } + + #[test] + fn test_advance_to_below_current_bitmap() { + let high = 1u64 << 32; + let dv = DeleteVector::new(RoaringTreemap::from_iter([high + 10, high + 20])); + let mut iter = dv.iter(); + assert_eq!(iter.next(), Some(high + 10)); + iter.advance_to(50); + assert_eq!(iter.collect::<Vec<_>>(), vec![high + 20]); + } + + #[test] + fn test_advance_to_before_first_next() { + let dv = DeleteVector::new(RoaringTreemap::from_iter([5u64, 200])); + let mut iter = dv.iter(); + iter.advance_to(100); + assert_eq!(iter.collect::<Vec<_>>(), vec![200]); + } + #[test] fn test_insertion_and_iteration() { let mut dv = DeleteVector::default();
