Skip to content
Merged
Show file tree
Hide file tree
Changes from 12 commits
Commits
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
5 changes: 4 additions & 1 deletion crates/iceberg/public-api.txt
Original file line number Diff line number Diff line change
Expand Up @@ -1310,12 +1310,15 @@ pub fn iceberg::scan::FileScanTask::serialize<__S>(&self, __serializer: __S) ->
impl<'de> serde_core::de::Deserialize<'de> for iceberg::scan::FileScanTask
pub fn iceberg::scan::FileScanTask::deserialize<__D>(__deserializer: __D) -> core::result::Result<Self, <__D as serde_core::de::Deserializer>::Error> where __D: serde_core::de::Deserializer<'de>
pub struct iceberg::scan::FileScanTaskDeleteFile
pub iceberg::scan::FileScanTaskDeleteFile::content_offset: core::option::Option<i64>
pub iceberg::scan::FileScanTaskDeleteFile::content_size_in_bytes: core::option::Option<i64>
pub iceberg::scan::FileScanTaskDeleteFile::equality_ids: core::option::Option<alloc::vec::Vec<i32>>
pub iceberg::scan::FileScanTaskDeleteFile::file_path: alloc::string::String
pub iceberg::scan::FileScanTaskDeleteFile::file_size_in_bytes: u64
pub iceberg::scan::FileScanTaskDeleteFile::file_type: iceberg::spec::DataContentType
pub iceberg::scan::FileScanTaskDeleteFile::key_metadata: core::option::Option<alloc::boxed::Box<[u8]>>
pub iceberg::scan::FileScanTaskDeleteFile::partition_spec_id: i32
pub iceberg::scan::FileScanTaskDeleteFile::referenced_data_file: core::option::Option<alloc::string::String>
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
Expand All @@ -1324,7 +1327,7 @@ 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
impl iceberg::scan::FileScanTaskDeleteFile
pub fn iceberg::scan::FileScanTaskDeleteFile::builder() -> FileScanTaskDeleteFileBuilder<((), (), (), (), (), ())>
pub fn iceberg::scan::FileScanTaskDeleteFile::builder() -> FileScanTaskDeleteFileBuilder<((), (), (), (), (), (), (), (), ())>
impl serde_core::ser::Serialize for iceberg::scan::FileScanTaskDeleteFile
pub fn iceberg::scan::FileScanTaskDeleteFile::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer
impl<'de> serde_core::de::Deserialize<'de> for iceberg::scan::FileScanTaskDeleteFile
Expand Down
6 changes: 6 additions & 0 deletions crates/iceberg/src/arrow/delete_file_loader.rs
Original file line number Diff line number Diff line change
Expand Up @@ -237,6 +237,9 @@ mod tests {
partition_spec_id: 0,
equality_ids: None,
key_metadata: Some(Box::from(key_metadata.as_ref())),
referenced_data_file: None,
content_offset: None,
content_size_in_bytes: None,
};

let scan_metrics = ScanMetrics::new();
Expand Down Expand Up @@ -311,6 +314,9 @@ mod tests {
partition_spec_id: 0,
equality_ids: Some(vec![1]),
key_metadata: Some(Box::from(key_metadata.as_ref())),
referenced_data_file: None,
content_offset: None,
content_size_in_bytes: None,
};

let scan_metrics = ScanMetrics::new();
Expand Down
3 changes: 3 additions & 0 deletions crates/iceberg/src/arrow/reader/row_filter.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1246,6 +1246,9 @@ mod tests {
partition_spec_id: 0,
equality_ids: None,
file_size_in_bytes: std::fs::metadata(&pos_del_path).unwrap().len(),
referenced_data_file: None,
content_offset: None,
content_size_in_bytes: None,
key_metadata: None,
}],
partition: None,
Expand Down
35 changes: 35 additions & 0 deletions crates/iceberg/src/delete_file_index.rs
Original file line number Diff line number Diff line change
Expand Up @@ -416,6 +416,41 @@ mod tests {
assert!(actual_paths_to_apply_for_different_spec.is_empty());
}

#[test]
fn test_deletion_vector_context_carries_coordinates() {
// A deletion vector is a PositionDeletes entry stored as a Puffin blob, located by
// content_offset / content_size_in_bytes and scoped by referenced_data_file. Those
// three fields must survive the conversion into a FileScanTaskDeleteFile so the loader
// can find and apply the blob.
let dv = DataFileBuilder::default()
.file_path("s3://bucket/data/part-0.parquet-deletes.puffin".to_string())
.file_format(DataFileFormat::Puffin)
.content(DataContentType::PositionDeletes)
.record_count(3)
.referenced_data_file(Some("s3://bucket/data/part-0.parquet".to_string()))
.content_offset(Some(4))
.content_size_in_bytes(Some(40))
.partition(Struct::empty())
.partition_spec_id(0)
.file_size_in_bytes(44)
.build()
.unwrap();

let ctx = DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &dv).into(),
partition_spec_id: 0,
};

