Skip to content

feat(scan): [4/N] address followups for reading and applying V3 deletion vectors #3180

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
mbutrovich wants to merge 2 commits into
base: main
Choose a base branch
from
Draft

feat(scan): [4/N] address followups for reading and applying V3 deletion vectors #3180

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1940c51
address followup comments from #3035
Sep 9, 2026
b738363
fix comment
Sep 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions crates/iceberg/public-api.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1325,6 +1325,8 @@ impl core::clone::Clone for iceberg::scan::FileScanTaskDeleteFile
pub fn iceberg::scan::FileScanTaskDeleteFile::clone(&self) -> iceberg::scan::FileScanTaskDeleteFile
impl core::cmp::PartialEq for iceberg::scan::FileScanTaskDeleteFile
pub fn iceberg::scan::FileScanTaskDeleteFile::eq(&self, other: &iceberg::scan::FileScanTaskDeleteFile) -> bool
impl core::convert::From<iceberg::scan::FileScanTaskDeleteFile> for iceberg::Result<iceberg::scan::FileScanTaskDeleteFile>
pub fn iceberg::Result<iceberg::scan::FileScanTaskDeleteFile>::from(task: iceberg::scan::FileScanTaskDeleteFile) -> Self
impl core::fmt::Debug for iceberg::scan::FileScanTaskDeleteFile
pub fn iceberg::scan::FileScanTaskDeleteFile::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
impl core::marker::StructuralPartialEq for iceberg::scan::FileScanTaskDeleteFile
Expand Down
176 changes: 12 additions & 164 deletions crates/iceberg/src/arrow/caching_delete_file_loader.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,9 +35,9 @@ use crate::io::FileIO;
use crate::runtime::Runtime;
use crate::scan::{ArrowRecordBatchStream, FileScanTaskDeleteFile};
use crate::spec::{
DataContentType, DataFileFormat, Datum, ListType, MapType, NestedField, NestedFieldRef,
PartnerAccessor, PrimitiveType, Schema, SchemaRef, SchemaWithPartnerVisitor, StructType, Type,
VariantType, visit_schema_with_partner,
DataContentType, Datum, ListType, MapType, NestedField, NestedFieldRef, PartnerAccessor,
PrimitiveType, Schema, SchemaRef, SchemaWithPartnerVisitor, StructType, Type, VariantType,
visit_schema_with_partner,
};
use crate::{Error, ErrorKind, Result};

