From 06eca0e814663ad754ee30dc028185f98d250eec Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Thu, 9 Jul 2026 17:56:43 -0400 Subject: [PATCH 1/9] feat(scan): carry deletion-vector coordinates on FileScanTaskDeleteFile Add referenced_data_file, content_offset, and content_size_in_bytes to FileScanTaskDeleteFile, populated from the delete file's manifest entry. These locate a deletion-vector blob and scope it to its data file, which the delete loader needs to read and apply V3 deletion vectors. Refs #2792. --- crates/iceberg/src/arrow/reader/row_filter.rs | 3 ++ crates/iceberg/src/delete_file_index.rs | 35 +++++++++++++++++++ crates/iceberg/src/scan/task.rs | 18 ++++++++++ 3 files changed, 56 insertions(+) diff --git a/crates/iceberg/src/arrow/reader/row_filter.rs b/crates/iceberg/src/arrow/reader/row_filter.rs index a94c159d1d..a4477431c7 100644 --- a/crates/iceberg/src/arrow/reader/row_filter.rs +++ b/crates/iceberg/src/arrow/reader/row_filter.rs @@ -1241,6 +1241,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, diff --git a/crates/iceberg/src/delete_file_index.rs b/crates/iceberg/src/delete_file_index.rs index e58bb671b6..0a7f74d5bf 100644 --- a/crates/iceberg/src/delete_file_index.rs +++ b/crates/iceberg/src/delete_file_index.rs @@ -416,6 +416,41 @@ mod tests { assert!(actual_paths_to_apply_for_different_spec.is_empty()); } + #[test] + fn 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) } diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs index faeac51be9..97efd0adf1 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -174,6 +174,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 @@ -205,6 +208,21 @@ pub struct FileScanTaskDeleteFile { #[builder(default)] pub equality_ids: Option>, + /// 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. + #[builder(default)] + pub referenced_data_file: Option, + + /// 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. + #[builder(default)] + pub content_offset: Option, + + /// For a deletion vector, the length in bytes of the blob within its Puffin file. Set + /// whenever `content_offset` is. + #[builder(default)] + pub content_size_in_bytes: Option, + /// Key metadata for encrypted delete files (Parquet Modular Encryption). /// When present, the reader uses this to build `FileDecryptionProperties`. /// From fb84ffca38c14359401235e4ab6570be0cd31fd4 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Tue, 21 Jul 2026 14:21:16 -0400 Subject: [PATCH 2/9] fix after merging in main --- crates/iceberg/src/arrow/delete_file_loader.rs | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/crates/iceberg/src/arrow/delete_file_loader.rs b/crates/iceberg/src/arrow/delete_file_loader.rs index efdc2632cf..af470cf561 100644 --- a/crates/iceberg/src/arrow/delete_file_loader.rs +++ b/crates/iceberg/src/arrow/delete_file_loader.rs @@ -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(); @@ -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(); From 4e54829d1c81855e3e3596bc9123d11e798034ac Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Tue, 21 Jul 2026 14:41:56 -0400 Subject: [PATCH 3/9] fix public-api.txt --- crates/iceberg/public-api.txt | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index e6de7ab4d8..26c7a60d27 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1287,12 +1287,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::Error> where __D: serde_core::de::Deserializer<'de> pub struct iceberg::scan::FileScanTaskDeleteFile +pub iceberg::scan::FileScanTaskDeleteFile::content_offset: core::option::Option +pub iceberg::scan::FileScanTaskDeleteFile::content_size_in_bytes: core::option::Option pub iceberg::scan::FileScanTaskDeleteFile::equality_ids: core::option::Option> 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> pub iceberg::scan::FileScanTaskDeleteFile::partition_spec_id: i32 +pub iceberg::scan::FileScanTaskDeleteFile::referenced_data_file: core::option::Option 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 @@ -1301,7 +1304,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 From e17dfa3ad188c6a3bc8f2258ccb08485cc62cc92 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Tue, 21 Jul 2026 15:59:45 -0400 Subject: [PATCH 4/9] fix test name --- crates/iceberg/src/delete_file_index.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/iceberg/src/delete_file_index.rs b/crates/iceberg/src/delete_file_index.rs index 0a7f74d5bf..fa1ceaaa31 100644 --- a/crates/iceberg/src/delete_file_index.rs +++ b/crates/iceberg/src/delete_file_index.rs @@ -417,7 +417,7 @@ mod tests { } #[test] - fn deletion_vector_context_carries_coordinates() { + 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 From ad2cb1b310c3b3ea260cf99f9377de7fcabb1d65 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Tue, 18 Aug 2026 15:24:40 -0400 Subject: [PATCH 5/9] update --- crates/iceberg/src/scan/task.rs | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs index 3a72538af0..50b7399cdb 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -244,16 +244,22 @@ pub struct FileScanTaskDeleteFile { /// 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, /// 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, /// For a deletion vector, the length in bytes of the blob within its Puffin file. Set /// whenever `content_offset` is. + #[serde(default)] + #[serde(skip_serializing_if = "Option::is_none")] #[builder(default)] pub content_size_in_bytes: Option, From 764173c2eab07b75f0fa5c87af807aa60f13467c Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Thu, 20 Aug 2026 12:04:48 -0400 Subject: [PATCH 6/9] apply V3 deletion vectors during scan --- crates/iceberg/public-api.txt | 3 +- .../src/arrow/caching_delete_file_loader.rs | 480 +++++++++++++- .../iceberg/src/arrow/delete_file_loader.rs | 2 + .../src/arrow/reader/positional_deletes.rs | 184 ++++++ crates/iceberg/src/arrow/reader/row_filter.rs | 1 + crates/iceberg/src/delete_file_index.rs | 611 ++++++++++++++++-- crates/iceberg/src/delete_vector.rs | 36 +- crates/iceberg/src/scan/context.rs | 2 +- crates/iceberg/src/scan/task.rs | 15 +- crates/iceberg/src/test_utils.rs | 15 + 10 files changed, 1247 insertions(+), 102 deletions(-) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 2dacd5801e..1c3250b7dd 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1318,6 +1318,7 @@ 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> pub iceberg::scan::FileScanTaskDeleteFile::partition_spec_id: i32 +pub iceberg::scan::FileScanTaskDeleteFile::record_count: core::option::Option pub iceberg::scan::FileScanTaskDeleteFile::referenced_data_file: core::option::Option impl core::clone::Clone for iceberg::scan::FileScanTaskDeleteFile pub fn iceberg::scan::FileScanTaskDeleteFile::clone(&self) -> iceberg::scan::FileScanTaskDeleteFile @@ -1327,7 +1328,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 diff --git a/crates/iceberg/src/arrow/caching_delete_file_loader.rs b/crates/iceberg/src/arrow/caching_delete_file_loader.rs index eb5e1ac4b0..9a618cd9f1 100644 --- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs +++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs @@ -19,6 +19,7 @@ use std::collections::{HashMap, HashSet}; use std::sync::Arc; use arrow_array::{Array, ArrayRef, Int64Array, StringArray, StructArray}; +use bytes::Bytes; use futures::{StreamExt, TryStreamExt}; use tokio::sync::oneshot::{Receiver, channel}; @@ -27,6 +28,7 @@ use crate::arrow::delete_file_loader::BasicDeleteFileLoader; use crate::arrow::scan_metrics::ScanMetrics; use crate::arrow::{arrow_primitive_to_literal, arrow_schema_to_schema}; use crate::delete_vector::DeleteVector; +use crate::encryption::{EncryptedInputFile, StandardKeyMetadata}; use crate::expr::Predicate::AlwaysTrue; use crate::expr::{Predicate, Reference}; use crate::io::FileIO; @@ -51,7 +53,6 @@ pub(crate) struct CachingDeleteFileLoader { // Intermediate context during processing of a delete file task. enum DeleteFileContext { - // TODO: Delete Vector loader from Puffin files ExistingEqDel, ExistingPosDel, PosDels { @@ -63,6 +64,15 @@ enum DeleteFileContext { equality_ids: HashSet, sender: tokio::sync::oneshot::Sender, }, + // A V3 deletion vector: the raw deletion-vector-v1 blob bytes, the data file whose rows it + // deletes, and the manifest's expected cardinality. The blob is decoded and validated in the + // parse phase. + DelVec { + data_file_path: String, + blob: Bytes, + record_count: Option, + dv_path: String, + }, } // Final result of the processing of a delete file task before @@ -72,6 +82,11 @@ enum ParsedDeleteFileContext { file_path: String, results: HashMap, }, + // A single deletion vector decoded from a Puffin blob, keyed by the data file it applies to. + DelVec { + data_file_path: String, + delete_vector: DeleteVector, + }, EqDel, ExistingPosDel, } @@ -117,12 +132,13 @@ impl CachingDeleteFileLoader { /// tasks from starting to load the same equality delete file. We spawn a task to load /// the EQ delete's record batch stream, convert it to a predicate, update the delete filter, /// and notify any task that was waiting for it. - /// * When this gets updated to add support for delete vectors, the load phase will return - /// a PuffinReader for them. + /// * For a V3 deletion vector, the load phase reads the blob's byte range directly from its + /// Puffin file (decrypting first if the entry carries key metadata), and the parse phase + /// decodes it into a single `DeleteVector`. /// * The parse phase parses each record batch stream according to its associated data type. /// The result of this is a map of data file paths to delete vectors for the positional - /// delete tasks (and in future for the delete vector tasks). For equality delete - /// file tasks, this results in an unbound Predicate. + /// delete tasks, or a single (data file path, delete vector) pair for a deletion vector + /// task. For equality delete file tasks, this results in an unbound Predicate. /// * The unbound Predicates resulting from equality deletes are sent to their associated oneshot /// channel to store them in the right place in the delete file managers state. /// * The results of all of these futures are awaited on in parallel with the specified @@ -143,10 +159,10 @@ impl CachingDeleteFileLoader { /// | /// | /// +-----------------------------+--------------------------+ - /// Pos Del Del Vec (Not yet Implemented) EQ Del + /// Pos Del Del Vec EQ Del /// | | | /// [parse pos del stream] [parse del vec puffin] [parse eq del] - /// HashMap HashMap (Predicate, Sender) + /// HashMap DeleteVector (Predicate, Sender) /// | | | /// | | [persist to state] /// | | () @@ -211,13 +227,22 @@ impl CachingDeleteFileLoader { .try_buffer_unordered(concurrency_limit_data_files); while let Some(item) = results_stream.next().await { - let item = item?; - if let ParsedDeleteFileContext::DelVecs { file_path, results } = item { - for (data_file_path, delete_vector) in results.into_iter() { + match item? { + ParsedDeleteFileContext::DelVecs { file_path, results } => { + for (data_file_path, delete_vector) in results.into_iter() { + del_filter.upsert_delete_vector(data_file_path, delete_vector); + } + // Mark the positional delete file as fully loaded so waiters can proceed + del_filter.finish_pos_del_load(&file_path); + } + ParsedDeleteFileContext::DelVec { + data_file_path, + delete_vector, + } => { del_filter.upsert_delete_vector(data_file_path, delete_vector); } - // Mark the positional delete file as fully loaded so waiters can proceed - del_filter.finish_pos_del_load(&file_path); + ParsedDeleteFileContext::EqDel + | ParsedDeleteFileContext::ExistingPosDel => {} } } @@ -239,6 +264,17 @@ impl CachingDeleteFileLoader { ) -> Result { match task.file_type { DataContentType::PositionDeletes => { + // A V3 deletion vector arrives as a PositionDeletes entry whose deletes live in a + // Puffin blob located by content_offset, not in a positional-delete parquet file. + if let Some(content_offset) = task.content_offset { + return Self::load_deletion_vector( + task, + content_offset, + basic_delete_file_loader, + ) + .await; + } + match del_filter.try_start_pos_del_load(&task.file_path) { PosDelLoadAction::AlreadyLoaded => Ok(DeleteFileContext::ExistingPosDel), PosDelLoadAction::WaitFor(notified) => { @@ -299,6 +335,129 @@ impl CachingDeleteFileLoader { } } + /// Validates a `PositionDeletes` task's deletion-vector coordinates and returns them as + /// typed values ready for the range read: `(start, len, referenced data file path)`. + /// + /// The spec requires `referenced_data_file` and `content_size_in_bytes` whenever + /// `content_offset` is set (a deletion vector), so their absence 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, + content_offset: i64, + ) -> Result<(u64, u64, String)> { + let content_size = task.content_size_in_bytes.ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!( + "deletion vector {} sets content_offset but not 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 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)) + } + + /// Validates a decoded deletion vector's cardinality against the manifest entry's + /// `record_count`, mirroring Iceberg-Java's `BitmapPositionDeleteIndex.deserializeBitmap`. + /// `record_count` is only absent for a scan task built without a manifest entry (e.g. in a + /// test fixture), in which case there is nothing to validate against. + fn validate_deletion_vector_cardinality( + delete_vector: &DeleteVector, + expected: Option, + dv_path: &str, + ) -> Result<()> { + let Some(expected) = expected else { + return Ok(()); + }; + let actual = delete_vector.len(); + if actual != expected { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "deletion vector {dv_path} decoded to {actual} positions, expected {expected} from record_count" + ), + )); + } + Ok(()) + } + + /// Reads a V3 deletion vector blob directly from its Puffin file. + /// + /// The spec requires a delete manifest entry's `content_offset` / `content_size_in_bytes` to + /// match the blob's offset and length in the Puffin footer, so the blob is read by range + /// without parsing the footer. It is decoded into a [`DeleteVector`] in the parse phase. + /// + /// Decrypts the range read when `task.key_metadata` is set, the same way `ManifestReader` + /// decrypts a manifest file (`spec/manifest/reader.rs`): the coordinate space of + /// `content_offset` / `content_size_in_bytes` is the plaintext file, which is what + /// `EncryptedInputFile` reads over. + async fn load_deletion_vector( + task: &FileScanTaskDeleteFile, + content_offset: i64, + basic_delete_file_loader: BasicDeleteFileLoader, + ) -> Result { + let (start, len, data_file_path) = + Self::validate_deletion_vector_task(task, content_offset)?; + + let input_file = basic_delete_file_loader + .file_io() + .new_input(&task.file_path)?; + let blob = match task.key_metadata.as_deref() { + Some(key_metadata) => { + let key_metadata = StandardKeyMetadata::decode(key_metadata)?; + EncryptedInputFile::new(input_file, key_metadata) + .reader() + .await? + .read(start..start + len) + .await? + } + None => input_file.reader().await?.read(start..start + len).await?, + }; + + Ok(DeleteFileContext::DelVec { + data_file_path, + blob, + record_count: task.record_count, + dv_path: task.file_path.clone(), + }) + } + async fn parse_file_content_for_task( ctx: DeleteFileContext, ) -> Result { @@ -312,6 +471,20 @@ impl CachingDeleteFileLoader { results: del_vecs, }) } + DeleteFileContext::DelVec { + data_file_path, + blob, + record_count, + dv_path, + } => { + let delete_vector = DeleteVector::deserialize(&blob)?; + Self::validate_deletion_vector_cardinality(&delete_vector, record_count, &dv_path)?; + + Ok(ParsedDeleteFileContext::DelVec { + data_file_path, + delete_vector, + }) + } DeleteFileContext::FreshEqDel { sender, batch_stream, @@ -644,6 +817,7 @@ mod tests { use crate::arrow::delete_filter::tests::setup; use crate::scan::FileScanTaskDeleteFile; use crate::spec::{DataContentType, Schema}; + use crate::test_utils::encode_dv_blob; #[tokio::test] async fn test_delete_file_loader_parse_equality_deletes() { @@ -1206,4 +1380,286 @@ mod tests { // confirming that the second load reused the result from the first. assert!(Arc::ptr_eq(&dv1, &dv2)); } + + fn dv_task( + dv_path: String, + file_size: u64, + data_file_path: String, + content_offset: i64, + content_size: i64, + record_count: u64, + key_metadata: Option>, + ) -> FileScanTaskDeleteFile { + FileScanTaskDeleteFile::builder() + .with_file_path(dv_path) + .with_file_size_in_bytes(file_size) + .with_file_type(DataContentType::PositionDeletes) + .with_partition_spec_id(0) + .with_referenced_data_file(Some(data_file_path)) + .with_content_offset(Some(content_offset)) + .with_content_size_in_bytes(Some(content_size)) + .with_record_count(Some(record_count)) + .with_key_metadata(key_metadata) + .build() + } + + #[tokio::test] + async fn test_load_deletes_applies_deletion_vector() { + let tmp_dir = TempDir::new().unwrap(); + let table_location = tmp_dir.path().to_str().unwrap().to_string(); + let file_io = FileIO::new_with_fs(); + + let blob = encode_dv_blob([0u64, 1, 5]); + + // Embed the blob in a Puffin-like file behind leading bytes so content_offset is + // non-zero, then let the loader read it back by range. + let content_offset = 12i64; + let content_size = blob.len() as i64; + let mut file_bytes = vec![0u8; content_offset as usize]; + file_bytes.extend_from_slice(&blob); + file_bytes.extend_from_slice(&[0u8; 8]); + let dv_path = format!("{table_location}/deletes.puffin"); + std::fs::write(&dv_path, &file_bytes).unwrap(); + + let data_file_path = format!("{table_location}/data-1.parquet"); + let dv = dv_task( + dv_path.clone(), + std::fs::metadata(&dv_path).unwrap().len(), + data_file_path.clone(), + content_offset, + content_size, + 3, + None, + ); + + let schema = Arc::new( + Schema::builder() + .with_fields(vec![ + NestedField::optional(1, "x", Type::Primitive(PrimitiveType::Long)).into(), + ]) + .build() + .unwrap(), + ); + + let loader = CachingDeleteFileLoader::new(file_io, 10, Runtime::current()); + let delete_filter = loader.load_deletes(&[dv], schema).await.unwrap().unwrap(); + + let delete_vector = delete_filter + .get_delete_vector_for_path(&data_file_path) + .expect("a delete vector should be indexed for the referenced data file"); + let mut positions: Vec = delete_vector.lock().unwrap().iter().collect(); + positions.sort_unstable(); + assert_eq!(positions, vec![0, 1, 5]); + } + + #[tokio::test] + async fn test_load_deletes_decrypts_deletion_vector() { + use crate::encryption::{EncryptedOutputFile, StandardKeyMetadata}; + + let tmp_dir = TempDir::new().unwrap(); + let table_location = tmp_dir.path().to_str().unwrap().to_string(); + let file_io = FileIO::new_with_fs(); + + let key_metadata = StandardKeyMetadata::try_new(b"0123456789abcdef") + .unwrap() + .with_aad_prefix(b"test-aad-prefix!"); + let encoded_key_metadata = key_metadata.encode().unwrap(); + + let blob = encode_dv_blob([2u64, 4]); + let plaintext_size = blob.len() as i64; + let dv_path = format!("{table_location}/deletes.puffin"); + let output = EncryptedOutputFile::new(file_io.new_output(&dv_path).unwrap(), key_metadata); + output.write(Bytes::from(blob)).await.unwrap(); + + // content_offset / content_size_in_bytes are in the plaintext coordinate space, distinct + // from the ciphertext's on-disk size (header, nonce, and tag overhead). + let file_size = std::fs::metadata(&dv_path).unwrap().len(); + let data_file_path = format!("{table_location}/data-1.parquet"); + let dv = dv_task( + dv_path.clone(), + file_size, + data_file_path.clone(), + 0, + plaintext_size, + 2, + Some(encoded_key_metadata), + ); + + let schema = Arc::new( + Schema::builder() + .with_fields(vec![ + NestedField::optional(1, "x", Type::Primitive(PrimitiveType::Long)).into(), + ]) + .build() + .unwrap(), + ); + + let loader = CachingDeleteFileLoader::new(file_io, 10, Runtime::current()); + let delete_filter = loader.load_deletes(&[dv], schema).await.unwrap().unwrap(); + + let delete_vector = delete_filter + .get_delete_vector_for_path(&data_file_path) + .expect("a delete vector should be indexed for the referenced data file"); + let mut positions: Vec = delete_vector.lock().unwrap().iter().collect(); + positions.sort_unstable(); + assert_eq!(positions, vec![2, 4]); + } + + #[tokio::test] + async fn test_load_deletes_rejects_deletion_vector_cardinality_mismatch() { + let tmp_dir = TempDir::new().unwrap(); + let table_location = tmp_dir.path().to_str().unwrap().to_string(); + let file_io = FileIO::new_with_fs(); + + let blob = encode_dv_blob([0u64, 1, 5]); + let dv_path = format!("{table_location}/deletes.puffin"); + std::fs::write(&dv_path, &blob).unwrap(); + + let data_file_path = format!("{table_location}/data-1.parquet"); + // record_count says 2 positions, but the blob decodes to 3. + let dv = dv_task( + dv_path.clone(), + std::fs::metadata(&dv_path).unwrap().len(), + data_file_path, + 0, + blob.len() as i64, + 2, + None, + ); + + let schema = Arc::new( + Schema::builder() + .with_fields(vec![ + NestedField::optional(1, "x", Type::Primitive(PrimitiveType::Long)).into(), + ]) + .build() + .unwrap(), + ); + + let loader = CachingDeleteFileLoader::new(file_io, 10, Runtime::current()); + let err = loader + .load_deletes(&[dv], schema) + .await + .unwrap() + .unwrap_err(); + + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!(err.message().contains("expected 2 from record_count")); + } + + #[test] + fn test_validate_deletion_vector_task_rejects_missing_content_size() { + let task = FileScanTaskDeleteFile::builder() + .with_file_path("deletes.puffin".to_string()) + .with_file_size_in_bytes(100) + .with_file_type(DataContentType::PositionDeletes) + .with_partition_spec_id(0) + .with_referenced_data_file(Some("data.parquet".to_string())) + .build(); + + let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task, 4).unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!(err.message().contains("content_size_in_bytes")); + } + + #[test] + fn test_validate_deletion_vector_task_rejects_missing_referenced_data_file() { + let task = FileScanTaskDeleteFile::builder() + .with_file_path("deletes.puffin".to_string()) + .with_file_size_in_bytes(100) + .with_file_type(DataContentType::PositionDeletes) + .with_partition_spec_id(0) + .with_content_size_in_bytes(Some(40)) + .build(); + + let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task, 4).unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!(err.message().contains("referenced_data_file")); + } + + #[test] + fn test_validate_deletion_vector_task_rejects_negative_content_offset() { + let task = FileScanTaskDeleteFile::builder() + .with_file_path("deletes.puffin".to_string()) + .with_file_size_in_bytes(100) + .with_file_type(DataContentType::PositionDeletes) + .with_partition_spec_id(0) + .with_referenced_data_file(Some("data.parquet".to_string())) + .with_content_size_in_bytes(Some(40)) + .build(); + + let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task, -1).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 task = FileScanTaskDeleteFile::builder() + .with_file_path("deletes.puffin".to_string()) + .with_file_size_in_bytes(100) + .with_file_type(DataContentType::PositionDeletes) + .with_partition_spec_id(0) + .with_referenced_data_file(Some("data.parquet".to_string())) + .with_content_size_in_bytes(Some(-1)) + .build(); + + let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task, 4).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 task = FileScanTaskDeleteFile::builder() + .with_file_path("deletes.puffin".to_string()) + .with_file_size_in_bytes(100) + .with_file_type(DataContentType::PositionDeletes) + .with_partition_spec_id(0) + .with_referenced_data_file(Some("data.parquet".to_string())) + .with_content_size_in_bytes(Some(40)) + .build(); + + let (start, len, data_file_path) = + CachingDeleteFileLoader::validate_deletion_vector_task(&task, 4).unwrap(); + assert_eq!(start, 4); + assert_eq!(len, 40); + assert_eq!(data_file_path, "data.parquet"); + } + + #[test] + fn test_validate_deletion_vector_cardinality_ignores_missing_record_count() { + let dv = DeleteVector::default(); + CachingDeleteFileLoader::validate_deletion_vector_cardinality(&dv, None, "deletes.puffin") + .expect("no record_count means nothing to validate"); + } + + #[test] + fn test_validate_deletion_vector_cardinality_accepts_matching_count() { + let mut dv = DeleteVector::default(); + dv.insert(1); + dv.insert(2); + + CachingDeleteFileLoader::validate_deletion_vector_cardinality( + &dv, + Some(2), + "deletes.puffin", + ) + .unwrap(); + } + + #[test] + fn test_validate_deletion_vector_cardinality_rejects_mismatched_count() { + let mut dv = DeleteVector::default(); + dv.insert(1); + + let err = CachingDeleteFileLoader::validate_deletion_vector_cardinality( + &dv, + Some(2), + "deletes.puffin", + ) + .unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!(err.message().contains("expected 2 from record_count")); + } } diff --git a/crates/iceberg/src/arrow/delete_file_loader.rs b/crates/iceberg/src/arrow/delete_file_loader.rs index af470cf561..6a232cea0d 100644 --- a/crates/iceberg/src/arrow/delete_file_loader.rs +++ b/crates/iceberg/src/arrow/delete_file_loader.rs @@ -240,6 +240,7 @@ mod tests { referenced_data_file: None, content_offset: None, content_size_in_bytes: None, + record_count: None, }; let scan_metrics = ScanMetrics::new(); @@ -317,6 +318,7 @@ mod tests { referenced_data_file: None, content_offset: None, content_size_in_bytes: None, + record_count: None, }; let scan_metrics = ScanMetrics::new(); diff --git a/crates/iceberg/src/arrow/reader/positional_deletes.rs b/crates/iceberg/src/arrow/reader/positional_deletes.rs index 900a3d46d6..e03e6c6093 100644 --- a/crates/iceberg/src/arrow/reader/positional_deletes.rs +++ b/crates/iceberg/src/arrow/reader/positional_deletes.rs @@ -172,6 +172,7 @@ mod tests { use crate::io::FileIO; use crate::scan::{FileScanTask, FileScanTaskDeleteFile, FileScanTaskStream}; use crate::spec::{DataContentType, DataFileFormat, NestedField, PrimitiveType, Schema, Type}; + use crate::test_utils::encode_dv_blob; fn build_test_row_group_meta( schema_descr: SchemaDescPtr, @@ -923,4 +924,187 @@ mod tests { "Should have ids 101-200 (all of row group 1)" ); } + + /// End-to-end read of a data file with a V3 deletion vector applied. Exercises the whole + /// deletion-vector read path together: the DV coordinates on the scan task, the loader + /// reading the blob by range from its Puffin file, decoding it with DeleteVector::deserialize, + /// validating cardinality against record_count, and ArrowReader filtering the deleted rows. + #[tokio::test] + async fn test_deletion_vector_applied_end_to_end() { + use arrow_array::Int32Array; + + let tmp_dir = TempDir::new().unwrap(); + let table_location = tmp_dir.path().to_str().unwrap().to_string(); + + let table_schema = Arc::new( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(), + ]) + .build() + .unwrap(), + ); + let arrow_schema = Arc::new(ArrowSchema::new(vec![ + Field::new("id", DataType::Int32, false).with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + "1".to_string(), + )])), + ])); + + // Data file: ids 1..=5 at positions 0..=4. + let data_file_path = format!("{table_location}/data.parquet"); + let batch = RecordBatch::try_new(arrow_schema.clone(), vec![Arc::new( + Int32Array::from_iter_values(1..=5), + )]) + .unwrap(); + let props = WriterProperties::builder() + .set_compression(Compression::SNAPPY) + .build(); + let file = File::create(&data_file_path).unwrap(); + let mut writer = ArrowWriter::try_new(file, arrow_schema.clone(), Some(props)).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + + // Deletion vector deleting positions 1 and 3 (ids 2 and 4), serialized as a + // deletion-vector-v1 blob and embedded in a Puffin-like file at a non-zero offset. + let blob = encode_dv_blob([1u64, 3]); + let content_offset = 12i64; + let content_size = blob.len() as i64; + let mut dv_file_bytes = vec![0u8; content_offset as usize]; + dv_file_bytes.extend_from_slice(&blob); + let dv_path = format!("{table_location}/deletes.puffin"); + std::fs::write(&dv_path, &dv_file_bytes).unwrap(); + + let file_io = FileIO::new_with_fs(); + let reader = ArrowReaderBuilder::new(file_io, Runtime::current()).build(); + + let task = FileScanTask::builder() + .with_file_size_in_bytes(std::fs::metadata(&data_file_path).unwrap().len()) + .with_start(0) + .with_length(0) + .with_record_count(Some(5)) + .with_data_file_path(data_file_path.clone()) + .with_data_file_format(DataFileFormat::Parquet) + .with_schema(table_schema.clone()) + .with_project_field_ids(vec![1]) + .with_deletes(vec![ + FileScanTaskDeleteFile::builder() + .with_file_path(dv_path.clone()) + .with_file_size_in_bytes(std::fs::metadata(&dv_path).unwrap().len()) + .with_file_type(DataContentType::PositionDeletes) + .with_partition_spec_id(0) + .with_referenced_data_file(Some(data_file_path.clone())) + .with_content_offset(Some(content_offset)) + .with_content_size_in_bytes(Some(content_size)) + .with_record_count(Some(2)) + .build(), + ]) + .with_case_sensitive(false) + .build(); + + let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream; + let result = reader + .read(tasks) + .unwrap() + .stream() + .try_collect::>() + .await + .unwrap(); + + let ids: Vec = result + .iter() + .flat_map(|batch| { + batch + .column(0) + .as_primitive::() + .values() + .iter() + .copied() + }) + .collect(); + + // Positions 1 and 3 (ids 2 and 4) are deleted; ids 1, 3, 5 remain. + assert_eq!(ids, vec![1, 3, 5]); + } + + /// A deletion vector whose decoded cardinality disagrees with the manifest entry's + /// `record_count` must fail the read rather than silently applying the wrong deletes, the + /// same invariant Iceberg-Java enforces in `BitmapPositionDeleteIndex.deserializeBitmap`. + #[tokio::test] + async fn test_deletion_vector_cardinality_mismatch_fails_read() { + use arrow_array::Int32Array; + + let tmp_dir = TempDir::new().unwrap(); + let table_location = tmp_dir.path().to_str().unwrap().to_string(); + + let table_schema = Arc::new( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(), + ]) + .build() + .unwrap(), + ); + let arrow_schema = Arc::new(ArrowSchema::new(vec![ + Field::new("id", DataType::Int32, false).with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + "1".to_string(), + )])), + ])); + + let data_file_path = format!("{table_location}/data.parquet"); + let batch = RecordBatch::try_new(arrow_schema.clone(), vec![Arc::new( + Int32Array::from_iter_values(1..=5), + )]) + .unwrap(); + let props = WriterProperties::builder() + .set_compression(Compression::SNAPPY) + .build(); + let file = File::create(&data_file_path).unwrap(); + let mut writer = ArrowWriter::try_new(file, arrow_schema.clone(), Some(props)).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + + // The blob decodes to 2 positions, but the manifest entry claims 5. + let blob = encode_dv_blob([1u64, 3]); + let dv_path = format!("{table_location}/deletes.puffin"); + std::fs::write(&dv_path, &blob).unwrap(); + + let file_io = FileIO::new_with_fs(); + let reader = ArrowReaderBuilder::new(file_io, Runtime::current()).build(); + + let task = FileScanTask::builder() + .with_file_size_in_bytes(std::fs::metadata(&data_file_path).unwrap().len()) + .with_start(0) + .with_length(0) + .with_record_count(Some(5)) + .with_data_file_path(data_file_path.clone()) + .with_data_file_format(DataFileFormat::Parquet) + .with_schema(table_schema.clone()) + .with_project_field_ids(vec![1]) + .with_deletes(vec![ + FileScanTaskDeleteFile::builder() + .with_file_path(dv_path.clone()) + .with_file_size_in_bytes(std::fs::metadata(&dv_path).unwrap().len()) + .with_file_type(DataContentType::PositionDeletes) + .with_partition_spec_id(0) + .with_referenced_data_file(Some(data_file_path.clone())) + .with_content_offset(Some(0)) + .with_content_size_in_bytes(Some(blob.len() as i64)) + .with_record_count(Some(5)) + .build(), + ]) + .with_case_sensitive(false) + .build(); + + let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream; + let result: Result, _> = + reader.read(tasks).unwrap().stream().try_collect().await; + + let err = result.unwrap_err(); + assert_eq!(err.kind(), crate::ErrorKind::DataInvalid); + assert!(err.message().contains("expected 5 from record_count")); + } } diff --git a/crates/iceberg/src/arrow/reader/row_filter.rs b/crates/iceberg/src/arrow/reader/row_filter.rs index 6a33e90fd7..8c7b00aef3 100644 --- a/crates/iceberg/src/arrow/reader/row_filter.rs +++ b/crates/iceberg/src/arrow/reader/row_filter.rs @@ -1249,6 +1249,7 @@ mod tests { referenced_data_file: None, content_offset: None, content_size_in_bytes: None, + record_count: None, key_metadata: None, }], partition: None, diff --git a/crates/iceberg/src/delete_file_index.rs b/crates/iceberg/src/delete_file_index.rs index 09f7b3c25b..1d634ad0d4 100644 --- a/crates/iceberg/src/delete_file_index.rs +++ b/crates/iceberg/src/delete_file_index.rs @@ -27,6 +27,7 @@ use crate::metadata_columns::RESERVED_FIELD_ID_DELETE_FILE_PATH; use crate::runtime::Runtime; use crate::scan::{DeleteFileContext, FileScanTaskDeleteFile}; use crate::spec::{DataContentType, DataFile, PrimitiveLiteral, Struct}; +use crate::{Error, ErrorKind, Result}; /// Index of delete files #[derive(Debug, Clone)] @@ -36,8 +37,18 @@ pub(crate) struct DeleteFileIndex { #[derive(Debug)] enum DeleteFileIndexState { + // Arc, not Box: a waiter clones this out and awaits `notified_owned()` after dropping the + // read lock (a borrowed `notified()` future can't outlive the guard it was created under). + // If multiple callers arrive while still Populating, each clones its own handle to the same + // Notify, so one `notify_waiters()` call wakes all of them. Populating(Arc), - Populated(PopulatedDeleteFileIndex), + // Boxed because PopulatedDeleteFileIndex is large relative to the other variants; there is + // exactly one owner (this enum, behind the RwLock), so this needs heap indirection, not + // shared ownership. + Populated(Box), + // Boxed for the same reason: Error is large enough to trip the same size check, and is never + // cloned out of the lock, only read as `&Error` via deref. + Failed(Box), } #[derive(Debug)] @@ -46,7 +57,10 @@ struct PopulatedDeleteFileIndex { eq_deletes_by_partition: HashMap>>, pos_deletes_by_partition: HashMap>>, pos_deletes_by_path: HashMap>>, - // TODO: Deletion Vector support + // V3 deletion vectors, keyed by the data file they apply to (referenced_data_file). At most + // one exists per data file per snapshot, and when one applies it supersedes any position + // delete files for that data file, partition-scoped or path-scoped alike. + dvs_by_referenced_data_file: HashMap>, } /// Determines the single data file referenced by a position delete file, if any. @@ -79,6 +93,13 @@ fn referenced_data_file(data_file: &DataFile) -> Option { } } +/// Rebuilds an owned `Error` from the boxed `Error` cached in `DeleteFileIndexState::Failed`. +/// `Error` isn't `Clone`, so each waiter gets its own copy of the kind and message rather than +/// sharing the original's backtrace and source chain. +fn clone_failed_index_error(err: &Error) -> Error { + Error::new(err.kind(), err.message().to_string()) +} + impl DeleteFileIndex { /// create a new `DeleteFileIndex` along with the sender that populates it with delete files pub(crate) fn new(runtime: Runtime) -> (DeleteFileIndex, Sender) { @@ -96,11 +117,14 @@ impl DeleteFileIndex { let delete_files: Vec = delete_file_stream.collect::>().await; - let populated_delete_file_index = PopulatedDeleteFileIndex::new(delete_files); + let new_state = match PopulatedDeleteFileIndex::new(delete_files) { + Ok(index) => DeleteFileIndexState::Populated(Box::new(index)), + Err(err) => DeleteFileIndexState::Failed(Box::new(err)), + }; { let mut guard = state.write().unwrap(); - *guard = DeleteFileIndexState::Populated(populated_delete_file_index); + *guard = new_state; } notify.notify_waiters(); } @@ -110,11 +134,15 @@ impl DeleteFileIndex { } /// Gets all the delete files that apply to the specified data file. + /// + /// Fails if building the index found a spec violation, such as multiple deletion vectors + /// referencing the same data file, or if a matched deletion vector's sequence number + /// violates the spec relative to `seq_num`. pub(crate) async fn get_deletes_for_data_file( &self, data_file: &DataFile, seq_num: Option, - ) -> Vec { + ) -> Result> { // Create the `Notified` while holding the read lock. The read lock ensures that // when we go inside it, either the state is already at Populated or it is still // at Populating AND `notify_waiters()` has not been called yet. Any `Notified` @@ -127,6 +155,9 @@ impl DeleteFileIndex { DeleteFileIndexState::Populated(index) => { return index.get_deletes_for_data_file(data_file, seq_num); } + DeleteFileIndexState::Failed(err) => { + return Err(clone_failed_index_error(err)); + } } }; @@ -137,7 +168,10 @@ impl DeleteFileIndex { DeleteFileIndexState::Populated(index) => { index.get_deletes_for_data_file(data_file, seq_num) } - _ => unreachable!("Cannot be any other state than loaded"), + DeleteFileIndexState::Failed(err) => Err(clone_failed_index_error(err)), + DeleteFileIndexState::Populating(_) => { + unreachable!("Cannot still be Populating after being notified") + } } } } @@ -146,31 +180,68 @@ impl PopulatedDeleteFileIndex { /// Creates a new populated delete file index from a list of delete file contexts, which /// allows for fast lookup when determining which delete files apply to a given data file. /// - /// 1. Position deletes that reference a single data file, either through the + /// 1. A V3 deletion vector (a `PositionDeletes` entry with `content_offset` set) is indexed + /// by the `referenced_data_file` field, which the spec requires for deletion vectors. + /// Fails if two deletion vectors reference the same data file: the spec allows at most + /// one deletion vector per data file per snapshot. + /// 2. Other position deletes that reference a single data file, either through the /// `referenced_data_file` field or through equal `file_path` column bounds, /// are indexed by that data file's path. - /// 2. All other position deletes are indexed by the partition extracted from + /// 3. All other position deletes are indexed by the partition extracted from /// their manifest entry. - /// 3. Equality deletes stored with an unpartitioned spec are applied as global + /// 4. Equality deletes stored with an unpartitioned spec are applied as global /// deletes, per the spec. All other equality deletes are indexed by partition. - fn new(files: Vec) -> PopulatedDeleteFileIndex { + fn new(files: Vec) -> Result { let mut eq_deletes_by_partition: HashMap>> = HashMap::default(); let mut pos_deletes_by_partition: HashMap>> = HashMap::default(); let mut pos_deletes_by_path: HashMap>> = HashMap::default(); + let mut dvs_by_referenced_data_file: HashMap> = + HashMap::default(); let mut global_equality_deletes: Vec> = vec![]; - files.into_iter().for_each(|ctx| { + for ctx in files { let arc_ctx = Arc::new(ctx); - let partition = arc_ctx.manifest_entry.data_file().partition(); + let data_file = arc_ctx.manifest_entry.data_file(); + let partition = data_file.partition(); match arc_ctx.manifest_entry.content_type() { DataContentType::PositionDeletes => { - if let Some(path) = referenced_data_file(arc_ctx.manifest_entry.data_file()) { + if data_file.content_offset().is_some() { + // The spec requires referenced_data_file whenever content_offset is set + // (a deletion vector), so its absence here is a malformed manifest entry, + // not an ordinary position delete to fall back on. + let Some(path) = data_file.referenced_data_file() else { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "deletion vector {} sets content_offset but is missing referenced_data_file", + arc_ctx.manifest_entry.file_path() + ), + )); + }; + + if let Some(existing) = + dvs_by_referenced_data_file.insert(path.clone(), arc_ctx) + { + let inserted = &dvs_by_referenced_data_file[&path]; + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "found multiple deletion vectors for data file {path}: {} and {}", + existing.manifest_entry.file_path(), + inserted.manifest_entry.file_path() + ), + )); + } + continue; + } + + if let Some(path) = referenced_data_file(data_file) { pos_deletes_by_path.entry(path).or_default().push(arc_ctx); } else { pos_deletes_by_partition @@ -192,22 +263,29 @@ impl PopulatedDeleteFileIndex { } _ => unreachable!(), } - }); + } - PopulatedDeleteFileIndex { + Ok(PopulatedDeleteFileIndex { global_equality_deletes, eq_deletes_by_partition, pos_deletes_by_partition, pos_deletes_by_path, - } + dvs_by_referenced_data_file, + }) } /// Determine all the delete files that apply to the provided `DataFile`. + /// + /// Fails if a matched deletion vector's partition or data sequence number is inconsistent + /// with the data file's: a data file's path is permanently tied to one partition, and the + /// spec guarantees a DV is only ever written at or after the sequence number of the data + /// file it applies to, so either violation means the delete manifest is inconsistent, not + /// that the DV simply doesn't apply. fn get_deletes_for_data_file( &self, data_file: &DataFile, seq_num: Option, - ) -> Vec { + ) -> Result> { let mut results = vec![]; self.global_equality_deletes @@ -233,6 +311,49 @@ impl PopulatedDeleteFileIndex { .for_each(|delete| results.push(delete.as_ref().into())); } + // A deletion vector supersedes all position delete files for its data file, per the spec: + // "readers ignore any position delete files that would otherwise match it, because the DV + // subsumes them". An exact path match on referenced_data_file is sufficient proof of + // applicability, the same as for pos_deletes_by_path below, so this is checked before + // (and instead of) pos_deletes_by_partition and pos_deletes_by_path. + if let Some(dv) = self.dvs_by_referenced_data_file.get(data_file.file_path()) { + let dv_data_file = dv.manifest_entry.data_file(); + // A file path belongs to exactly one partition for its lifetime, so an exact path + // match already implies partition equality; this checks that the manifest agrees, + // per the spec's explicit partition-equality condition for deletion vectors. + if data_file.partition() != dv_data_file.partition() + || data_file.partition_spec_id != dv.partition_spec_id + { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "deletion vector {} references data file {} but its partition (spec {}, {:?}) does not match the data file's partition (spec {}, {:?})", + dv.manifest_entry.file_path(), + data_file.file_path(), + dv.partition_spec_id, + dv_data_file.partition(), + data_file.partition_spec_id, + data_file.partition() + ), + )); + } + + if let Some(seq_num) = seq_num { + let dv_seq = dv.manifest_entry.sequence_number(); + if dv_seq < Some(seq_num) { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "deletion vector {} has data sequence number {dv_seq:?}, which must be >= the data file's sequence number {seq_num}", + dv.manifest_entry.file_path() + ), + )); + } + } + results.push(dv.as_ref().into()); + return Ok(results); + } + if let Some(deletes) = self.pos_deletes_by_partition.get(data_file.partition()) { deletes .iter() @@ -261,7 +382,7 @@ impl PopulatedDeleteFileIndex { .for_each(|delete| results.push(delete.as_ref().into())); } - results + Ok(results) } } @@ -297,23 +418,26 @@ mod tests { }) .collect(); - let delete_file_index = PopulatedDeleteFileIndex::new(delete_contexts); + let delete_file_index = PopulatedDeleteFileIndex::new(delete_contexts).unwrap(); let data_file = build_unpartitioned_data_file(); // All deletes apply to sequence 0 - let delete_files_to_apply_for_seq_0 = - delete_file_index.get_deletes_for_data_file(&data_file, Some(0)); + let delete_files_to_apply_for_seq_0 = delete_file_index + .get_deletes_for_data_file(&data_file, Some(0)) + .unwrap(); assert_eq!(delete_files_to_apply_for_seq_0.len(), 4); // All deletes apply to sequence 3 - let delete_files_to_apply_for_seq_3 = - delete_file_index.get_deletes_for_data_file(&data_file, Some(3)); + let delete_files_to_apply_for_seq_3 = delete_file_index + .get_deletes_for_data_file(&data_file, Some(3)) + .unwrap(); assert_eq!(delete_files_to_apply_for_seq_3.len(), 4); // Last 3 deletes apply to sequence 4 - let delete_files_to_apply_for_seq_4 = - delete_file_index.get_deletes_for_data_file(&data_file, Some(4)); + let delete_files_to_apply_for_seq_4 = delete_file_index + .get_deletes_for_data_file(&data_file, Some(4)) + .unwrap(); let actual_paths_to_apply_for_seq_4: Vec = delete_files_to_apply_for_seq_4 .into_iter() .map(|file| file.file_path) @@ -325,8 +449,9 @@ mod tests { ); // Last 3 deletes apply to sequence 5 - let delete_files_to_apply_for_seq_5 = - delete_file_index.get_deletes_for_data_file(&data_file, Some(5)); + let delete_files_to_apply_for_seq_5 = delete_file_index + .get_deletes_for_data_file(&data_file, Some(5)) + .unwrap(); let actual_paths_to_apply_for_seq_5: Vec = delete_files_to_apply_for_seq_5 .into_iter() .map(|file| file.file_path) @@ -337,8 +462,9 @@ mod tests { ); // Only the last position delete applies to sequence 6 - let delete_files_to_apply_for_seq_6 = - delete_file_index.get_deletes_for_data_file(&data_file, Some(6)); + let delete_files_to_apply_for_seq_6 = delete_file_index + .get_deletes_for_data_file(&data_file, Some(6)) + .unwrap(); let actual_paths_to_apply_for_seq_6: Vec = delete_files_to_apply_for_seq_6 .into_iter() .map(|file| file.file_path) @@ -352,8 +478,9 @@ mod tests { let partitioned_file = build_partitioned_data_file(&Struct::from_iter([Some(Literal::long(100))]), 1); - let delete_files_to_apply_for_partitioned_file = - delete_file_index.get_deletes_for_data_file(&partitioned_file, Some(0)); + let delete_files_to_apply_for_partitioned_file = delete_file_index + .get_deletes_for_data_file(&partitioned_file, Some(0)) + .unwrap(); let actual_paths_to_apply_for_partitioned_file: Vec = delete_files_to_apply_for_partitioned_file .into_iter() @@ -389,24 +516,27 @@ mod tests { }) .collect(); - let delete_file_index = PopulatedDeleteFileIndex::new(delete_contexts); + let delete_file_index = PopulatedDeleteFileIndex::new(delete_contexts).unwrap(); let partitioned_file = build_partitioned_data_file(&Struct::from_iter([Some(Literal::long(100))]), spec_id); // All deletes apply to sequence 0 - let delete_files_to_apply_for_seq_0 = - delete_file_index.get_deletes_for_data_file(&partitioned_file, Some(0)); + let delete_files_to_apply_for_seq_0 = delete_file_index + .get_deletes_for_data_file(&partitioned_file, Some(0)) + .unwrap(); assert_eq!(delete_files_to_apply_for_seq_0.len(), 4); // All deletes apply to sequence 3 - let delete_files_to_apply_for_seq_3 = - delete_file_index.get_deletes_for_data_file(&partitioned_file, Some(3)); + let delete_files_to_apply_for_seq_3 = delete_file_index + .get_deletes_for_data_file(&partitioned_file, Some(3)) + .unwrap(); assert_eq!(delete_files_to_apply_for_seq_3.len(), 4); // Last 3 deletes apply to sequence 4 - let delete_files_to_apply_for_seq_4 = - delete_file_index.get_deletes_for_data_file(&partitioned_file, Some(4)); + let delete_files_to_apply_for_seq_4 = delete_file_index + .get_deletes_for_data_file(&partitioned_file, Some(4)) + .unwrap(); let actual_paths_to_apply_for_seq_4: Vec = delete_files_to_apply_for_seq_4 .into_iter() .map(|file| file.file_path) @@ -418,8 +548,9 @@ mod tests { ); // Last 3 deletes apply to sequence 5 - let delete_files_to_apply_for_seq_5 = - delete_file_index.get_deletes_for_data_file(&partitioned_file, Some(5)); + let delete_files_to_apply_for_seq_5 = delete_file_index + .get_deletes_for_data_file(&partitioned_file, Some(5)) + .unwrap(); let actual_paths_to_apply_for_seq_5: Vec = delete_files_to_apply_for_seq_5 .into_iter() .map(|file| file.file_path) @@ -430,8 +561,9 @@ mod tests { ); // Only the last position delete applies to sequence 6 - let delete_files_to_apply_for_seq_6 = - delete_file_index.get_deletes_for_data_file(&partitioned_file, Some(6)); + let delete_files_to_apply_for_seq_6 = delete_file_index + .get_deletes_for_data_file(&partitioned_file, Some(6)) + .unwrap(); let actual_paths_to_apply_for_seq_6: Vec = delete_files_to_apply_for_seq_6 .into_iter() .map(|file| file.file_path) @@ -444,8 +576,9 @@ mod tests { // Data file with different partition tuples does not match any delete files let partitioned_second_file = build_partitioned_data_file(&Struct::from_iter([Some(Literal::long(200))]), 1); - let delete_files_to_apply_for_different_partition = - delete_file_index.get_deletes_for_data_file(&partitioned_second_file, Some(0)); + let delete_files_to_apply_for_different_partition = delete_file_index + .get_deletes_for_data_file(&partitioned_second_file, Some(0)) + .unwrap(); let actual_paths_to_apply_for_different_partition: Vec = delete_files_to_apply_for_different_partition .into_iter() @@ -455,8 +588,9 @@ mod tests { // Data file with same tuple but different spec ID does not match any delete files let partitioned_different_spec = build_partitioned_data_file(&partition_one, 2); - let delete_files_to_apply_for_different_spec = - delete_file_index.get_deletes_for_data_file(&partitioned_different_spec, Some(0)); + let delete_files_to_apply_for_different_spec = delete_file_index + .get_deletes_for_data_file(&partitioned_different_spec, Some(0)) + .unwrap(); let actual_paths_to_apply_for_different_spec: Vec = delete_files_to_apply_for_different_spec .into_iter() @@ -474,15 +608,20 @@ mod tests { let index = PopulatedDeleteFileIndex::new(vec![DeleteFileContext { manifest_entry: build_added_manifest_entry(5, &pos_delete).into(), partition_spec_id: 0, - }]); + }]) + .unwrap(); - let deletes_for_a = index.get_deletes_for_data_file(&data_file_a, Some(0)); + let deletes_for_a = index + .get_deletes_for_data_file(&data_file_a, Some(0)) + .unwrap(); assert_eq!(deletes_for_a.len(), 1); assert_eq!(deletes_for_a[0].file_path, pos_delete.file_path()); // The delete references data file A, so it must not apply to data file B // even though B shares A's partition. - let deletes_for_b = index.get_deletes_for_data_file(&data_file_b, Some(0)); + let deletes_for_b = index + .get_deletes_for_data_file(&data_file_b, Some(0)) + .unwrap(); assert!(deletes_for_b.is_empty()); } @@ -498,15 +637,20 @@ mod tests { let index = PopulatedDeleteFileIndex::new(vec![DeleteFileContext { manifest_entry: build_added_manifest_entry(5, &pos_delete).into(), partition_spec_id: 0, - }]); + }]) + .unwrap(); assert_eq!( - index.get_deletes_for_data_file(&data_file_a, Some(0)).len(), + index + .get_deletes_for_data_file(&data_file_a, Some(0)) + .unwrap() + .len(), 1 ); assert!( index .get_deletes_for_data_file(&data_file_b, Some(0)) + .unwrap() .is_empty() ); } @@ -523,14 +667,21 @@ mod tests { let index = PopulatedDeleteFileIndex::new(vec![DeleteFileContext { manifest_entry: build_added_manifest_entry(5, &pos_delete).into(), partition_spec_id: 0, - }]); + }]) + .unwrap(); assert_eq!( - index.get_deletes_for_data_file(&data_file_a, Some(0)).len(), + index + .get_deletes_for_data_file(&data_file_a, Some(0)) + .unwrap() + .len(), 1 ); assert_eq!( - index.get_deletes_for_data_file(&data_file_b, Some(0)).len(), + index + .get_deletes_for_data_file(&data_file_b, Some(0)) + .unwrap() + .len(), 1 ); } @@ -546,10 +697,14 @@ mod tests { let index = PopulatedDeleteFileIndex::new(vec![DeleteFileContext { manifest_entry: build_added_manifest_entry(5, &pos_delete).into(), partition_spec_id: 0, - }]); + }]) + .unwrap(); assert_eq!( - index.get_deletes_for_data_file(&data_file, Some(0)).len(), + index + .get_deletes_for_data_file(&data_file, Some(0)) + .unwrap() + .len(), 1 ); } @@ -562,25 +717,39 @@ mod tests { let index = PopulatedDeleteFileIndex::new(vec![DeleteFileContext { manifest_entry: build_added_manifest_entry(5, &pos_delete).into(), partition_spec_id: 0, - }]); + }]) + .unwrap(); // Position deletes apply when the delete's sequence number is greater // than or equal to the data file's. assert_eq!( - index.get_deletes_for_data_file(&data_file, Some(4)).len(), + index + .get_deletes_for_data_file(&data_file, Some(4)) + .unwrap() + .len(), 1 ); assert_eq!( - index.get_deletes_for_data_file(&data_file, Some(5)).len(), + index + .get_deletes_for_data_file(&data_file, Some(5)) + .unwrap() + .len(), 1 ); assert!( index .get_deletes_for_data_file(&data_file, Some(6)) + .unwrap() .is_empty() ); // Without a sequence number, the delete applies unconditionally. - assert_eq!(index.get_deletes_for_data_file(&data_file, None).len(), 1); + assert_eq!( + index + .get_deletes_for_data_file(&data_file, None) + .unwrap() + .len(), + 1 + ); } #[test] @@ -607,10 +776,12 @@ mod tests { manifest_entry: build_added_manifest_entry(5, &eq_delete).into(), partition_spec_id: 0, }, - ]); + ]) + .unwrap(); let mut actual_paths: Vec = index .get_deletes_for_data_file(&data_file, Some(0)) + .unwrap() .into_iter() .map(|delete| delete.file_path) .collect(); @@ -642,10 +813,12 @@ mod tests { manifest_entry: build_added_manifest_entry(6, &second_delete).into(), partition_spec_id: 0, }, - ]); + ]) + .unwrap(); let mut actual_paths: Vec = index .get_deletes_for_data_file(&data_file, Some(0)) + .unwrap() .into_iter() .map(|delete| delete.file_path) .collect(); @@ -679,15 +852,19 @@ mod tests { let index = PopulatedDeleteFileIndex::new(vec![DeleteFileContext { manifest_entry: build_added_manifest_entry(5, &pos_delete).into(), partition_spec_id: 1, - }]); + }]) + .unwrap(); - let deletes_for_referenced = index.get_deletes_for_data_file(&data_file, Some(0)); + let deletes_for_referenced = index + .get_deletes_for_data_file(&data_file, Some(0)) + .unwrap(); assert_eq!(deletes_for_referenced.len(), 1); assert_eq!(deletes_for_referenced[0].file_path, pos_delete.file_path()); assert!( index .get_deletes_for_data_file(&same_partition_neighbor, Some(0)) + .unwrap() .is_empty() ); } @@ -706,15 +883,20 @@ mod tests { let index = PopulatedDeleteFileIndex::new(vec![DeleteFileContext { manifest_entry: build_added_manifest_entry(5, &pos_delete).into(), partition_spec_id: 0, - }]); + }]) + .unwrap(); assert_eq!( - index.get_deletes_for_data_file(&data_file, Some(0)).len(), + index + .get_deletes_for_data_file(&data_file, Some(0)) + .unwrap() + .len(), 1 ); assert_eq!( index .get_deletes_for_data_file(&other_data_file, Some(0)) + .unwrap() .len(), 1 ); @@ -784,12 +966,309 @@ mod tests { 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.record_count, Some(3)); assert_eq!( task.referenced_data_file.as_deref(), Some("s3://bucket/data/part-0.parquet") ); } + #[test] + fn test_deletion_vector_supersedes_position_deletes() { + let partition = Struct::from_iter([Some(Literal::long(100))]); + let spec_id = 1; + let data_file = build_partitioned_data_file(&partition, spec_id); + + let dv = build_deletion_vector(data_file.file_path(), &partition, spec_id); + let dv_path = dv.file_path().to_string(); + let pos_del = build_partitioned_pos_delete(&partition, spec_id); + + let contexts = vec![ + DeleteFileContext { + manifest_entry: build_added_manifest_entry(5, &dv).into(), + partition_spec_id: spec_id, + }, + DeleteFileContext { + manifest_entry: build_added_manifest_entry(5, &pos_del).into(), + partition_spec_id: spec_id, + }, + ]; + + let index = PopulatedDeleteFileIndex::new(contexts).unwrap(); + let applied: Vec = index + .get_deletes_for_data_file(&data_file, Some(0)) + .unwrap() + .into_iter() + .map(|f| f.file_path) + .collect(); + + // Only the deletion vector applies; the partition-scoped position delete file, which + // would otherwise also match, is superseded. + assert_eq!(applied, vec![dv_path]); + } + + #[test] + fn test_deletion_vector_with_stale_sequence_number_is_rejected() { + let partition = Struct::from_iter([Some(Literal::long(100))]); + let spec_id = 1; + let data_file = build_partitioned_data_file(&partition, spec_id); + + let dv = build_deletion_vector(data_file.file_path(), &partition, spec_id); + + let contexts = vec![DeleteFileContext { + manifest_entry: build_added_manifest_entry(3, &dv).into(), + partition_spec_id: spec_id, + }]; + + let index = PopulatedDeleteFileIndex::new(contexts).unwrap(); + // The DV's own sequence number (3) is less than the data file's (5): the spec guarantees + // a DV is never written before the data file it applies to, so this is an inconsistent + // manifest rather than a case where the DV simply doesn't apply. + let err = index + .get_deletes_for_data_file(&data_file, Some(5)) + .unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!( + err.message() + .contains("must be >= the data file's sequence number") + ); + } + + #[test] + fn test_deletion_vector_with_mismatched_partition_is_rejected() { + let partition = Struct::from_iter([Some(Literal::long(100))]); + let other_partition = Struct::from_iter([Some(Literal::long(200))]); + let spec_id = 1; + let data_file = build_partitioned_data_file(&partition, spec_id); + + // Malformed: the DV's referenced_data_file matches data_file's path exactly, but the + // DV's own partition disagrees, a state that cannot arise from a valid writer since a + // file path is permanently tied to one partition. + let dv = build_deletion_vector(data_file.file_path(), &other_partition, spec_id); + + let contexts = vec![DeleteFileContext { + manifest_entry: build_added_manifest_entry(5, &dv).into(), + partition_spec_id: spec_id, + }]; + + let index = PopulatedDeleteFileIndex::new(contexts).unwrap(); + let err = index + .get_deletes_for_data_file(&data_file, Some(0)) + .unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!( + err.message() + .contains("does not match the data file's partition") + ); + } + + #[test] + fn test_deletion_vector_with_mismatched_partition_spec_is_rejected() { + let partition = Struct::from_iter([Some(Literal::long(100))]); + let data_file = build_partitioned_data_file(&partition, 1); + + // Same partition value, but the DV's context was populated under a different partition + // spec id than the data file's: also a manifest inconsistency, since a file path is + // permanently tied to one partition spec. + let dv = build_deletion_vector(data_file.file_path(), &partition, 1); + + let contexts = vec![DeleteFileContext { + manifest_entry: build_added_manifest_entry(5, &dv).into(), + partition_spec_id: 2, + }]; + + let index = PopulatedDeleteFileIndex::new(contexts).unwrap(); + let err = index + .get_deletes_for_data_file(&data_file, Some(0)) + .unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!( + err.message() + .contains("does not match the data file's partition") + ); + } + + #[test] + fn test_deletion_vector_supersedes_path_scoped_position_delete() { + let partition = Struct::from_iter([Some(Literal::long(100))]); + let spec_id = 1; + let data_file = build_partitioned_data_file(&partition, spec_id); + + let dv = build_deletion_vector(data_file.file_path(), &partition, spec_id); + let dv_path = dv.file_path().to_string(); + let pos_del = build_pos_delete_referencing(data_file.file_path()); + + let contexts = vec![ + DeleteFileContext { + manifest_entry: build_added_manifest_entry(5, &dv).into(), + partition_spec_id: spec_id, + }, + DeleteFileContext { + manifest_entry: build_added_manifest_entry(5, &pos_del).into(), + partition_spec_id: 0, + }, + ]; + + let index = PopulatedDeleteFileIndex::new(contexts).unwrap(); + let applied: Vec = index + .get_deletes_for_data_file(&data_file, Some(0)) + .unwrap() + .into_iter() + .map(|f| f.file_path) + .collect(); + + // Only the deletion vector applies; the path-scoped position delete file, which would + // otherwise also match by exact referenced_data_file path, is superseded. + assert_eq!(applied, vec![dv_path]); + } + + #[test] + fn test_deletion_vector_coexists_with_equality_delete() { + let partition = Struct::from_iter([Some(Literal::long(100))]); + let spec_id = 1; + let data_file = build_partitioned_data_file(&partition, spec_id); + + let dv = build_deletion_vector(data_file.file_path(), &partition, spec_id); + let dv_path = dv.file_path().to_string(); + let eq_del = build_unpartitioned_eq_delete(); + let eq_del_path = eq_del.file_path().to_string(); + + let contexts = vec![ + DeleteFileContext { + manifest_entry: build_added_manifest_entry(5, &dv).into(), + partition_spec_id: spec_id, + }, + DeleteFileContext { + manifest_entry: build_added_manifest_entry(5, &eq_del).into(), + partition_spec_id: 0, + }, + ]; + + let index = PopulatedDeleteFileIndex::new(contexts).unwrap(); + let mut applied: Vec = index + .get_deletes_for_data_file(&data_file, Some(0)) + .unwrap() + .into_iter() + .map(|f| f.file_path) + .collect(); + applied.sort(); + + // A deletion vector only supersedes position deletes, not equality deletes: both apply. + let mut expected = vec![dv_path, eq_del_path]; + expected.sort(); + assert_eq!(applied, expected); + } + + #[test] + fn test_deletion_vector_missing_referenced_data_file_is_rejected() { + let malformed_dv = DataFileBuilder::default() + .file_path("deletes.puffin".to_string()) + .file_format(DataFileFormat::Puffin) + .content(DataContentType::PositionDeletes) + .record_count(1) + .content_offset(Some(4)) + .content_size_in_bytes(Some(40)) + .partition(Struct::empty()) + .partition_spec_id(0) + .file_size_in_bytes(60) + .build() + .unwrap(); + + let contexts = vec![DeleteFileContext { + manifest_entry: build_added_manifest_entry(5, &malformed_dv).into(), + partition_spec_id: 0, + }]; + + let err = PopulatedDeleteFileIndex::new(contexts).unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!(err.message().contains("missing referenced_data_file")); + } + + #[test] + fn test_multiple_deletion_vectors_for_same_data_file_is_rejected() { + let partition = Struct::from_iter([Some(Literal::long(100))]); + let spec_id = 1; + let data_file = build_partitioned_data_file(&partition, spec_id); + + let dv_1 = build_deletion_vector(data_file.file_path(), &partition, spec_id); + let dv_2 = build_deletion_vector(data_file.file_path(), &partition, spec_id); + + let contexts = vec![ + DeleteFileContext { + manifest_entry: build_added_manifest_entry(5, &dv_1).into(), + partition_spec_id: spec_id, + }, + DeleteFileContext { + manifest_entry: build_added_manifest_entry(6, &dv_2).into(), + partition_spec_id: spec_id, + }, + ]; + + let err = PopulatedDeleteFileIndex::new(contexts).unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!(err.message().contains("multiple deletion vectors")); + } + + #[tokio::test] + async fn test_delete_file_index_propagates_multiple_dv_error_to_waiters() { + let partition = Struct::from_iter([Some(Literal::long(100))]); + let spec_id = 1; + let data_file = build_partitioned_data_file(&partition, spec_id); + + let dv_1 = build_deletion_vector(data_file.file_path(), &partition, spec_id); + let dv_2 = build_deletion_vector(data_file.file_path(), &partition, spec_id); + + let (index, mut tx) = DeleteFileIndex::new(Runtime::current()); + tx.try_send(DeleteFileContext { + manifest_entry: build_added_manifest_entry(5, &dv_1).into(), + partition_spec_id: spec_id, + }) + .unwrap(); + tx.try_send(DeleteFileContext { + manifest_entry: build_added_manifest_entry(6, &dv_2).into(), + partition_spec_id: spec_id, + }) + .unwrap(); + drop(tx); + + let err = index + .get_deletes_for_data_file(&data_file, Some(0)) + .await + .unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!(err.message().contains("multiple deletion vectors")); + + // A second caller, arriving after the index has already settled into the Failed state, + // must see the same error rather than panicking on an unexpected state. + let err = index + .get_deletes_for_data_file(&data_file, Some(0)) + .await + .unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + } + + // A V3 deletion vector: a PositionDeletes entry stored as a Puffin blob (content_offset set) + // scoped to a single data file via referenced_data_file. + fn build_deletion_vector( + referenced_data_file: &str, + partition: &Struct, + spec_id: i32, + ) -> DataFile { + DataFileBuilder::default() + .file_path(format!("{}-deletes.puffin", Uuid::new_v4())) + .file_format(DataFileFormat::Puffin) + .content(DataContentType::PositionDeletes) + .record_count(1) + .referenced_data_file(Some(referenced_data_file.to_string())) + .content_offset(Some(4)) + .content_size_in_bytes(Some(40)) + .partition(partition.clone()) + .partition_spec_id(spec_id) + .file_size_in_bytes(60) + .build() + .unwrap() + } + fn build_unpartitioned_eq_delete() -> DataFile { build_partitioned_eq_delete(&Struct::empty(), 0) } diff --git a/crates/iceberg/src/delete_vector.rs b/crates/iceberg/src/delete_vector.rs index cfcc29700c..0274023653 100644 --- a/crates/iceberg/src/delete_vector.rs +++ b/crates/iceberg/src/delete_vector.rs @@ -73,7 +73,6 @@ impl DeleteVector { Ok(positions.len()) } - #[allow(unused)] pub fn len(&self) -> u64 { self.inner.len() } @@ -92,17 +91,12 @@ impl DeleteVector { /// format: a directory of 32-bit key / 32-bit roaring bitmap pairs, ordered by unsigned /// comparison of the keys, one bitmap per key. /// - /// Cardinality is not checked here. The caller validates the decoded length against the - /// delete file's `record_count`, where the manifest metadata is available. - /// /// # Errors /// /// Returns [`ErrorKind::DataInvalid`] if the blob is shorter than the minimum, the length /// prefix or CRC does not match, the magic is wrong, the roaring bitmap count exceeds the /// portable format's maximum, the roaring directory's keys are not ordered by unsigned /// comparison, or the roaring payload fails to decode. - // Consumed by the scan delete loader once the deletion-vector read path is wired up. - #[allow(dead_code)] pub fn deserialize(blob: &[u8]) -> Result { if blob.len() < DV_MIN_BLOB_BYTES { return Err(Error::new( @@ -316,6 +310,22 @@ impl BitOrAssign for DeleteVector { } } +// Reproduces Iceberg-Java's `deletion-vector-v1` framing so tests can round-trip through +// `deserialize` without a Java writer, and so other test modules can build blob fixtures. +// Cross-implementation golden fixtures produced by Iceberg-Java are tracked separately; this +// only checks that our decode matches our encode. +#[cfg(test)] +pub(crate) fn frame_dv_blob(vector: &[u8]) -> Vec { + let body_len = DV_MAGIC_BYTES + vector.len(); + let mut blob = Vec::with_capacity(DV_LENGTH_PREFIX_BYTES + body_len + DV_CRC_BYTES); + blob.extend_from_slice(&(body_len as u32).to_be_bytes()); + blob.extend_from_slice(&DV_MAGIC); + blob.extend_from_slice(vector); + let crc = crc32fast::hash(&blob[DV_LENGTH_PREFIX_BYTES..]); + blob.extend_from_slice(&crc.to_be_bytes()); + blob +} + #[cfg(test)] mod tests { use super::*; @@ -373,20 +383,6 @@ mod tests { assert!(res.is_err()); } - // Reproduces Iceberg-Java's `deletion-vector-v1` framing so tests can round-trip through - // `deserialize` without a Java writer. Cross-implementation golden fixtures produced by - // Iceberg-Java are tracked separately; this only checks that our decode matches our encode. - fn frame_dv_blob(vector: &[u8]) -> Vec { - let body_len = DV_MAGIC_BYTES + vector.len(); - let mut blob = Vec::with_capacity(DV_LENGTH_PREFIX_BYTES + body_len + DV_CRC_BYTES); - blob.extend_from_slice(&(body_len as u32).to_be_bytes()); - blob.extend_from_slice(&DV_MAGIC); - blob.extend_from_slice(vector); - let crc = crc32fast::hash(&blob[DV_LENGTH_PREFIX_BYTES..]); - blob.extend_from_slice(&crc.to_be_bytes()); - blob - } - fn encode_dv_blob(dv: &DeleteVector) -> Vec { let mut vector = Vec::with_capacity(dv.inner.serialized_size()); dv.inner.serialize_into(&mut vector).unwrap(); diff --git a/crates/iceberg/src/scan/context.rs b/crates/iceberg/src/scan/context.rs index 67b8cde4d8..2b5e6f7fef 100644 --- a/crates/iceberg/src/scan/context.rs +++ b/crates/iceberg/src/scan/context.rs @@ -126,7 +126,7 @@ impl ManifestEntryContext { self.manifest_entry.data_file(), self.manifest_entry.sequence_number(), ) - .await; + .await?; Ok(FileScanTask::builder() .with_file_size_in_bytes(self.manifest_entry.file_size_in_bytes()) diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs index 50b7399cdb..24095705cd 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -211,6 +211,7 @@ impl From<&DeleteFileContext> for FileScanTaskDeleteFile { .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_record_count(Some(ctx.manifest_entry.data_file.record_count)) .with_key_metadata( ctx.manifest_entry .data_file @@ -263,8 +264,18 @@ pub struct FileScanTaskDeleteFile { #[builder(default)] pub content_size_in_bytes: Option, - /// Key metadata for encrypted delete files (Parquet Modular Encryption). - /// When present, the reader uses this to build `FileDecryptionProperties`. + /// The delete file's record count, from the manifest entry. For a deletion vector, this is + /// the cardinality of the bitmap, used to validate the decoded blob against the manifest. + #[serde(default)] + #[serde(skip_serializing_if = "Option::is_none")] + #[builder(default)] + pub record_count: Option, + + /// Key metadata for an encrypted delete file. When present, the reader uses this to + /// decrypt the file: for a Parquet equality or position delete file, this builds + /// `FileDecryptionProperties` (Parquet Modular Encryption); for a deletion vector, whose + /// Puffin file has no native encryption, this wraps the range read in an + /// `EncryptedInputFile` (AGS1 stream encryption). /// /// Same plaintext-DEK trust boundary as [`FileScanTask::key_metadata`]: /// this is serialized into the scan plan and crosses the planner -> worker diff --git a/crates/iceberg/src/test_utils.rs b/crates/iceberg/src/test_utils.rs index cdcade7299..cfe3aa0ad8 100644 --- a/crates/iceberg/src/test_utils.rs +++ b/crates/iceberg/src/test_utils.rs @@ -24,6 +24,8 @@ use std::sync::{Arc, OnceLock}; use arrow_array::RecordBatch; use expect_test::Expect; use itertools::Itertools; +#[cfg(test)] +use roaring::RoaringTreemap; use crate::TableIdent; #[cfg(test)] @@ -120,6 +122,19 @@ pub(crate) fn make_encryption_manager(table_key_id: &str) -> Arc) -> Vec { + let mut bitmap = RoaringTreemap::new(); + for pos in positions { + bitmap.insert(pos); + } + let mut vector = Vec::new(); + bitmap.serialize_into(&mut vector).unwrap(); + crate::delete_vector::frame_dv_blob(&vector) +} + /// Build a table backed by the V3 encryption fixture and an in-memory KMS, /// so it has an [`EncryptionManager`](crate::encryption::EncryptionManager). /// From 22899c920786e16aa2b7e0f415eb33a41435193f Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Thu, 27 Aug 2026 10:51:28 -0400 Subject: [PATCH 7/9] identify deletion vectors by content type and file format --- .../src/arrow/caching_delete_file_loader.rs | 191 +++++++++--------- .../iceberg/src/arrow/delete_file_loader.rs | 6 +- crates/iceberg/src/arrow/delete_filter.rs | 4 + .../src/arrow/reader/positional_deletes.rs | 5 + crates/iceberg/src/arrow/reader/row_filter.rs | 1 + crates/iceberg/src/delete_file_index.rs | 59 +++++- crates/iceberg/src/scan/task.rs | 5 + 7 files changed, 163 insertions(+), 108 deletions(-) diff --git a/crates/iceberg/src/arrow/caching_delete_file_loader.rs b/crates/iceberg/src/arrow/caching_delete_file_loader.rs index 8cc73e5293..82eb9d884a 100644 --- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs +++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs @@ -35,9 +35,9 @@ use crate::io::FileIO; use crate::runtime::Runtime; use crate::scan::{ArrowRecordBatchStream, FileScanTaskDeleteFile}; use crate::spec::{ - DataContentType, Datum, ListType, MapType, NestedField, NestedFieldRef, PartnerAccessor, - PrimitiveType, Schema, SchemaRef, SchemaWithPartnerVisitor, StructType, Type, VariantType, - visit_schema_with_partner, + DataContentType, DataFileFormat, Datum, ListType, MapType, NestedField, NestedFieldRef, + PartnerAccessor, PrimitiveType, Schema, SchemaRef, SchemaWithPartnerVisitor, StructType, Type, + VariantType, visit_schema_with_partner, }; use crate::{Error, ErrorKind, Result}; @@ -70,7 +70,7 @@ enum DeleteFileContext { DelVec { data_file_path: String, blob: Bytes, - record_count: Option, + record_count: u64, dv_path: String, }, } @@ -264,15 +264,10 @@ impl CachingDeleteFileLoader { ) -> Result { match task.file_type { DataContentType::PositionDeletes => { - // A V3 deletion vector arrives as a PositionDeletes entry whose deletes live in a - // Puffin blob located by content_offset, not in a positional-delete parquet file. - if let Some(content_offset) = task.content_offset { - return Self::load_deletion_vector( - task, - content_offset, - basic_delete_file_loader, - ) - .await; + // 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 { + return Self::load_deletion_vector(task, basic_delete_file_loader).await; } match del_filter.try_start_pos_del_load(&task.file_path) { @@ -335,12 +330,13 @@ impl CachingDeleteFileLoader { } } - /// Validates a `PositionDeletes` task's deletion-vector coordinates and returns them as - /// typed values ready for the range read: `(start, len, referenced data file path)`. + /// 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` and `content_size_in_bytes` whenever - /// `content_offset` is set (a deletion vector), so their absence is a manifest-entry - /// inconsistency rather than an I/O failure. + /// 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 @@ -350,13 +346,21 @@ impl CachingDeleteFileLoader { /// `BitmapPositionDeleteIndex.deserializeBitmap`. fn validate_deletion_vector_task( task: &FileScanTaskDeleteFile, - content_offset: i64, - ) -> Result<(u64, u64, String)> { + ) -> 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 {} sets content_offset but not content_size_in_bytes", + "deletion vector {} is missing content_size_in_bytes", task.file_path ), ) @@ -370,6 +374,12 @@ impl CachingDeleteFileLoader { ), ) })?; + 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( @@ -390,21 +400,16 @@ impl CachingDeleteFileLoader { ) })?; - Ok((start, len, data_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`. - /// `record_count` is only absent for a scan task built without a manifest entry (e.g. in a - /// test fixture), in which case there is nothing to validate against. fn validate_deletion_vector_cardinality( delete_vector: &DeleteVector, - expected: Option, + expected: u64, dv_path: &str, ) -> Result<()> { - let Some(expected) = expected else { - return Ok(()); - }; let actual = delete_vector.len(); if actual != expected { return Err(Error::new( @@ -429,11 +434,9 @@ impl CachingDeleteFileLoader { /// `EncryptedInputFile` reads over. async fn load_deletion_vector( task: &FileScanTaskDeleteFile, - content_offset: i64, basic_delete_file_loader: BasicDeleteFileLoader, ) -> Result { - let (start, len, data_file_path) = - Self::validate_deletion_vector_task(task, content_offset)?; + let (start, len, data_file_path, record_count) = Self::validate_deletion_vector_task(task)?; let input_file = basic_delete_file_loader .file_io() @@ -453,7 +456,7 @@ impl CachingDeleteFileLoader { Ok(DeleteFileContext::DelVec { data_file_path, blob, - record_count: task.record_count, + record_count, dv_path: task.file_path.clone(), }) } @@ -1418,6 +1421,7 @@ mod tests { .with_file_path(pos_del_path.clone()) .with_file_size_in_bytes(std::fs::metadata(&pos_del_path).unwrap().len()) .with_file_type(DataContentType::PositionDeletes) + .with_file_format(DataFileFormat::Parquet) .with_partition_spec_id(0) .build(); @@ -1425,6 +1429,7 @@ mod tests { .with_file_path(eq_delete_path.clone()) .with_file_size_in_bytes(std::fs::metadata(&eq_delete_path).unwrap().len()) .with_file_type(DataContentType::EqualityDeletes) + .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(); @@ -1562,6 +1567,7 @@ mod tests { .with_file_path(dv_path) .with_file_size_in_bytes(file_size) .with_file_type(DataContentType::PositionDeletes) + .with_file_format(DataFileFormat::Puffin) .with_partition_spec_id(0) .with_referenced_data_file(Some(data_file_path)) .with_content_offset(Some(content_offset)) @@ -1715,91 +1721,87 @@ 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 task = FileScanTaskDeleteFile::builder() - .with_file_path("deletes.puffin".to_string()) - .with_file_size_in_bytes(100) - .with_file_type(DataContentType::PositionDeletes) - .with_partition_spec_id(0) - .with_referenced_data_file(Some("data.parquet".to_string())) - .build(); + let mut task = valid_dv_task(); + task.content_size_in_bytes = None; - let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task, 4).unwrap_err(); + let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task).unwrap_err(); assert_eq!(err.kind(), ErrorKind::DataInvalid); - assert!(err.message().contains("content_size_in_bytes")); + assert!(err.message().contains("missing content_size_in_bytes")); } #[test] fn test_validate_deletion_vector_task_rejects_missing_referenced_data_file() { - let task = FileScanTaskDeleteFile::builder() - .with_file_path("deletes.puffin".to_string()) - .with_file_size_in_bytes(100) - .with_file_type(DataContentType::PositionDeletes) - .with_partition_spec_id(0) - .with_content_size_in_bytes(Some(40)) - .build(); + let mut task = valid_dv_task(); + task.referenced_data_file = None; - let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task, 4).unwrap_err(); + let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task).unwrap_err(); assert_eq!(err.kind(), ErrorKind::DataInvalid); - assert!(err.message().contains("referenced_data_file")); + 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 task = FileScanTaskDeleteFile::builder() - .with_file_path("deletes.puffin".to_string()) - .with_file_size_in_bytes(100) - .with_file_type(DataContentType::PositionDeletes) - .with_partition_spec_id(0) - .with_referenced_data_file(Some("data.parquet".to_string())) - .with_content_size_in_bytes(Some(40)) - .build(); + let mut task = valid_dv_task(); + task.content_offset = Some(-1); - let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task, -1).unwrap_err(); + 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 task = FileScanTaskDeleteFile::builder() - .with_file_path("deletes.puffin".to_string()) - .with_file_size_in_bytes(100) - .with_file_type(DataContentType::PositionDeletes) - .with_partition_spec_id(0) - .with_referenced_data_file(Some("data.parquet".to_string())) - .with_content_size_in_bytes(Some(-1)) - .build(); + let mut task = valid_dv_task(); + task.content_size_in_bytes = Some(-1); - let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task, 4).unwrap_err(); + 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 task = FileScanTaskDeleteFile::builder() - .with_file_path("deletes.puffin".to_string()) - .with_file_size_in_bytes(100) - .with_file_type(DataContentType::PositionDeletes) - .with_partition_spec_id(0) - .with_referenced_data_file(Some("data.parquet".to_string())) - .with_content_size_in_bytes(Some(40)) - .build(); - - let (start, len, data_file_path) = - CachingDeleteFileLoader::validate_deletion_vector_task(&task, 4).unwrap(); + 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"); - } - - #[test] - fn test_validate_deletion_vector_cardinality_ignores_missing_record_count() { - let dv = DeleteVector::default(); - CachingDeleteFileLoader::validate_deletion_vector_cardinality(&dv, None, "deletes.puffin") - .expect("no record_count means nothing to validate"); + assert_eq!(record_count, 2); } #[test] @@ -1808,12 +1810,8 @@ mod tests { dv.insert(1); dv.insert(2); - CachingDeleteFileLoader::validate_deletion_vector_cardinality( - &dv, - Some(2), - "deletes.puffin", - ) - .unwrap(); + CachingDeleteFileLoader::validate_deletion_vector_cardinality(&dv, 2, "deletes.puffin") + .unwrap(); } #[test] @@ -1821,12 +1819,9 @@ mod tests { let mut dv = DeleteVector::default(); dv.insert(1); - let err = CachingDeleteFileLoader::validate_deletion_vector_cardinality( - &dv, - Some(2), - "deletes.puffin", - ) - .unwrap_err(); + let err = + CachingDeleteFileLoader::validate_deletion_vector_cardinality(&dv, 2, "deletes.puffin") + .unwrap_err(); assert_eq!(err.kind(), ErrorKind::DataInvalid); assert!(err.message().contains("expected 2 from record_count")); } diff --git a/crates/iceberg/src/arrow/delete_file_loader.rs b/crates/iceberg/src/arrow/delete_file_loader.rs index 6a232cea0d..26b73b6193 100644 --- a/crates/iceberg/src/arrow/delete_file_loader.rs +++ b/crates/iceberg/src/arrow/delete_file_loader.rs @@ -182,7 +182,7 @@ mod tests { use crate::arrow::delete_filter::tests::create_pos_del_schema; use crate::encryption::StandardKeyMetadata; use crate::scan::FileScanTaskDeleteFile; - use crate::spec::DataContentType; + use crate::spec::{DataContentType, DataFileFormat}; let encryption_key = b"0123456789abcdef"; let aad_prefix = b"aad_prefix"; @@ -234,6 +234,7 @@ mod tests { file_path: del_path.clone(), file_size_in_bytes: std::fs::metadata(&del_path).unwrap().len(), file_type: DataContentType::PositionDeletes, + file_format: DataFileFormat::Parquet, partition_spec_id: 0, equality_ids: None, key_metadata: Some(Box::from(key_metadata.as_ref())), @@ -266,7 +267,7 @@ mod tests { use crate::encryption::StandardKeyMetadata; use crate::scan::FileScanTaskDeleteFile; - use crate::spec::DataContentType; + use crate::spec::{DataContentType, DataFileFormat}; let encryption_key = b"0123456789abcdef"; let aad_prefix = b"my-table-uuid!!"; @@ -312,6 +313,7 @@ mod tests { file_path: del_path.clone(), file_size_in_bytes: std::fs::metadata(&del_path).unwrap().len(), file_type: DataContentType::EqualityDeletes, + file_format: DataFileFormat::Parquet, partition_spec_id: 0, equality_ids: Some(vec![1]), key_metadata: Some(Box::from(key_metadata.as_ref())), diff --git a/crates/iceberg/src/arrow/delete_filter.rs b/crates/iceberg/src/arrow/delete_filter.rs index bcfcbd233d..48c82c4118 100644 --- a/crates/iceberg/src/arrow/delete_filter.rs +++ b/crates/iceberg/src/arrow/delete_filter.rs @@ -435,6 +435,7 @@ pub(crate) mod tests { .len(), ) .with_file_type(DataContentType::PositionDeletes) + .with_file_format(DataFileFormat::Parquet) .with_partition_spec_id(0) .build(); @@ -452,6 +453,7 @@ pub(crate) mod tests { .len(), ) .with_file_type(DataContentType::PositionDeletes) + .with_file_format(DataFileFormat::Parquet) .with_partition_spec_id(0) .build(); @@ -469,6 +471,7 @@ pub(crate) mod tests { .len(), ) .with_file_type(DataContentType::PositionDeletes) + .with_file_format(DataFileFormat::Parquet) .with_partition_spec_id(0) .build(); @@ -543,6 +546,7 @@ pub(crate) mod tests { .with_file_path("eq-del.parquet".to_string()) .with_file_size_in_bytes(1) // never read; this test fails before opening the file .with_file_type(DataContentType::EqualityDeletes) + .with_file_format(DataFileFormat::Parquet) .with_partition_spec_id(0) .build(), ]) diff --git a/crates/iceberg/src/arrow/reader/positional_deletes.rs b/crates/iceberg/src/arrow/reader/positional_deletes.rs index e03e6c6093..6e7efeccc8 100644 --- a/crates/iceberg/src/arrow/reader/positional_deletes.rs +++ b/crates/iceberg/src/arrow/reader/positional_deletes.rs @@ -450,6 +450,7 @@ mod tests { .with_file_size_in_bytes(std::fs::metadata(&delete_file_path).unwrap().len()) .with_file_path(delete_file_path) .with_file_type(DataContentType::PositionDeletes) + .with_file_format(DataFileFormat::Parquet) .with_partition_spec_id(0) .build(), ]) @@ -668,6 +669,7 @@ mod tests { .with_file_size_in_bytes(std::fs::metadata(&delete_file_path).unwrap().len()) .with_file_path(delete_file_path) .with_file_type(DataContentType::PositionDeletes) + .with_file_format(DataFileFormat::Parquet) .with_partition_spec_id(0) .build(), ]) @@ -880,6 +882,7 @@ mod tests { .with_file_size_in_bytes(std::fs::metadata(&delete_file_path).unwrap().len()) .with_file_path(delete_file_path) .with_file_type(DataContentType::PositionDeletes) + .with_file_format(DataFileFormat::Parquet) .with_partition_spec_id(0) .build(), ]) @@ -993,6 +996,7 @@ mod tests { .with_file_path(dv_path.clone()) .with_file_size_in_bytes(std::fs::metadata(&dv_path).unwrap().len()) .with_file_type(DataContentType::PositionDeletes) + .with_file_format(DataFileFormat::Puffin) .with_partition_spec_id(0) .with_referenced_data_file(Some(data_file_path.clone())) .with_content_offset(Some(content_offset)) @@ -1089,6 +1093,7 @@ mod tests { .with_file_path(dv_path.clone()) .with_file_size_in_bytes(std::fs::metadata(&dv_path).unwrap().len()) .with_file_type(DataContentType::PositionDeletes) + .with_file_format(DataFileFormat::Puffin) .with_partition_spec_id(0) .with_referenced_data_file(Some(data_file_path.clone())) .with_content_offset(Some(0)) diff --git a/crates/iceberg/src/arrow/reader/row_filter.rs b/crates/iceberg/src/arrow/reader/row_filter.rs index 8c7b00aef3..fa083ca085 100644 --- a/crates/iceberg/src/arrow/reader/row_filter.rs +++ b/crates/iceberg/src/arrow/reader/row_filter.rs @@ -1243,6 +1243,7 @@ mod tests { deletes: vec![FileScanTaskDeleteFile { file_path: pos_del_path.clone(), file_type: DataContentType::PositionDeletes, + file_format: DataFileFormat::Parquet, partition_spec_id: 0, equality_ids: None, file_size_in_bytes: std::fs::metadata(&pos_del_path).unwrap().len(), diff --git a/crates/iceberg/src/delete_file_index.rs b/crates/iceberg/src/delete_file_index.rs index 6c62a77056..4c7ce25d09 100644 --- a/crates/iceberg/src/delete_file_index.rs +++ b/crates/iceberg/src/delete_file_index.rs @@ -26,7 +26,7 @@ use tokio::sync::Notify; use crate::metadata_columns::RESERVED_FIELD_ID_DELETE_FILE_PATH; use crate::runtime::Runtime; use crate::scan::{DeleteFileContext, FileScanTaskDeleteFile}; -use crate::spec::{DataContentType, DataFile, PrimitiveLiteral, Struct}; +use crate::spec::{DataContentType, DataFile, DataFileFormat, PrimitiveLiteral, Struct}; use crate::{Error, ErrorKind, Result}; /// Index of delete files @@ -180,8 +180,8 @@ impl PopulatedDeleteFileIndex { /// Creates a new populated delete file index from a list of delete file contexts, which /// allows for fast lookup when determining which delete files apply to a given data file. /// - /// 1. A V3 deletion vector (a `PositionDeletes` entry with `content_offset` set) is indexed - /// by the `referenced_data_file` field, which the spec requires for deletion vectors. + /// 1. A V3 deletion vector (a `PositionDeletes` entry stored as `Puffin`) is indexed by the + /// `referenced_data_file` field, which the spec requires for deletion vectors. /// Fails if two deletion vectors reference the same data file: the spec allows at most /// one deletion vector per data file per snapshot. /// 2. Other position deletes that reference a single data file, either through the @@ -211,20 +211,35 @@ impl PopulatedDeleteFileIndex { match arc_ctx.manifest_entry.content_type() { DataContentType::PositionDeletes => { - if data_file.content_offset().is_some() { - // The spec requires referenced_data_file whenever content_offset is set - // (a deletion vector), so its absence here is a malformed manifest entry, - // not an ordinary position delete to fall back on. + // A deletion vector is a position delete stored as a Puffin blob. The file + // format is what distinguishes it from a position delete parquet file. + if data_file.file_format() == DataFileFormat::Puffin { + // The spec requires referenced_data_file, content_offset and + // content_size_in_bytes on a deletion vector, so a missing one is a + // malformed manifest entry, not an ordinary position delete to fall back + // on. let Some(path) = data_file.referenced_data_file() else { return Err(Error::new( ErrorKind::DataInvalid, format!( - "deletion vector {} sets content_offset but is missing referenced_data_file", + "deletion vector {} is missing referenced_data_file", arc_ctx.manifest_entry.file_path() ), )); }; + if data_file.content_offset().is_none() + || data_file.content_size_in_bytes().is_none() + { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "deletion vector {} is missing content_offset or content_size_in_bytes", + arc_ctx.manifest_entry.file_path() + ), + )); + } + if let Some(existing) = dvs_by_referenced_data_file.insert(path.clone(), arc_ctx) { @@ -1184,6 +1199,34 @@ mod tests { assert!(err.message().contains("missing referenced_data_file")); } + #[test] + fn test_deletion_vector_missing_coordinates_is_rejected() { + let malformed_dv = DataFileBuilder::default() + .file_path("deletes.puffin".to_string()) + .file_format(DataFileFormat::Puffin) + .content(DataContentType::PositionDeletes) + .record_count(1) + .referenced_data_file(Some("data.parquet".to_string())) + .content_size_in_bytes(Some(40)) + .partition(Struct::empty()) + .partition_spec_id(0) + .file_size_in_bytes(60) + .build() + .unwrap(); + + let contexts = vec![DeleteFileContext { + manifest_entry: build_added_manifest_entry(5, &malformed_dv).into(), + partition_spec_id: 0, + }]; + + let err = PopulatedDeleteFileIndex::new(contexts).unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!( + err.message() + .contains("missing content_offset or content_size_in_bytes") + ); + } + #[test] fn test_multiple_deletion_vectors_for_same_data_file_is_rejected() { let partition = Struct::from_iter([Some(Literal::long(100))]); diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs index 249e859c3a..57567d6665 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -206,6 +206,7 @@ impl From<&DeleteFileContext> for FileScanTaskDeleteFile { .with_file_path(ctx.manifest_entry.file_path().to_string()) .with_file_size_in_bytes(ctx.manifest_entry.file_size_in_bytes()) .with_file_type(ctx.manifest_entry.content_type()) + .with_file_format(ctx.manifest_entry.data_file().file_format()) .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()) @@ -236,6 +237,10 @@ pub struct FileScanTaskDeleteFile { /// delete file type pub file_type: DataContentType, + /// The delete file's format, from the manifest entry. A `PositionDeletes` entry written as + /// `Puffin` is a V3 deletion vector; one written as `Parquet` is a position delete file. + pub file_format: DataFileFormat, + /// partition id pub partition_spec_id: i32, From c9ddf66d4cb4d5869aec4e1fe64735359d650081 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Thu, 27 Aug 2026 10:59:46 -0400 Subject: [PATCH 8/9] Fix after merging upstream/main --- crates/iceberg/public-api.txt | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 19f6c26968..d775d8541f 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1313,6 +1313,7 @@ pub struct iceberg::scan::FileScanTaskDeleteFile pub iceberg::scan::FileScanTaskDeleteFile::content_offset: core::option::Option pub iceberg::scan::FileScanTaskDeleteFile::content_size_in_bytes: core::option::Option pub iceberg::scan::FileScanTaskDeleteFile::equality_ids: core::option::Option> +pub iceberg::scan::FileScanTaskDeleteFile::file_format: iceberg::spec::DataFileFormat 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 @@ -1328,7 +1329,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 From c8e6e8abc68dde3f2ccd37c3251302bb324b21bc Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Thu, 27 Aug 2026 11:11:42 -0400 Subject: [PATCH 9/9] Update comment wording. --- crates/iceberg/src/delete_file_index.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/iceberg/src/delete_file_index.rs b/crates/iceberg/src/delete_file_index.rs index 4c7ce25d09..2e749cab1f 100644 --- a/crates/iceberg/src/delete_file_index.rs +++ b/crates/iceberg/src/delete_file_index.rs @@ -1290,8 +1290,8 @@ mod tests { assert_eq!(err.kind(), ErrorKind::DataInvalid); } - // A V3 deletion vector: a PositionDeletes entry stored as a Puffin blob (content_offset set) - // scoped to a single data file via referenced_data_file. + // A V3 deletion vector: a PositionDeletes entry stored as a Puffin blob, scoped to a single + // data file via referenced_data_file. fn build_deletion_vector( referenced_data_file: &str, partition: &Struct,