let task: FileScanTaskDeleteFile = (&ctx).into();
assert_eq!(task.file_type, DataContentType::PositionDeletes);
assert_eq!(task.content_offset, Some(4));
assert_eq!(task.content_size_in_bytes, Some(40));
assert_eq!(
task.referenced_data_file.as_deref(),
Some("s3://bucket/data/part-0.parquet")
);
}

fn build_unpartitioned_eq_delete() -> DataFile {
build_partitioned_eq_delete(&Struct::empty(), 0)
}
Expand Down
24 changes: 24 additions & 0 deletions crates/iceberg/src/scan/task.rs
Original file line number Diff line number Diff line change
Expand Up @@ -208,6 +208,9 @@ impl From<&DeleteFileContext> for FileScanTaskDeleteFile {
.with_file_type(ctx.manifest_entry.content_type())
.with_partition_spec_id(ctx.partition_spec_id)
.with_equality_ids(ctx.manifest_entry.data_file.equality_ids.clone())
.with_referenced_data_file(ctx.manifest_entry.data_file.referenced_data_file.clone())
.with_content_offset(ctx.manifest_entry.data_file.content_offset)
.with_content_size_in_bytes(ctx.manifest_entry.data_file.content_size_in_bytes)
.with_key_metadata(
ctx.manifest_entry
.data_file
Expand Down Expand Up @@ -239,6 +242,27 @@ pub struct FileScanTaskDeleteFile {
#[builder(default)]
pub equality_ids: Option<Vec<i32>>,

/// For a deletion vector, the location of the data file whose rows it deletes. Required for
/// deletion vectors, and may also be set on a position delete file scoped to one data file.
#[serde(default)]
#[serde(skip_serializing_if = "Option::is_none")]
#[builder(default)]
pub referenced_data_file: Option<String>,

/// For a deletion vector, the offset of the blob within its Puffin file. Set only for
/// deletion vectors, where it locates the blob for direct access.
#[serde(default)]
#[serde(skip_serializing_if = "Option::is_none")]
#[builder(default)]
pub content_offset: Option<i64>,

/// For a deletion vector, the length in bytes of the blob within its Puffin file. Set
/// whenever `content_offset` is.

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this is a bit unclear, I'd recommend

Suggested change
/// For a deletion vector, the length in bytes of the blob within its Puffin file. Set
/// whenever `content_offset` is.
/// For a deletion vector, the length in bytes of the blob within its Puffin file.
/// Required together with `content_offset`; both are absent for non-DV delete files.

also I noticed in #3035 , we are checking if a file is DV by checking if content_offset.is_some(). I believe instead of that, we should check isDv = (if type == PosDel && format == puffin). and we should throw error if (isDV && (content_offset.is_none() || content_size_in_bytes.is_none())) because both of them are required for DVs

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Took your wording verbatim.

Agreed on the DV check: the rule should be content == POSITION_DELETES && format == PUFFIN, and a DV missing content_offset or content_size_in_bytes should error instead of falling through to the position-delete path.

FileScanTaskDeleteFile has no format field today, only file_type, which is why #3035 keys on content_offset.is_some() in the loader. DeleteFileIndex works on DataFile, which has file_format(), so it can use the rule as written.

Rather than land an unused field here, I'll do this in #3035: add file_format to FileScanTaskDeleteFile, switch both the index and the loader to the type + format check, and error when a DV is missing either coordinate.

Let me know if you'd rather the field land in this PR.

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Adding file_format to FileScanTaskDeleteFile sounds good to me, we can do it in the next PR

#[serde(default)]
#[serde(skip_serializing_if = "Option::is_none")]
#[builder(default)]
pub content_size_in_bytes: Option<i64>,

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In #3035 I noticed we are also trying to add record_count for cardinality check. I'm leaning toward adding it here

@mbutrovich mbutrovich Aug 26, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added record_count here, populated from the manifest entry, which leaves #3035 as pure behavior.

It stays Option<u64> because a task isn't always built from a manifest entry. A deletion vector always is, so #3035 will error on a missing record count instead of skipping the cardinality check.

/// Key metadata for encrypted delete files (Parquet Modular Encryption).
/// When present, the reader uses this to build `FileDecryptionProperties`.
///
Expand Down
Loading