Expand Down Expand Up @@ -266,7 +266,7 @@ impl CachingDeleteFileLoader {
DataContentType::PositionDeletes => {
// A V3 deletion vector arrives as a PositionDeletes entry whose deletes live in
// a Puffin blob, not in a positional-delete parquet file.
if task.file_format == DataFileFormat::Puffin {
if task.is_deletion_vector() {
return Self::load_deletion_vector(task, basic_delete_file_loader).await;
}

Expand Down Expand Up @@ -330,79 +330,6 @@ impl CachingDeleteFileLoader {
}
}

/// Validates a deletion-vector task and returns what the read needs as typed values:
/// `(start, len, referenced data file path, expected cardinality)`.
///
/// The spec requires `referenced_data_file`, `content_offset` and `content_size_in_bytes` on
/// a deletion vector, and a deletion vector is always built from a manifest entry, so it
/// always carries `record_count`. A missing one is a manifest-entry inconsistency rather
/// than an I/O failure.
///
/// Equality and ordinary position deletes have no equivalent validation in this loader: a
/// malformed equality/position delete file fails loudly when the Parquet reader can't open
/// it. A deletion vector's coordinates instead drive a raw byte-range read with no format
/// to fail against, so a bad coordinate would otherwise decode silently into the wrong (or
/// no) deletes, per the same corrupted-blob concern Iceberg-Java validates in
/// `BitmapPositionDeleteIndex.deserializeBitmap`.
fn validate_deletion_vector_task(
task: &FileScanTaskDeleteFile,
) -> Result<(u64, u64, String, u64)> {
let content_offset = task.content_offset.ok_or_else(|| {
Error::new(
ErrorKind::DataInvalid,
format!(
"deletion vector {} is missing content_offset",
task.file_path
),
)
})?;
let content_size = task.content_size_in_bytes.ok_or_else(|| {
Error::new(
ErrorKind::DataInvalid,
format!(
"deletion vector {} is missing content_size_in_bytes",
task.file_path
),
)
})?;
let data_file_path = task.referenced_data_file.clone().ok_or_else(|| {
Error::new(
ErrorKind::DataInvalid,
format!(
"deletion vector {} is missing referenced_data_file",
task.file_path
),
)
})?;
let record_count = task.record_count.ok_or_else(|| {
Error::new(
ErrorKind::DataInvalid,
format!("deletion vector {} is missing record_count", task.file_path),
)
})?;

let start = u64::try_from(content_offset).map_err(|_| {
Error::new(
ErrorKind::DataInvalid,
format!(
"deletion vector {} has negative content_offset {content_offset}",
task.file_path
),
)
})?;
let len = u64::try_from(content_size).map_err(|_| {
Error::new(
ErrorKind::DataInvalid,
format!(
"deletion vector {} has negative content_size_in_bytes {content_size}",
task.file_path
),
)
})?;

Ok((start, len, data_file_path, record_count))
}

/// Validates a decoded deletion vector's cardinality against the manifest entry's
/// `record_count`, mirroring Iceberg-Java's `BitmapPositionDeleteIndex.deserializeBitmap`.
fn validate_deletion_vector_cardinality(
Expand Down Expand Up @@ -436,7 +363,8 @@ impl CachingDeleteFileLoader {
task: &FileScanTaskDeleteFile,
basic_delete_file_loader: BasicDeleteFileLoader,
) -> Result<DeleteFileContext> {
let (start, len, data_file_path, record_count) = Self::validate_deletion_vector_task(task)?;
let (start, len, data_file_path, record_count) = task.deletion_vector_coordinates()?;
let data_file_path = data_file_path.to_string();

let input_file = basic_delete_file_loader
.file_io()
Expand Down Expand Up @@ -875,7 +803,7 @@ mod tests {
use super::*;
use crate::arrow::delete_filter::tests::setup;
use crate::scan::FileScanTaskDeleteFile;
use crate::spec::{DataContentType, Schema};
use crate::spec::{DataContentType, DataFileFormat, Schema};
use crate::test_utils::encode_dv_blob;

#[tokio::test]
Expand Down Expand Up @@ -1426,7 +1354,8 @@ mod tests {
.with_file_type(DataContentType::PositionDeletes)
.with_file_format(DataFileFormat::Parquet)
.with_partition_spec_id(0)
.build();
.build()
.unwrap();

let eq_del = FileScanTaskDeleteFile::builder()
.with_file_path(eq_delete_path.clone())
Expand All @@ -1435,7 +1364,8 @@ mod tests {
.with_file_format(DataFileFormat::Parquet)
.with_partition_spec_id(0)
.with_equality_ids(Some(vec![2, 3])) // Only use field IDs that exist in both schemas
.build();
.build()
.unwrap();

let file_scan_task = FileScanTask::builder()
.with_file_size_in_bytes(0)
Expand Down Expand Up @@ -1585,6 +1515,7 @@ mod tests {
.with_record_count(Some(record_count))
.with_key_metadata(key_metadata)
.build()
.unwrap()
}

#[tokio::test]
Expand Down Expand Up @@ -1731,89 +1662,6 @@ mod tests {
assert!(err.message().contains("expected 2 from record_count"));
}

// A well-formed deletion-vector task, for tests that then clear or corrupt one field.
fn valid_dv_task() -> FileScanTaskDeleteFile {
dv_task(
"deletes.puffin".to_string(),
100,
"data.parquet".to_string(),
4,
40,
2,
None,
)
}

#[test]
fn test_validate_deletion_vector_task_rejects_missing_content_offset() {
let mut task = valid_dv_task();
task.content_offset = None;

let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task).unwrap_err();
assert_eq!(err.kind(), ErrorKind::DataInvalid);
assert!(err.message().contains("missing content_offset"));
}

#[test]
fn test_validate_deletion_vector_task_rejects_missing_content_size() {
let mut task = valid_dv_task();
task.content_size_in_bytes = None;

let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task).unwrap_err();
assert_eq!(err.kind(), ErrorKind::DataInvalid);
assert!(err.message().contains("missing content_size_in_bytes"));
}

#[test]
fn test_validate_deletion_vector_task_rejects_missing_referenced_data_file() {
let mut task = valid_dv_task();
task.referenced_data_file = None;

let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task).unwrap_err();
assert_eq!(err.kind(), ErrorKind::DataInvalid);
assert!(err.message().contains("missing referenced_data_file"));
}

#[test]
fn test_validate_deletion_vector_task_rejects_missing_record_count() {
let mut task = valid_dv_task();
task.record_count = None;

let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task).unwrap_err();
assert_eq!(err.kind(), ErrorKind::DataInvalid);
assert!(err.message().contains("missing record_count"));
}

#[test]
fn test_validate_deletion_vector_task_rejects_negative_content_offset() {
let mut task = valid_dv_task();
task.content_offset = Some(-1);

let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task).unwrap_err();
assert_eq!(err.kind(), ErrorKind::DataInvalid);
assert!(err.message().contains("negative content_offset"));
}

#[test]
fn test_validate_deletion_vector_task_rejects_negative_content_size() {
let mut task = valid_dv_task();
task.content_size_in_bytes = Some(-1);

let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task).unwrap_err();
assert_eq!(err.kind(), ErrorKind::DataInvalid);
assert!(err.message().contains("negative content_size_in_bytes"));
}

#[test]
fn test_validate_deletion_vector_task_accepts_valid_coordinates() {
let (start, len, data_file_path, record_count) =
CachingDeleteFileLoader::validate_deletion_vector_task(&valid_dv_task()).unwrap();
assert_eq!(start, 4);
assert_eq!(len, 40);
assert_eq!(data_file_path, "data.parquet");
assert_eq!(record_count, 2);
}

#[test]
fn test_validate_deletion_vector_cardinality_accepts_matching_count() {
let mut dv = DeleteVector::default();
Expand Down
12 changes: 8 additions & 4 deletions crates/iceberg/src/arrow/delete_filter.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -443,7 +443,8 @@ pub(crate) mod tests {
.with_file_type(DataContentType::PositionDeletes)
.with_file_format(DataFileFormat::Parquet)
.with_partition_spec_id(0)
.build();
.build()
.unwrap();

let pos_del_2 = FileScanTaskDeleteFile::builder()
.with_file_path(format!(
Expand All @@ -461,7 +462,8 @@ pub(crate) mod tests {
.with_file_type(DataContentType::PositionDeletes)
.with_file_format(DataFileFormat::Parquet)
.with_partition_spec_id(0)
.build();
.build()
.unwrap();

let pos_del_3 = FileScanTaskDeleteFile::builder()
.with_file_path(format!(
Expand All @@ -479,7 +481,8 @@ pub(crate) mod tests {
.with_file_type(DataContentType::PositionDeletes)
.with_file_format(DataFileFormat::Parquet)
.with_partition_spec_id(0)
.build();
.build()
.unwrap();

let file_scan_tasks = vec![
FileScanTask::builder()
Expand Down Expand Up @@ -556,7 +559,8 @@ pub(crate) mod tests {
.with_file_type(DataContentType::EqualityDeletes)
.with_file_format(DataFileFormat::Parquet)
.with_partition_spec_id(0)
.build(),
.build()
.unwrap(),
])
.with_case_sensitive(true)
.build()
Expand Down
3 changes: 2 additions & 1 deletion crates/iceberg/src/arrow/reader/pipeline.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2414,7 +2414,8 @@ mod tests {
.with_file_size_in_bytes(std::fs::metadata(&del_path).unwrap().len())
.with_file_type(DataContentType::PositionDeletes)
.with_partition_spec_id(0)
.build();
.build()
.unwrap();
let task = row_id_task_with_options(data_path, Some(100), 0, 0, vec![delete]);

let reader = ArrowReaderBuilder::new(FileIO::new_with_fs(), Runtime::current()).build();
Expand Down
15 changes: 10 additions & 5 deletions crates/iceberg/src/arrow/reader/positional_deletes.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -452,7 +452,8 @@ mod tests {
.with_file_type(DataContentType::PositionDeletes)
.with_file_format(DataFileFormat::Parquet)
.with_partition_spec_id(0)
.build(),
.build()
.unwrap(),
])
.with_case_sensitive(false)
.build()
Expand Down Expand Up @@ -672,7 +673,8 @@ mod tests {
.with_file_type(DataContentType::PositionDeletes)
.with_file_format(DataFileFormat::Parquet)
.with_partition_spec_id(0)
.build(),
.build()
.unwrap(),
])
.with_case_sensitive(false)
.build()
Expand Down Expand Up @@ -886,7 +888,8 @@ mod tests {
.with_file_type(DataContentType::PositionDeletes)
.with_file_format(DataFileFormat::Parquet)
.with_partition_spec_id(0)
.build(),
.build()
.unwrap(),
])
.with_case_sensitive(false)
.build()
Expand Down Expand Up @@ -1005,7 +1008,8 @@ mod tests {
.with_content_offset(Some(content_offset))
.with_content_size_in_bytes(Some(content_size))
.with_record_count(Some(2))
.build(),
.build()
.unwrap(),
])
.with_case_sensitive(false)
.build();
Expand Down Expand Up @@ -1102,7 +1106,8 @@ mod tests {
.with_content_offset(Some(0))
.with_content_size_in_bytes(Some(blob.len() as i64))
.with_record_count(Some(5))
.build(),
.build()
.unwrap(),
])
.with_case_sensitive(false)
.build();
Expand Down
Loading
Loading