Decode a deletion vector once across the splits of a DV-backed file in the native Delta scan

What is the problem the feature request solves?

Follow-up from the native Delta Lake scan review (#5365). The native scan splits DV-backed files like any other file, and each split already builds its access plan only for the row groups it owns: build_access_plan takes the split's FileRange, skips row groups whose first page starts outside it while keeping file-global row positions, validates the deletion vector against the whole file, and reserves memory only for the selectors the split keeps. The split_access_plan_* tests in delta_dv.rs pin that behavior.

What is left is the deletion vector itself. attach_access_plan runs once per split, and every split of a file fetches the DV blob (or takes the inline payload), checks its CRC, deserializes the whole bitmap and validates its cardinality, even though it only uses the deletions inside its own row groups. A DV-backed file larger than maxSplitBytes repeats that work once per split, and the splits usually run as separate tasks.

Describe the potential solution

Decode a file's deletion vector once and share it across that file's splits, for example through a bounded executor-level cache keyed by the DV descriptor (path, offset and size, or the inline payload) and charged to the memory pool. Start with a split-file benchmark (one large DV-backed file read at several maxPartitionBytes settings) to measure the per-split DV cost before and after, since for a small DV the fetch may already be cheap next to the Parquet read.

Additional context

Raised in the #5365 review discussion. Earlier versions of this issue asked for split-aware access plans, which landed with #5365. #6945 (closes #5656) keeps native-format DV bitmaps compressed while decoding, which lowers each split's decode cost but does not remove the repeated work.

@ErikBPF

Related candidate patch: dwsmith1983#4 by @ErikBPF replaces the per row deletion walk in build_access_plan with coalesced deletion runs clipped across row group and 32 bit partition boundaries, collapsing fully deleted groups to skips. Single commit on delta_dv.rs, applies cleanly to the current #5365 head, 36 focused tests including a run crossing the partition boundary. It does not enable splitting by itself but is the natural base for split aware preparation; plan is for it to be opened here once #5365 lands.

The problem statement above was wrong: DV-backed files are split today. Rewrote it to say what each split currently does and that the work here is making a split cheaper, not enabling splitting.

Narrowed the scope of this issue, since most of what it asked for is already on main.

When this was filed, every split of a DV-backed file built an access plan for the whole file and reserved memory for all of its selectors. That changed in #5365 before it merged. build_access_plan now takes the split's FileRange, and only the row groups the split owns get selectors:

datafusion-comet/native/core/src/execution/delta_dv.rs

Lines 163 to 254 in 967d09b

    
           /// Translate deleted row indexes into a [`ParquetAccessPlan`]: fully-deleted 
        
           /// row groups become `Skip`, untouched groups stay `Scan`, and partially 
        
           /// deleted groups get a `RowSelection` selecting the complement of the deleted 
        
           /// rows. Page-index pruning later INTERSECTS with these selections, so DV 
        
           /// skips and page skips compose. When a file range is present, only its row groups acquire 
        
           /// selectors; skipped groups still advance the file-global deletion-vector positions. 
        
           pub fn build_access_plan( 
        
               metadata: &ParquetMetaData, 
        
               deleted: &RoaringTreemap, 
        
               file_range: Option<&FileRange>, 
        
           ) -> Result<ParquetAccessPlan, ExecutionError> { 
        
               let mut plan = ParquetAccessPlan::new_all(metadata.num_row_groups()); 
        
               // Empty bitmaps create no selectors to avoid. Let the reader apply its ordinary range 
        
               // pruning instead of doing an extra page-offset pass for this case. 
        
               let file_range = file_range.filter(|_| !deleted.is_empty()); 
        
               let mut deleted_iter = deleted.iter(); 
        
               let mut next_deleted = deleted_iter.next(); 
        
               let mut group_start = 0u64; 
        
               for (idx, row_group) in metadata.row_groups().iter().enumerate() { 
        
                   let num_rows = row_group.num_rows(); 
        
                   // A corrupt footer can report a negative row count. `num_rows as u64` would otherwise 
        
                   // wrap it into a huge positive value, silently corrupting every row-group boundary 
        
                   // computed from `group_start`/`group_end` below (and therefore which deleted row indexes 
        
                   // land in which row group) instead of failing loudly. 
        
                   if num_rows < 0 { 
        
                       return Err(GeneralError(format!( 
        
                           "Parquet footer reports a negative row count ({num_rows}) for row group {idx}" 
        
                       ))); 
        
                   } 
        
                   let num_rows = num_rows as u64; 
        
                   let group_end = group_start.checked_add(num_rows).ok_or_else(|| { 
        
                       GeneralError(format!( 
        
                           "Parquet footer row counts overflow at row group {idx} ({group_start} + {num_rows})" 
        
                       )) 
        
                   })?; 
        
                   // Match DataFusion's prune_by_range: dictionary-page start takes precedence over 
        
                   // data-page start, and FileRange is half-open. Keep an entry for every row group and 
        
                   // advance file-global row positions even when this split cannot read the group. 
        
                   if file_range.is_some_and(|range| { 
        
                       let column = row_group.column(0); 
        
                       let offset = column 
        
                           .dictionary_page_offset() 
        
                           .unwrap_or_else(|| column.data_page_offset()); 
        
                       !range.contains(offset) 
        
                   }) { 
        
                       plan.skip(idx); 
        
                       group_start = group_end; 
        
                       continue; 
        
                   } 
        
                   // Seeking skips deletion positions from preceding, excluded groups without visiting 
        
                   // each one. The pinned Roaring implementation also handles absent 32-bit containers. 
        
                   if next_deleted.is_some_and(|row| row < group_start) { 
        
                       deleted_iter.advance_to(group_start); 
        
                       next_deleted = deleted_iter.next(); 
        
                   } 
        
                   let mut selectors: Vec<RowSelector> = Vec::new(); 
        
                   let mut cursor = group_start; 
        
                   let mut deleted_in_group = 0u64; 
        
                   while let Some(row) = next_deleted.filter(|row| *row < group_end) { 
        
                       next_deleted = deleted_iter.next(); 
        
                       deleted_in_group += 1; 
        
                       if row > cursor { 
        
                           selectors.push(RowSelector::select((row - cursor) as usize)); 
        
                       } 
        
                       // Merge runs of consecutive deleted rows into one skip. 
        
                       match selectors.last_mut() { 
        
                           Some(last) if last.skip => last.row_count += 1, 
        
                           _ => selectors.push(RowSelector::skip(1)), 
        
                       } 
        
                       cursor = row + 1; 
        
                   } 
        
                   if deleted_in_group == num_rows && num_rows > 0 { 
        
                       plan.skip(idx); 
        
                   } else if deleted_in_group > 0 { 
        
                       if group_end > cursor { 
        
                           selectors.push(RowSelector::select((group_end - cursor) as usize)); 
        
                       } 
        
                       plan.scan_selection(idx, RowSelection::from(selectors)); 
        
                   } 
        
                   group_start = group_end; 
        
               } 
        
               // Validate against the whole file even when this split excludes the final groups. 
        
               // A deleted index beyond the file's total row count means the DV does not belong to this 
        
               // file. Checking the maximum also catches corruption beyond an absent Roaring container. 
        
               if let Some(row) = deleted.max().filter(|row| *row >= group_start) { 
        
                   return Err(GeneralError(format!( 
        
                       "Deletion vector marks row {row} but the file only has {group_start} rows" 
        
                   ))); 
        
               } 
        
               Ok(plan) 
        
           }

It keeps file-global row positions across the groups it skips, validates the deletion vector against the whole file, follows DataFusion's half-open dictionary-page boundaries, and reserves memory only for the selectors it keeps. The split_access_plan_* tests cover each of these:

datafusion-comet/native/core/src/execution/delta_dv.rs

Lines 1278 to 1523 in 967d09b

    
               #[test] 
        
               fn split_access_plan_preserves_file_global_positions() { 
        
                   let metadata = 
        
                       metadata_with_page_offsets(&[3, 3, 3], &[(10, None), (20, None), (30, None)]); 
        
                   let plan = build_access_plan( 
        
                       &metadata, 
        
                       &RoaringTreemap::from_iter([1, 4, 7]), 
        
                       Some(&FileRange { start: 20, end: 21 }), 
        
                   ) 
        
                   .unwrap(); 
        
                   assert_eq!(plan.inner()[0], RowGroupAccess::Skip); 
        
                   assert_eq!(plan.inner()[2], RowGroupAccess::Skip); 
        
                   assert_eq!( 
        
                       plan.inner()[1], 
        
                       RowGroupAccess::Selection(RowSelection::from(vec![ 
        
                           RowSelector::select(1), 
        
                           RowSelector::skip(1), 
        
                           RowSelector::select(1), 
        
                       ])) 
        
                   ); 
        
               } 
        
               #[test] 
        
               fn split_access_plan_uses_half_open_dictionary_page_boundaries() { 
        
                   let metadata = 
        
                       metadata_with_page_offsets(&[3, 3, 3], &[(20, Some(10)), (25, Some(20)), (30, None)]); 
        
                   let plan = build_access_plan( 
        
                       &metadata, 
        
                       &RoaringTreemap::from_iter([1]), 
        
                       Some(&FileRange { start: 20, end: 30 }), 
        
                   ) 
        
                   .unwrap(); 
        
                   assert_eq!( 
        
                       plan.inner(), 
        
                       &[ 
        
                           RowGroupAccess::Skip, 
        
                           RowGroupAccess::Scan, 
        
                           RowGroupAccess::Skip 
        
                       ] 
        
                   ); 
        
               } 
        
               #[test] 
        
               fn split_access_plan_validates_excluded_rows_and_row_groups() { 
        
                   let metadata = 
        
                       metadata_with_page_offsets(&[3, 3, 3], &[(10, None), (20, None), (30, None)]); 
        
                   for range in [ 
        
                       FileRange { start: 20, end: 21 }, 
        
                       FileRange { start: 40, end: 50 }, 
        
                   ] { 
        
                       let err = 
        
                           build_access_plan(&metadata, &RoaringTreemap::from_iter([1, 9]), Some(&range)) 
        
                               .unwrap_err(); 
        
                       assert!(err.to_string().contains("only has 9 rows")); 
        
                   } 
        
                   // Metadata must be valid even when the split excludes the corrupt row group. 
        
                   let metadata = metadata_with_page_offsets(&[3, -1], &[(10, None), (20, None)]); 
        
                   assert!(build_access_plan( 
        
                       &metadata, 
        
                       &RoaringTreemap::new(), 
        
                       Some(&FileRange { start: 10, end: 11 }), 
        
                   ) 
        
                   .is_err()); 
        
                   let metadata = metadata_with_page_offsets( 
        
                       &[i64::MAX, i64::MAX, i64::MAX], 
        
                       &[(10, None), (20, None), (30, None)], 
        
                   ); 
        
                   assert!(build_access_plan( 
        
                       &metadata, 
        
                       &RoaringTreemap::new(), 
        
                       Some(&FileRange { start: 10, end: 11 }), 
        
                   ) 
        
                   .is_err()); 
        
               } 
        
               #[test] 
        
               fn split_access_plan_seeks_across_present_and_absent_roaring_containers() { 
        
                   for (first_group_rows, second_group_rows, marked_row) in [ 
        
                       ((1_i64 << 32) + 3, 3, (1_u64 << 32) + 4), 
        
                       ((1_i64 << 32) + 100, 1_i64 << 32, (2_u64 << 32) + 50), 
        
                   ] { 
        
                       let metadata = metadata_with_page_offsets( 
        
                           &[first_group_rows, second_group_rows], 
        
                           &[(10, None), (20, None)], 
        
                       ); 
        
                       let plan = build_access_plan( 
        
                           &metadata, 
        
                           &RoaringTreemap::from_iter([1, 2, marked_row]), 
        
                           Some(&FileRange { start: 20, end: 21 }), 
        
                       ) 
        
                       .unwrap(); 
        
                       assert_eq!(plan.inner()[0], RowGroupAccess::Skip); 
        
                       let RowGroupAccess::Selection(selection) = &plan.inner()[1] else { 
        
                           panic!("expected a selection in the second row group"); 
        
                       }; 
        
                       assert_eq!(selection.row_count(), second_group_rows as usize - 1); 
        
                       assert_eq!(selection.skipped_row_count(), 1); 
        
                   } 
        
                   // A corrupt position beyond an absent high-32-bit container remains an error. 
        
                   let metadata = metadata_with_page_offsets(&[(1_i64 << 32) + 100], &[(10, None)]); 
        
                   assert!(build_access_plan( 
        
                       &metadata, 
        
                       &RoaringTreemap::from_iter([1, (2_u64 << 32) + 50]), 
        
                       Some(&FileRange { start: 20, end: 21 }), 
        
                   ) 
        
                   .is_err()); 
        
               } 
        
               #[test] 
        
               fn split_access_plan_has_no_selectors_for_excluded_groups() { 
        
                   let metadata = metadata_with_page_offsets( 
        
                       &[1_000, 1_000, 1_000], 
        
                       &[(10, None), (20, None), (30, None)], 
        
                   ); 
        
                   let deleted = (0..1_000).chain(2_000..3_000).step_by(2).collect(); 
        
                   let plan = build_access_plan(&metadata, &deleted, Some(&FileRange { start: 20, end: 21 })) 
        
                       .unwrap(); 
        
                   assert_eq!(total_selectors(&plan), 0); 
        
                   assert_eq!(plan.inner()[1], RowGroupAccess::Scan); 
        
                   assert!(total_selectors(&build_access_plan(&metadata, &deleted, None).unwrap()) > 0); 
        
               } 
        
               /// The confirmed worst case: deleting every even row leaves 
        
               /// no adjacent skips or selects to merge, so `build_access_plan` emits 
        
               /// one non-coalescing `RowSelector` per row of the group. 
        
               fn alternating_deleted(num_rows: u64) -> RoaringTreemap { 
        
                   (0..num_rows).step_by(2).collect() 
        
               } 
        
               #[test] 
        
               fn total_selectors_counts_one_per_row_for_alternating_bitmap() { 
        
                   let deleted = alternating_deleted(1024); 
        
                   let plan = full_file_access_plan(&[1024], &deleted).unwrap(); 
        
                   assert_eq!(total_selectors(&plan), 1024); 
        
               } 
        
               #[test] 
        
               fn total_selectors_ignores_scan_and_skip_row_groups() { 
        
                   // Group 0 untouched (Scan), group 1 fully deleted (Skip): neither 
        
                   // carries a RowSelection, so both must contribute zero selectors. 
        
                   let deleted: RoaringTreemap = (10u64..20).collect(); 
        
                   let plan = full_file_access_plan(&[10, 10], &deleted).unwrap(); 
        
                   assert_eq!(total_selectors(&plan), 0); 
        
               } 
        
               /// Writes one file's on-disk parquet data for a full-file, alternating-bitmap deletion 
        
               /// vector, returning its path, byte size, and deleted-row bitmap so callers needing the 
        
               /// file's on-disk metadata (to size a memory pool exactly, or to replay the real reader 
        
               /// path) can inspect it before building a [`DvScanFile`] from it. 
        
               fn write_alternating_parquet( 
        
                   dir: &std::path::Path, 
        
                   num_rows: i64, 
        
               ) -> (std::path::PathBuf, i64, RoaringTreemap) { 
        
                   let deleted = alternating_deleted(num_rows as u64); 
        
                   let path = dir.join("alternating.parquet"); 
        
                   let size = write_parquet(&path, num_rows); 
        
                   (path, size, deleted) 
        
               } 
        
               /// Builds a [`DvScanFile`] with an inline deletion vector for an already-written parquet 
        
               /// file at `path`. 
        
               fn dv_scan_file_for_alternating( 
        
                   runtime_env: &Arc<RuntimeEnv>, 
        
                   path: &std::path::Path, 
        
                   size: i64, 
        
                   deleted: &RoaringTreemap, 
        
               ) -> DvScanFile { 
        
                   let inline_data = portable_bytes(deleted); 
        
                   let file_path = format!("file://{}", path.display()); 
        
                   let (data_store, _) = resolve_store(runtime_env, &file_path); 
        
                   DvScanFile { 
        
                       file: PartitionedFile::new(path.display().to_string(), size as u64), 
        
                       file_path, 
        
                       dv: Some(DeltaSparkDvDescriptor { 
        
                           storage_type: "i".to_string(), 
        
                           absolute_path: None, 
        
                           inline_data: Some(inline_data.clone()), 
        
                           offset: None, 
        
                           size_in_bytes: inline_data.len() as i32, 
        
                           cardinality: deleted.len() as i64, 
        
                       }), 
        
                       data_store, 
        
                       dv_store: None, 
        
                   } 
        
               } 
        
               #[tokio::test] 
        
               async fn split_access_plan_reservation_tracks_only_retained_selectors() { 
        
                   let tmp = tempfile::tempdir().unwrap(); 
        
                   let path = tmp.path().join("split.parquet"); 
        
                   let size = write_parquet_with_properties( 
        
                       &path, 
        
                       3_000, 
        
                       WriterProperties::builder() 
        
                           .set_max_row_group_row_count(Some(1_000)) 
        
                           .build(), 
        
                   ); 
        
                   let metadata = read_metadata_with_page_index(&path); 
        
                   let column = metadata.row_group(1).column(0); 
        
                   let offset = column 
        
                       .dictionary_page_offset() 
        
                       .unwrap_or_else(|| column.data_page_offset()); 
        
                   let pool: Arc<dyn MemoryPool> = Arc::new(GreedyMemoryPool::new(10_000_000)); 
        
                   let runtime_env = RuntimeEnvBuilder::new() 
        
                       .with_memory_pool(Arc::clone(&pool)) 
        
                       .build_arc() 
        
                       .unwrap(); 
        
                   let deleted = alternating_deleted(3_000); 
        
                   let full = dv_scan_file_for_alternating(&runtime_env, &path, size, &deleted); 
        
                   let mut split = dv_scan_file_for_alternating(&runtime_env, &path, size, &deleted); 
        
                   split.file.range = Some(FileRange { 
        
                       start: offset, 
        
                       end: offset + 1, 
        
                   }); 
        
                   let files = attach_access_plans(runtime_env, vec![full, split]) 
        
                       .await 
        
                       .unwrap(); 
        
                   let full_plan = files[0].extensions.get::<ParquetAccessPlan>().unwrap(); 
        
                   let split_plan = files[1].extensions.get::<ParquetAccessPlan>().unwrap(); 
        
                   assert_eq!(total_selectors(full_plan), 3_000); 
        
                   assert_eq!(total_selectors(split_plan), 1_000); 
        
                   let full_reserved = files[0] 
        
                       .extensions 
        
                       .get::<DvAccessPlanReservation>() 
        
                       .unwrap() 
        
                       .0 
        
                       .size(); 
        
                   let split_reserved = files[1] 
        
                       .extensions 
        
                       .get::<DvAccessPlanReservation>() 
        
                       .unwrap() 
        
                       .0 
        
                       .size(); 
        
                   // Keep the existing reader-clone and page-index inflation allowance. Only the actual 
        
                   // DV selector count changes, not the proof or the whole-file admission bound. 
        
                   let page_bound = page_selection_bound_selectors(&metadata).unwrap(); 
        
                   assert_eq!( 
        
                       split_reserved, 
        
                       reader_peak_bytes((1_000 + page_bound).min(3_000), 3).unwrap() 
        
                   ); 
        
                   assert!(split_reserved < full_reserved); 
        
                   assert_eq!(pool.reserved(), full_reserved + split_reserved); 
        
                   drop(files); 
        
                   assert_eq!(pool.reserved(), 0); 
        
               }

After that, #6945 (closes #5656) stopped native-format decoding from expanding the bitmap row by row, which makes each decode cheaper.

Neither change removes the repeated deletion vector work. attach_access_plan still fetches the blob, checks its CRC, deserializes the full bitmap and validates its cardinality once per split:

datafusion-comet/native/core/src/execution/delta_dv.rs

Lines 646 to 678 in 967d09b

    
           let data: Vec<u8> = if let Some(inline) = dv.inline_data { 
        
               check_inline_payload_size(&file_path, &inline, dv.size_in_bytes)?; 
        
               inline 
        
           } else if let Some(dv_path) = &dv.absolute_path { 
        
               let offset = dv 
        
                   .offset 
        
                   .ok_or_else(|| GeneralError("On-disk deletion vector missing offset".into()))?; 
        
               if offset < 0 { 
        
                   return Err(GeneralError(format!( 
        
                       "Deletion vector for {file_path} has negative offset {offset}" 
        
                   ))); 
        
               } 
        
               let offset = offset as u64; 
        
               // [i32 BE size][data: size_in_bytes][i32 BE crc] 
        
               let framed_len = 4 + dv.size_in_bytes as u64 + 4; 
        
               let (store, dv_store_path) = dv_store.ok_or_else(|| { 
        
                   GeneralError(format!( 
        
                       "Deletion vector for {file_path} has an absolute path but no pre-resolved object store" 
        
                   )) 
        
               })?; 
        
               let blob = store 
        
                   .get_range(&dv_store_path, offset..offset + framed_len) 
        
                   .await 
        
                   .map_err(|e| GeneralError(format!("Failed to read deletion vector {dv_path}: {e}")))?; 
        
               unframe_dv_blob(&blob, dv.size_in_bytes as usize)?.to_vec() 
        
           } else { 
        
               return Err(GeneralError( 
        
                   "Deletion vector descriptor has neither inline data nor a path".into(), 
        
               )); 
        
           }; 
        
           let deleted = deserialize_dv_bitmap(&data) 
        
               .map_err(|e| GeneralError(format!("Invalid deletion vector for {file_path}: {e}")))?; 
        
           validate_cardinality(&file_path, dv.cardinality, &deleted)?;

So the issue now covers decoding the DV once and sharing it across a file's splits, plus a split-file benchmark to show whether that is worth doing.

take

Navigation Menu

Repository navigation

Decode a deletion vector once across the splits of a DV-backed file in the native Delta scan #5655

What is the problem the feature request solves?

Describe the potential solution

Additional context

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	/// Translate deleted row indexes into a [`ParquetAccessPlan`]: fully-deleted
	/// row groups become `Skip`, untouched groups stay `Scan`, and partially
	/// deleted groups get a `RowSelection` selecting the complement of the deleted
	/// rows. Page-index pruning later INTERSECTS with these selections, so DV
	/// skips and page skips compose. When a file range is present, only its row groups acquire
	/// selectors; skipped groups still advance the file-global deletion-vector positions.
	pub fn build_access_plan(
	metadata: &ParquetMetaData,
	deleted: &RoaringTreemap,
	file_range: Option<&FileRange>,
	) -> Result<ParquetAccessPlan, ExecutionError> {
	let mut plan = ParquetAccessPlan::new_all(metadata.num_row_groups());
	// Empty bitmaps create no selectors to avoid. Let the reader apply its ordinary range
	// pruning instead of doing an extra page-offset pass for this case.
	let file_range = file_range.filter(\|_\| !deleted.is_empty());
	let mut deleted_iter = deleted.iter();
	let mut next_deleted = deleted_iter.next();
	let mut group_start = 0u64;
	for (idx, row_group) in metadata.row_groups().iter().enumerate() {
	let num_rows = row_group.num_rows();
	// A corrupt footer can report a negative row count. `num_rows as u64` would otherwise
	// wrap it into a huge positive value, silently corrupting every row-group boundary
	// computed from `group_start`/`group_end` below (and therefore which deleted row indexes
	// land in which row group) instead of failing loudly.
	if num_rows < 0 {
	return Err(GeneralError(format!(
	"Parquet footer reports a negative row count ({num_rows}) for row group {idx}"
	)));
	}
	let num_rows = num_rows as u64;
	let group_end = group_start.checked_add(num_rows).ok_or_else(\|\| {
	GeneralError(format!(
	"Parquet footer row counts overflow at row group {idx} ({group_start} + {num_rows})"
	))
	})?;
	// Match DataFusion's prune_by_range: dictionary-page start takes precedence over
	// data-page start, and FileRange is half-open. Keep an entry for every row group and
	// advance file-global row positions even when this split cannot read the group.
	if file_range.is_some_and(\|range\| {
	let column = row_group.column(0);
	let offset = column
	.dictionary_page_offset()
	.unwrap_or_else(\|\| column.data_page_offset());
	!range.contains(offset)
	}) {
	plan.skip(idx);
	group_start = group_end;
	continue;
	}

	// Seeking skips deletion positions from preceding, excluded groups without visiting
	// each one. The pinned Roaring implementation also handles absent 32-bit containers.
	if next_deleted.is_some_and(\|row\| row < group_start) {
	deleted_iter.advance_to(group_start);
	next_deleted = deleted_iter.next();
	}
	let mut selectors: Vec<RowSelector> = Vec::new();
	let mut cursor = group_start;
	let mut deleted_in_group = 0u64;
	while let Some(row) = next_deleted.filter(\|row\| *row < group_end) {
	next_deleted = deleted_iter.next();
	deleted_in_group += 1;
	if row > cursor {
	selectors.push(RowSelector::select((row - cursor) as usize));
	}
	// Merge runs of consecutive deleted rows into one skip.
	match selectors.last_mut() {
	Some(last) if last.skip => last.row_count += 1,
	_ => selectors.push(RowSelector::skip(1)),
	}
	cursor = row + 1;
	}
	if deleted_in_group == num_rows && num_rows > 0 {
	plan.skip(idx);
	} else if deleted_in_group > 0 {
	if group_end > cursor {
	selectors.push(RowSelector::select((group_end - cursor) as usize));
	}
	plan.scan_selection(idx, RowSelection::from(selectors));
	}
	group_start = group_end;
	}
	// Validate against the whole file even when this split excludes the final groups.
	// A deleted index beyond the file's total row count means the DV does not belong to this
	// file. Checking the maximum also catches corruption beyond an absent Roaring container.
	if let Some(row) = deleted.max().filter(\|row\| *row >= group_start) {
	return Err(GeneralError(format!(
	"Deletion vector marks row {row} but the file only has {group_start} rows"
	)));
	}
	Ok(plan)
	}

	#[test]
	fn split_access_plan_preserves_file_global_positions() {
	let metadata =
	metadata_with_page_offsets(&[3, 3, 3], &[(10, None), (20, None), (30, None)]);
	let plan = build_access_plan(
	&metadata,
	&RoaringTreemap::from_iter([1, 4, 7]),
	Some(&FileRange { start: 20, end: 21 }),
	)
	.unwrap();
	assert_eq!(plan.inner()[0], RowGroupAccess::Skip);
	assert_eq!(plan.inner()[2], RowGroupAccess::Skip);
	assert_eq!(
	plan.inner()[1],
	RowGroupAccess::Selection(RowSelection::from(vec![
	RowSelector::select(1),
	RowSelector::skip(1),
	RowSelector::select(1),
	]))
	);
	}

	#[test]
	fn split_access_plan_uses_half_open_dictionary_page_boundaries() {
	let metadata =
	metadata_with_page_offsets(&[3, 3, 3], &[(20, Some(10)), (25, Some(20)), (30, None)]);
	let plan = build_access_plan(
	&metadata,
	&RoaringTreemap::from_iter([1]),
	Some(&FileRange { start: 20, end: 30 }),
	)
	.unwrap();
	assert_eq!(
	plan.inner(),
	&[
	RowGroupAccess::Skip,
	RowGroupAccess::Scan,
	RowGroupAccess::Skip
	]
	);
	}

	#[test]
	fn split_access_plan_validates_excluded_rows_and_row_groups() {
	let metadata =
	metadata_with_page_offsets(&[3, 3, 3], &[(10, None), (20, None), (30, None)]);
	for range in [
	FileRange { start: 20, end: 21 },
	FileRange { start: 40, end: 50 },
	] {
	let err =
	build_access_plan(&metadata, &RoaringTreemap::from_iter([1, 9]), Some(&range))
	.unwrap_err();
	assert!(err.to_string().contains("only has 9 rows"));
	}
	// Metadata must be valid even when the split excludes the corrupt row group.
	let metadata = metadata_with_page_offsets(&[3, -1], &[(10, None), (20, None)]);
	assert!(build_access_plan(
	&metadata,
	&RoaringTreemap::new(),
	Some(&FileRange { start: 10, end: 11 }),
	)
	.is_err());
	let metadata = metadata_with_page_offsets(
	&[i64::MAX, i64::MAX, i64::MAX],
	&[(10, None), (20, None), (30, None)],
	);
	assert!(build_access_plan(
	&metadata,
	&RoaringTreemap::new(),
	Some(&FileRange { start: 10, end: 11 }),
	)
	.is_err());
	}

	#[test]
	fn split_access_plan_seeks_across_present_and_absent_roaring_containers() {
	for (first_group_rows, second_group_rows, marked_row) in [
	((1_i64 << 32) + 3, 3, (1_u64 << 32) + 4),
	((1_i64 << 32) + 100, 1_i64 << 32, (2_u64 << 32) + 50),
	] {
	let metadata = metadata_with_page_offsets(
	&[first_group_rows, second_group_rows],
	&[(10, None), (20, None)],
	);
	let plan = build_access_plan(
	&metadata,
	&RoaringTreemap::from_iter([1, 2, marked_row]),
	Some(&FileRange { start: 20, end: 21 }),
	)
	.unwrap();
	assert_eq!(plan.inner()[0], RowGroupAccess::Skip);
	let RowGroupAccess::Selection(selection) = &plan.inner()[1] else {
	panic!("expected a selection in the second row group");
	};
	assert_eq!(selection.row_count(), second_group_rows as usize - 1);
	assert_eq!(selection.skipped_row_count(), 1);
	}

	// A corrupt position beyond an absent high-32-bit container remains an error.
	let metadata = metadata_with_page_offsets(&[(1_i64 << 32) + 100], &[(10, None)]);
	assert!(build_access_plan(
	&metadata,
	&RoaringTreemap::from_iter([1, (2_u64 << 32) + 50]),
	Some(&FileRange { start: 20, end: 21 }),
	)
	.is_err());
	}

	#[test]
	fn split_access_plan_has_no_selectors_for_excluded_groups() {
	let metadata = metadata_with_page_offsets(
	&[1_000, 1_000, 1_000],
	&[(10, None), (20, None), (30, None)],
	);
	let deleted = (0..1_000).chain(2_000..3_000).step_by(2).collect();
	let plan = build_access_plan(&metadata, &deleted, Some(&FileRange { start: 20, end: 21 }))
	.unwrap();
	assert_eq!(total_selectors(&plan), 0);
	assert_eq!(plan.inner()[1], RowGroupAccess::Scan);
	assert!(total_selectors(&build_access_plan(&metadata, &deleted, None).unwrap()) > 0);
	}

	/// The confirmed worst case: deleting every even row leaves
	/// no adjacent skips or selects to merge, so `build_access_plan` emits
	/// one non-coalescing `RowSelector` per row of the group.
	fn alternating_deleted(num_rows: u64) -> RoaringTreemap {
	(0..num_rows).step_by(2).collect()
	}

	#[test]
	fn total_selectors_counts_one_per_row_for_alternating_bitmap() {
	let deleted = alternating_deleted(1024);
	let plan = full_file_access_plan(&[1024], &deleted).unwrap();
	assert_eq!(total_selectors(&plan), 1024);
	}

	#[test]
	fn total_selectors_ignores_scan_and_skip_row_groups() {
	// Group 0 untouched (Scan), group 1 fully deleted (Skip): neither
	// carries a RowSelection, so both must contribute zero selectors.
	let deleted: RoaringTreemap = (10u64..20).collect();
	let plan = full_file_access_plan(&[10, 10], &deleted).unwrap();
	assert_eq!(total_selectors(&plan), 0);
	}

	/// Writes one file's on-disk parquet data for a full-file, alternating-bitmap deletion
	/// vector, returning its path, byte size, and deleted-row bitmap so callers needing the
	/// file's on-disk metadata (to size a memory pool exactly, or to replay the real reader
	/// path) can inspect it before building a [`DvScanFile`] from it.
	fn write_alternating_parquet(
	dir: &std::path::Path,
	num_rows: i64,
	) -> (std::path::PathBuf, i64, RoaringTreemap) {
	let deleted = alternating_deleted(num_rows as u64);
	let path = dir.join("alternating.parquet");
	let size = write_parquet(&path, num_rows);
	(path, size, deleted)
	}

	/// Builds a [`DvScanFile`] with an inline deletion vector for an already-written parquet
	/// file at `path`.
	fn dv_scan_file_for_alternating(
	runtime_env: &Arc<RuntimeEnv>,
	path: &std::path::Path,
	size: i64,
	deleted: &RoaringTreemap,
	) -> DvScanFile {
	let inline_data = portable_bytes(deleted);
	let file_path = format!("file://{}", path.display());
	let (data_store, _) = resolve_store(runtime_env, &file_path);
	DvScanFile {
	file: PartitionedFile::new(path.display().to_string(), size as u64),
	file_path,
	dv: Some(DeltaSparkDvDescriptor {
	storage_type: "i".to_string(),
	absolute_path: None,
	inline_data: Some(inline_data.clone()),
	offset: None,
	size_in_bytes: inline_data.len() as i32,
	cardinality: deleted.len() as i64,
	}),
	data_store,
	dv_store: None,
	}
	}

	#[tokio::test]
	async fn split_access_plan_reservation_tracks_only_retained_selectors() {
	let tmp = tempfile::tempdir().unwrap();
	let path = tmp.path().join("split.parquet");
	let size = write_parquet_with_properties(
	&path,
	3_000,
	WriterProperties::builder()
	.set_max_row_group_row_count(Some(1_000))
	.build(),
	);
	let metadata = read_metadata_with_page_index(&path);
	let column = metadata.row_group(1).column(0);
	let offset = column
	.dictionary_page_offset()
	.unwrap_or_else(\|\| column.data_page_offset());
	let pool: Arc<dyn MemoryPool> = Arc::new(GreedyMemoryPool::new(10_000_000));
	let runtime_env = RuntimeEnvBuilder::new()
	.with_memory_pool(Arc::clone(&pool))
	.build_arc()
	.unwrap();
	let deleted = alternating_deleted(3_000);
	let full = dv_scan_file_for_alternating(&runtime_env, &path, size, &deleted);
	let mut split = dv_scan_file_for_alternating(&runtime_env, &path, size, &deleted);
	split.file.range = Some(FileRange {
	start: offset,
	end: offset + 1,
	});
	let files = attach_access_plans(runtime_env, vec![full, split])
	.await
	.unwrap();
	let full_plan = files[0].extensions.get::<ParquetAccessPlan>().unwrap();
	let split_plan = files[1].extensions.get::<ParquetAccessPlan>().unwrap();
	assert_eq!(total_selectors(full_plan), 3_000);
	assert_eq!(total_selectors(split_plan), 1_000);
	let full_reserved = files[0]
	.extensions
	.get::<DvAccessPlanReservation>()
	.unwrap()
	.0
	.size();
	let split_reserved = files[1]
	.extensions
	.get::<DvAccessPlanReservation>()
	.unwrap()
	.0
	.size();
	// Keep the existing reader-clone and page-index inflation allowance. Only the actual
	// DV selector count changes, not the proof or the whole-file admission bound.
	let page_bound = page_selection_bound_selectors(&metadata).unwrap();
	assert_eq!(
	split_reserved,
	reader_peak_bytes((1_000 + page_bound).min(3_000), 3).unwrap()
	);
	assert!(split_reserved < full_reserved);
	assert_eq!(pool.reserved(), full_reserved + split_reserved);
	drop(files);
	assert_eq!(pool.reserved(), 0);
	}

	let data: Vec<u8> = if let Some(inline) = dv.inline_data {
	check_inline_payload_size(&file_path, &inline, dv.size_in_bytes)?;
	inline
	} else if let Some(dv_path) = &dv.absolute_path {
	let offset = dv
	.offset
	.ok_or_else(\|\| GeneralError("On-disk deletion vector missing offset".into()))?;
	if offset < 0 {
	return Err(GeneralError(format!(
	"Deletion vector for {file_path} has negative offset {offset}"
	)));
	}
	let offset = offset as u64;
	// [i32 BE size][data: size_in_bytes][i32 BE crc]
	let framed_len = 4 + dv.size_in_bytes as u64 + 4;
	let (store, dv_store_path) = dv_store.ok_or_else(\|\| {
	GeneralError(format!(
	"Deletion vector for {file_path} has an absolute path but no pre-resolved object store"
	))
	})?;
	let blob = store
	.get_range(&dv_store_path, offset..offset + framed_len)
	.await
	.map_err(\|e\| GeneralError(format!("Failed to read deletion vector {dv_path}: {e}")))?;
	unframe_dv_blob(&blob, dv.size_in_bytes as usize)?.to_vec()
	} else {
	return Err(GeneralError(
	"Deletion vector descriptor has neither inline data nor a path".into(),
	));
	};
	let deleted = deserialize_dv_bitmap(&data)
	.map_err(\|e\| GeneralError(format!("Invalid deletion vector for {file_path}: {e}")))?;
	validate_cardinality(&file_path, dv.cardinality, &deleted)?;

Decode a deletion vector once across the splits of a DV-backed file in the native Delta scan #5655

Description

What is the problem the feature request solves?

Describe the potential solution

Additional context

Activity

dwsmith1983 commented on Sep 3, 2026

dwsmith1983 commented on Sep 23, 2026

dwsmith1983 commented on Oct 28, 2026

dwsmith1983 commented on Oct 28, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions