feat(scan): [3/N] read and apply V3 deletion vectors (#3035)
* 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.
* fix after merging in main
* fix public-api.txt
* fix test name
* update
* apply V3 deletion vectors during scan
* identify deletion vectors by content type and file format
* Fix after merging upstream/main
* Update comment wording.
diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt
index 95df9e9..57edfd6 100644
--- a/crates/iceberg/public-api.txt
+++ b/crates/iceberg/public-api.txt
@@ -1313,6 +1313,7 @@
pub iceberg::scan::FileScanTaskDeleteFile::content_offset: core::option::Option<i64>
pub iceberg::scan::FileScanTaskDeleteFile::content_size_in_bytes: core::option::Option<i64>
pub iceberg::scan::FileScanTaskDeleteFile::equality_ids: core::option::Option<alloc::vec::Vec<i32>>
+pub iceberg::scan::FileScanTaskDeleteFile::file_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 @@
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 896248f..905fa86 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::sync::Arc;
use arrow_array::{Array, ArrayRef, Int64Array, StringArray, StructArray};
+use bytes::Bytes;
use futures::{StreamExt, TryStreamExt};
use tokio::sync::oneshot::{Receiver, channel};
@@ -27,15 +28,16 @@
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;
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};
@@ -51,7 +53,6 @@
// 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 @@
equality_ids: HashSet<i32>,
sender: tokio::sync::oneshot::Sender<Predicate>,
},
+ // 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: u64,
+ dv_path: String,
+ },
}
// Final result of the processing of a delete file task before
@@ -72,6 +82,11 @@
file_path: String,
results: HashMap<String, DeleteVector>,
},
+ // 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 @@
/// 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 @@
/// |
/// |
/// +-----------------------------+--------------------------+
- /// 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<String, RoaringTreeMap> HashMap<String, RoaringTreeMap> (Predicate, Sender)
+ /// HashMap<String, RoaringTreeMap> DeleteVector (Predicate, Sender)
/// | | |
/// | | [persist to state]
/// | | ()
@@ -211,13 +227,22 @@
.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,12 @@
) -> Result<DeleteFileContext> {
match task.file_type {
DataContentType::PositionDeletes => {
+ // A V3 deletion vector arrives as a PositionDeletes entry whose deletes live in
+ // a Puffin blob, not in a positional-delete parquet file.
+ if task.file_format == DataFileFormat::Puffin {
+ return Self::load_deletion_vector(task, 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 +330,137 @@
}
}
+ /// Validates a deletion-vector task and returns what the read needs as typed values:
+ /// `(start, len, referenced data file path, expected cardinality)`.
+ ///
+ /// The spec requires `referenced_data_file`, `content_offset` and `content_size_in_bytes` on
+ /// a deletion vector, and a deletion vector is always built from a manifest entry, so it
+ /// always carries `record_count`. A missing one is a manifest-entry inconsistency rather
+ /// than an I/O failure.
+ ///
+ /// Equality and ordinary position deletes have no equivalent validation in this loader: a
+ /// malformed equality/position delete file fails loudly when the Parquet reader can't open
+ /// it. A deletion vector's coordinates instead drive a raw byte-range read with no format
+ /// to fail against, so a bad coordinate would otherwise decode silently into the wrong (or
+ /// no) deletes, per the same corrupted-blob concern Iceberg-Java validates in
+ /// `BitmapPositionDeleteIndex.deserializeBitmap`.
+ fn validate_deletion_vector_task(
+ task: &FileScanTaskDeleteFile,
+ ) -> Result<(u64, u64, String, u64)> {
+ let content_offset = task.content_offset.ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "deletion vector {} is missing content_offset",
+ task.file_path
+ ),
+ )
+ })?;
+ let content_size = task.content_size_in_bytes.ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "deletion vector {} is missing content_size_in_bytes",
+ task.file_path
+ ),
+ )
+ })?;
+ let data_file_path = task.referenced_data_file.clone().ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "deletion vector {} is missing referenced_data_file",
+ task.file_path
+ ),
+ )
+ })?;
+ let record_count = task.record_count.ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!("deletion vector {} is missing record_count", task.file_path),
+ )
+ })?;
+
+ let start = u64::try_from(content_offset).map_err(|_| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "deletion vector {} has negative content_offset {content_offset}",
+ task.file_path
+ ),
+ )
+ })?;
+ let len = u64::try_from(content_size).map_err(|_| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "deletion vector {} has negative content_size_in_bytes {content_size}",
+ task.file_path
+ ),
+ )
+ })?;
+
+ Ok((start, len, data_file_path, record_count))
+ }
+
+ /// Validates a decoded deletion vector's cardinality against the manifest entry's
+ /// `record_count`, mirroring Iceberg-Java's `BitmapPositionDeleteIndex.deserializeBitmap`.
+ fn validate_deletion_vector_cardinality(
+ delete_vector: &DeleteVector,
+ expected: u64,
+ dv_path: &str,
+ ) -> Result<()> {
+ 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,
+ basic_delete_file_loader: BasicDeleteFileLoader,
+ ) -> Result<DeleteFileContext> {
+ let (start, len, data_file_path, record_count) = Self::validate_deletion_vector_task(task)?;
+
+ 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,
+ dv_path: task.file_path.clone(),
+ })
+ }
+
async fn parse_file_content_for_task(
ctx: DeleteFileContext,
) -> Result<ParsedDeleteFileContext> {
@@ -312,6 +474,20 @@
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,
@@ -700,6 +876,7 @@
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() {
@@ -1247,6 +1424,7 @@
.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();
@@ -1254,6 +1432,7 @@
.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();
@@ -1384,4 +1563,276 @@
// 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<Box<[u8]>>,
+ ) -> FileScanTaskDeleteFile {
+ FileScanTaskDeleteFile::builder()
+ .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))
+ .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<u64> = 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<u64> = 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"));
+ }
+
+ // A well-formed deletion-vector task, for tests that then clear or corrupt one field.
+ fn valid_dv_task() -> FileScanTaskDeleteFile {
+ dv_task(
+ "deletes.puffin".to_string(),
+ 100,
+ "data.parquet".to_string(),
+ 4,
+ 40,
+ 2,
+ None,
+ )
+ }
+
+ #[test]
+ fn test_validate_deletion_vector_task_rejects_missing_content_offset() {
+ let mut task = valid_dv_task();
+ task.content_offset = None;
+
+ let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task).unwrap_err();
+ assert_eq!(err.kind(), ErrorKind::DataInvalid);
+ assert!(err.message().contains("missing content_offset"));
+ }
+
+ #[test]
+ fn test_validate_deletion_vector_task_rejects_missing_content_size() {
+ let mut task = valid_dv_task();
+ task.content_size_in_bytes = None;
+
+ let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task).unwrap_err();
+ assert_eq!(err.kind(), ErrorKind::DataInvalid);
+ assert!(err.message().contains("missing content_size_in_bytes"));
+ }
+
+ #[test]
+ fn test_validate_deletion_vector_task_rejects_missing_referenced_data_file() {
+ let mut task = valid_dv_task();
+ task.referenced_data_file = None;
+
+ let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task).unwrap_err();
+ assert_eq!(err.kind(), ErrorKind::DataInvalid);
+ assert!(err.message().contains("missing referenced_data_file"));
+ }
+
+ #[test]
+ fn test_validate_deletion_vector_task_rejects_missing_record_count() {
+ let mut task = valid_dv_task();
+ task.record_count = None;
+
+ let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task).unwrap_err();
+ assert_eq!(err.kind(), ErrorKind::DataInvalid);
+ assert!(err.message().contains("missing record_count"));
+ }
+
+ #[test]
+ fn test_validate_deletion_vector_task_rejects_negative_content_offset() {
+ let mut task = valid_dv_task();
+ task.content_offset = Some(-1);
+
+ let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task).unwrap_err();
+ assert_eq!(err.kind(), ErrorKind::DataInvalid);
+ assert!(err.message().contains("negative content_offset"));
+ }
+
+ #[test]
+ fn test_validate_deletion_vector_task_rejects_negative_content_size() {
+ let mut task = valid_dv_task();
+ task.content_size_in_bytes = Some(-1);
+
+ let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task).unwrap_err();
+ assert_eq!(err.kind(), ErrorKind::DataInvalid);
+ assert!(err.message().contains("negative content_size_in_bytes"));
+ }
+
+ #[test]
+ fn test_validate_deletion_vector_task_accepts_valid_coordinates() {
+ let (start, len, data_file_path, record_count) =
+ CachingDeleteFileLoader::validate_deletion_vector_task(&valid_dv_task()).unwrap();
+ assert_eq!(start, 4);
+ assert_eq!(len, 40);
+ assert_eq!(data_file_path, "data.parquet");
+ assert_eq!(record_count, 2);
+ }
+
+ #[test]
+ fn test_validate_deletion_vector_cardinality_accepts_matching_count() {
+ let mut dv = DeleteVector::default();
+ dv.insert(1);
+ dv.insert(2);
+
+ CachingDeleteFileLoader::validate_deletion_vector_cardinality(&dv, 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, 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 13fa974..5852cf1 100644
--- a/crates/iceberg/src/arrow/delete_file_loader.rs
+++ b/crates/iceberg/src/arrow/delete_file_loader.rs
@@ -182,7 +182,7 @@
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 @@
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 @@
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 @@
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 fb67d02..cef81af 100644
--- a/crates/iceberg/src/arrow/delete_filter.rs
+++ b/crates/iceberg/src/arrow/delete_filter.rs
@@ -441,6 +441,7 @@
.len(),
)
.with_file_type(DataContentType::PositionDeletes)
+ .with_file_format(DataFileFormat::Parquet)
.with_partition_spec_id(0)
.build();
@@ -458,6 +459,7 @@
.len(),
)
.with_file_type(DataContentType::PositionDeletes)
+ .with_file_format(DataFileFormat::Parquet)
.with_partition_spec_id(0)
.build();
@@ -475,6 +477,7 @@
.len(),
)
.with_file_type(DataContentType::PositionDeletes)
+ .with_file_format(DataFileFormat::Parquet)
.with_partition_spec_id(0)
.build();
@@ -551,6 +554,7 @@
.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 a66d9da..f0c7bf7 100644
--- a/crates/iceberg/src/arrow/reader/positional_deletes.rs
+++ b/crates/iceberg/src/arrow/reader/positional_deletes.rs
@@ -172,6 +172,7 @@
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,
@@ -449,6 +450,7 @@
.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 +670,7 @@
.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(),
])
@@ -881,6 +884,7 @@
.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(),
])
@@ -926,4 +930,189 @@
"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_file_format(DataFileFormat::Puffin)
+ .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::<Vec<RecordBatch>>()
+ .await
+ .unwrap();
+
+ let ids: Vec<i32> = result
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column(0)
+ .as_primitive::<arrow_array::types::Int32Type>()
+ .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_file_format(DataFileFormat::Puffin)
+ .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<Vec<RecordBatch>, _> =
+ 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 2980392..389231c 100644
--- a/crates/iceberg/src/arrow/reader/row_filter.rs
+++ b/crates/iceberg/src/arrow/reader/row_filter.rs
@@ -1237,6 +1237,7 @@
.with_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 bab5893..2e749ca 100644
--- a/crates/iceberg/src/delete_file_index.rs
+++ b/crates/iceberg/src/delete_file_index.rs
@@ -26,7 +26,8 @@
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
#[derive(Debug, Clone)]
@@ -36,8 +37,18 @@
#[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<Notify>),
- 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<PopulatedDeleteFileIndex>),
+ // 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<Error>),
}
#[derive(Debug)]
@@ -46,7 +57,10 @@
eq_deletes_by_partition: HashMap<Struct, Vec<Arc<DeleteFileContext>>>,
pos_deletes_by_partition: HashMap<Struct, Vec<Arc<DeleteFileContext>>>,
pos_deletes_by_path: HashMap<String, Vec<Arc<DeleteFileContext>>>,
- // 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<String, Arc<DeleteFileContext>>,
}
/// Determines the single data file referenced by a position delete file, if any.
@@ -79,6 +93,13 @@
}
}
+/// 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<DeleteFileContext>) {
@@ -96,11 +117,14 @@
let delete_files: Vec<DeleteFileContext> =
delete_file_stream.collect::<Vec<_>>().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 @@
}
/// 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<i64>,
- ) -> Vec<FileScanTaskDeleteFile> {
+ ) -> Result<Vec<FileScanTaskDeleteFile>> {
// 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 @@
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 @@
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,83 @@
/// 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 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
/// `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<DeleteFileContext>) -> PopulatedDeleteFileIndex {
+ fn new(files: Vec<DeleteFileContext>) -> Result<PopulatedDeleteFileIndex> {
let mut eq_deletes_by_partition: HashMap<Struct, Vec<Arc<DeleteFileContext>>> =
HashMap::default();
let mut pos_deletes_by_partition: HashMap<Struct, Vec<Arc<DeleteFileContext>>> =
HashMap::default();
let mut pos_deletes_by_path: HashMap<String, Vec<Arc<DeleteFileContext>>> =
HashMap::default();
+ let mut dvs_by_referenced_data_file: HashMap<String, Arc<DeleteFileContext>> =
+ HashMap::default();
let mut global_equality_deletes: Vec<Arc<DeleteFileContext>> = 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()) {
+ // 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 {} 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)
+ {
+ 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 +278,29 @@
}
_ => 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<i64>,
- ) -> Vec<FileScanTaskDeleteFile> {
+ ) -> Result<Vec<FileScanTaskDeleteFile>> {
let mut results = vec![];
self.global_equality_deletes
@@ -233,6 +326,49 @@
.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 +397,7 @@
.for_each(|delete| results.push(delete.as_ref().into()));
}
- results
+ Ok(results)
}
}
@@ -297,23 +433,26 @@
})
.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<String> = delete_files_to_apply_for_seq_4
.into_iter()
.map(|file| file.file_path)
@@ -325,8 +464,9 @@
);
// 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<String> = delete_files_to_apply_for_seq_5
.into_iter()
.map(|file| file.file_path)
@@ -337,8 +477,9 @@
);
// 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<String> = delete_files_to_apply_for_seq_6
.into_iter()
.map(|file| file.file_path)
@@ -352,8 +493,9 @@
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<String> =
delete_files_to_apply_for_partitioned_file
.into_iter()
@@ -389,24 +531,27 @@
})
.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<String> = delete_files_to_apply_for_seq_4
.into_iter()
.map(|file| file.file_path)
@@ -418,8 +563,9 @@
);
// 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<String> = delete_files_to_apply_for_seq_5
.into_iter()
.map(|file| file.file_path)
@@ -430,8 +576,9 @@
);
// 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<String> = delete_files_to_apply_for_seq_6
.into_iter()
.map(|file| file.file_path)
@@ -444,8 +591,9 @@
// 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<String> =
delete_files_to_apply_for_different_partition
.into_iter()
@@ -455,8 +603,9 @@
// 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<String> =
delete_files_to_apply_for_different_spec
.into_iter()
@@ -474,15 +623,20 @@
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 +652,20 @@
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 +682,21 @@
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 +712,14 @@
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 +732,39 @@
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 +791,12 @@
manifest_entry: build_added_manifest_entry(5, &eq_delete).into(),
partition_spec_id: 0,
},
- ]);
+ ])
+ .unwrap();
let mut actual_paths: Vec<String> = index
.get_deletes_for_data_file(&data_file, Some(0))
+ .unwrap()
.into_iter()
.map(|delete| delete.file_path)
.collect();
@@ -642,10 +828,12 @@
manifest_entry: build_added_manifest_entry(6, &second_delete).into(),
partition_spec_id: 0,
},
- ]);
+ ])
+ .unwrap();
let mut actual_paths: Vec<String> = index
.get_deletes_for_data_file(&data_file, Some(0))
+ .unwrap()
.into_iter()
.map(|delete| delete.file_path)
.collect();
@@ -679,15 +867,19 @@
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 +898,20 @@
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
);
@@ -791,6 +988,330 @@
);
}
+ #[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<String> = 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<String> = 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<String> = 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_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))]);
+ 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, 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 e938ce5..4ab13d5 100644
--- a/crates/iceberg/src/delete_vector.rs
+++ b/crates/iceberg/src/delete_vector.rs
@@ -72,7 +72,6 @@
Ok(positions.len())
}
- #[allow(unused)]
pub fn len(&self) -> u64 {
self.inner.len()
}
@@ -91,17 +90,12 @@
/// 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<Self> {
if blob.len() < DV_MIN_BLOB_BYTES {
return Err(Error::new(
@@ -315,6 +309,22 @@
}
}
+// 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<u8> {
+ 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::*;
@@ -372,20 +382,6 @@
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<u8> {
- 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<u8> {
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 6768fee..a598975 100644
--- a/crates/iceberg/src/scan/context.rs
+++ b/crates/iceberg/src/scan/context.rs
@@ -126,7 +126,7 @@
self.manifest_entry.data_file(),
self.manifest_entry.sequence_number(),
)
- .await;
+ .await?;
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 647fb39..3aff0f4 100644
--- a/crates/iceberg/src/scan/task.rs
+++ b/crates/iceberg/src/scan/task.rs
@@ -318,6 +318,7 @@
.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())
@@ -348,6 +349,10 @@
/// 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,
@@ -383,8 +388,11 @@
#[builder(default)]
pub record_count: Option<u64>,
- /// Key metadata for encrypted delete files (Parquet Modular Encryption).
- /// When present, the reader uses this to build `FileDecryptionProperties`.
+ /// 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 63f4011..e80a51d 100644
--- a/crates/iceberg/src/test_utils.rs
+++ b/crates/iceberg/src/test_utils.rs
@@ -24,6 +24,8 @@
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 @@
)
}
+/// Encodes a `deletion-vector-v1` Puffin blob for the given positions, matching the framing in
+/// [`DeleteVector::deserialize`](crate::delete_vector::DeleteVector::deserialize).
+#[cfg(test)]
+pub(crate) fn encode_dv_blob(positions: impl IntoIterator<Item = u64>) -> Vec<u8> {
+ 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).
///
