You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by gi...@apache.org on 2026/10/09 09:53:21 UTC
(iceberg-rust) branch gh-readonly-queue/main/pr-3145-49b91d1c9508758f116bdc6947d2f2cba9f9980b created (now 45e2b3199)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-3145-49b91d1c9508758f116bdc6947d2f2cba9f9980b
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
at 45e2b3199 fix(scan): validate deletion vector delete tasks (#3145)
This branch includes the following new commits:
new 45e2b3199 fix(scan): validate deletion vector delete tasks (#3145)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iceberg-rust) 01/01: fix(scan): validate deletion vector delete tasks (#3145)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-3145-49b91d1c9508758f116bdc6947d2f2cba9f9980b
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit 45e2b31991daf728a052cf77dabe92a50ebc80b0
Author: hylin <li...@gmail.com>
AuthorDate: Fri Oct 9 09:15:03 2026 +0000
fix(scan): validate deletion vector delete tasks (#3145)
* docs: design delete task validation
* fix(scan): validate delete task construction
* refactor(scan): use validated delete task builder
* refactor(scan): centralize deletion vector validation
* style(scan): format validated delete task changes
* fix(scan): satisfy public API and license checks
* test(scan): adapt merged reader fixtures
* refactor(scan): encapsulate delete file metadata
* fix(scan): propagate invalid delete task errors
* fix(scan): unwrap the delete-file builder in the scan test
`FileScanTaskDeleteFile::builder().build()` returns `Result` since this
branch made it validate deletion-vector metadata, and this call site was
missed, so `cargo check --all-targets` failed with E0308 once the branch
was rebased onto a main that also compiles the lib test target.
The task built here is a complete equality delete, so validation cannot
fail; `.unwrap()` matches how the other tests in this file build tasks.
* refactor(scan): make delete-file byte offsets unsigned
Addresses @blackmwk's review. content_offset and content_size_in_bytes
are byte quantities, so Option<u64> makes a negative unrepresentable
instead of something every layer re-checks.
That collapses the three findings into one change:
- 'Why not change it to u64?' / 'Ditto': the getters and fields are u64.
- 'This duplicates a lot.': the duplication existed only to build the
'deletion vector' vs 'delete file' prefix for two near-identical
negative-value errors. Both are gone, and the remaining presence
checks collapse to one lookup plus one error site.
- 'I think this should also apply to non deletion vector?': it now does,
and unconditionally. The i64 -> u64 conversion happens once in
TryFrom<&DeleteFileContext>, where the manifest value enters, so every
delete-file kind is covered rather than only Puffin position deletes.
Also addresses @mbutrovich's deserialization concern from the same
review: FileScanTaskDeleteFile derives plain Deserialize, so a scan plan
rebuilt on a worker never passes through build(). With u64 fields, serde
itself rejects a negative offset, so a hand-written or corrupted plan
fails before any loader code runs -- stronger than the runtime check it
replaces.
Downstream, validate_deletion_vector_task loses its two
u64::try_from conversions; the values are already u64.
Tests: the negative cases moved from the builder to the manifest
boundary (test_delete_context_rejects_negative_*), including the non-DV
case, and the two loader tests now assert serde refuses the value.
cargo test -p iceberg --lib: 1683 passed, 0 failed.
cargo fmt --all -- --check and cargo clippy -p iceberg --all-targets
-D warnings: clean. public-api.txt updated for the two getters.
Signed-off-by: linhongyu510 <li...@gmail.com>
* refactor(scan): validate delete files on deserialize and enforce size pairing
Address review follow-ups on the delete-file validation design:
- Route FileScanTaskDeleteFile deserialization through a FileScanTaskDeleteFileSerde
mirror that builds via builder().build(), so a deletion vector rebuilt from a
serialized scan plan (planner->worker path) passes validate() like a manifest-
built task. A malformed plan now fails at deserialize time instead of reaching a
loader expect/ok_or_else.
- Enforce the spec pairing rule for every delete-file kind: content_size_in_bytes
is required whenever content_offset is present, not only for deletion vectors.
- Replace two builder-only duplicate tests with one driving the missing-coordinate
check through get_deletes_for_data_file, where the manifest entry is converted.
- Reword PR-history-laden comments to describe the code as it is now.
cargo test -p iceberg --lib: 1683 passed, 0 failed.
* docs(scan): clarify deletion vector loader extraction
---------
Signed-off-by: linhongyu510 <li...@gmail.com>
Co-authored-by: linhongyu510 <li...@users.noreply.github.com>
Co-authored-by: linhongyu510 <ma...@gmail.com>
---
crates/iceberg/public-api.txt | 25 +-
.../src/arrow/caching_delete_file_loader.rs | 207 ++++-----
crates/iceberg/src/arrow/delete_file_loader.rs | 55 ++-
crates/iceberg/src/arrow/delete_filter.rs | 18 +-
crates/iceberg/src/arrow/reader/pipeline.rs | 5 +-
.../iceberg/src/arrow/reader/positional_deletes.rs | 31 +-
crates/iceberg/src/arrow/reader/row_filter.rs | 23 +-
crates/iceberg/src/cow_rewrite/plan.rs | 2 +-
crates/iceberg/src/delete_file_index.rs | 154 +++----
crates/iceberg/src/scan/mod.rs | 5 +-
crates/iceberg/src/scan/task.rs | 468 ++++++++++++++++++++-
11 files changed, 707 insertions(+), 286 deletions(-)
diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt
index bb4d350e5..bd5539534 100644
--- a/crates/iceberg/public-api.txt
+++ b/crates/iceberg/public-api.txt
@@ -1367,21 +1367,24 @@ pub fn iceberg::scan::FileScanTask::serialize<S>(&self, serializer: S) -> core::
impl<'de> serde_core::de::Deserialize<'de> for iceberg::scan::FileScanTask
pub fn iceberg::scan::FileScanTask::deserialize<__D>(__deserializer: __D) -> core::result::Result<Self, <__D as serde_core::de::Deserializer>::Error> where __D: serde_core::de::Deserializer<'de>
pub struct iceberg::scan::FileScanTaskDeleteFile
-pub iceberg::scan::FileScanTaskDeleteFile::content_offset: core::option::Option<i64>
-pub iceberg::scan::FileScanTaskDeleteFile::content_size_in_bytes: core::option::Option<i64>
-pub iceberg::scan::FileScanTaskDeleteFile::equality_ids: core::option::Option<alloc::vec::Vec<i32>>
-pub iceberg::scan::FileScanTaskDeleteFile::file_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
-pub iceberg::scan::FileScanTaskDeleteFile::key_metadata: core::option::Option<alloc::boxed::Box<[u8]>>
-pub iceberg::scan::FileScanTaskDeleteFile::partition_spec_id: i32
-pub iceberg::scan::FileScanTaskDeleteFile::record_count: core::option::Option<u64>
-pub iceberg::scan::FileScanTaskDeleteFile::referenced_data_file: core::option::Option<alloc::string::String>
+impl iceberg::scan::FileScanTaskDeleteFile
+pub fn iceberg::scan::FileScanTaskDeleteFile::content_offset(&self) -> core::option::Option<u64>
+pub fn iceberg::scan::FileScanTaskDeleteFile::content_size_in_bytes(&self) -> core::option::Option<u64>
+pub fn iceberg::scan::FileScanTaskDeleteFile::equality_ids(&self) -> core::option::Option<&[i32]>
+pub fn iceberg::scan::FileScanTaskDeleteFile::file_format(&self) -> iceberg::spec::DataFileFormat
+pub fn iceberg::scan::FileScanTaskDeleteFile::file_path(&self) -> &str
+pub fn iceberg::scan::FileScanTaskDeleteFile::file_size_in_bytes(&self) -> u64
+pub fn iceberg::scan::FileScanTaskDeleteFile::file_type(&self) -> iceberg::spec::DataContentType
+pub fn iceberg::scan::FileScanTaskDeleteFile::key_metadata(&self) -> core::option::Option<&[u8]>
+pub fn iceberg::scan::FileScanTaskDeleteFile::partition_spec_id(&self) -> i32
+pub fn iceberg::scan::FileScanTaskDeleteFile::record_count(&self) -> core::option::Option<u64>
+pub fn iceberg::scan::FileScanTaskDeleteFile::referenced_data_file(&self) -> core::option::Option<&str>
impl core::clone::Clone for iceberg::scan::FileScanTaskDeleteFile
pub fn iceberg::scan::FileScanTaskDeleteFile::clone(&self) -> iceberg::scan::FileScanTaskDeleteFile
impl core::cmp::PartialEq for iceberg::scan::FileScanTaskDeleteFile
pub fn iceberg::scan::FileScanTaskDeleteFile::eq(&self, other: &iceberg::scan::FileScanTaskDeleteFile) -> bool
+impl core::convert::From<iceberg::scan::FileScanTaskDeleteFile> for iceberg::Result<iceberg::scan::FileScanTaskDeleteFile>
+pub fn iceberg::Result<iceberg::scan::FileScanTaskDeleteFile>::from(task: iceberg::scan::FileScanTaskDeleteFile) -> Self
impl core::fmt::Debug for iceberg::scan::FileScanTaskDeleteFile
pub fn iceberg::scan::FileScanTaskDeleteFile::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
impl core::marker::StructuralPartialEq for iceberg::scan::FileScanTaskDeleteFile
diff --git a/crates/iceberg/src/arrow/caching_delete_file_loader.rs b/crates/iceberg/src/arrow/caching_delete_file_loader.rs
index e1cb6de15..5582e4397 100644
--- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs
+++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs
@@ -263,15 +263,15 @@ impl CachingDeleteFileLoader {
del_filter: DeleteFilter,
schema: SchemaRef,
) -> Result<DeleteFileContext> {
- match task.file_type {
+ 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 {
+ 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) {
+ match del_filter.try_start_pos_del_load(task.file_path()) {
PosDelLoadAction::AlreadyLoaded => Ok(DeleteFileContext::ExistingPosDel),
PosDelLoadAction::WaitFor(notified) => {
// Positional deletes are accessed synchronously by ArrowReader.
@@ -281,12 +281,12 @@ impl CachingDeleteFileLoader {
Ok(DeleteFileContext::ExistingPosDel)
}
PosDelLoadAction::Load => Ok(DeleteFileContext::PosDels {
- file_path: task.file_path.clone(),
+ file_path: task.file_path().to_string(),
stream: basic_delete_file_loader
.parquet_to_batch_stream(
- &task.file_path,
- task.file_size_in_bytes,
- task.key_metadata.as_deref(),
+ task.file_path(),
+ task.file_size_in_bytes(),
+ task.key_metadata(),
)
.await?,
}),
@@ -294,22 +294,22 @@ impl CachingDeleteFileLoader {
}
DataContentType::EqualityDeletes => {
- let Some(notify) = del_filter.try_start_eq_del_load(&task.file_path) else {
+ let Some(notify) = del_filter.try_start_eq_del_load(task.file_path()) else {
return Ok(DeleteFileContext::ExistingEqDel);
};
let (sender, receiver) = channel();
- del_filter.insert_equality_delete(&task.file_path, receiver);
+ del_filter.insert_equality_delete(task.file_path(), receiver);
// Per the Iceberg spec, evolve schema for equality deletes but only for the
// equality_ids columns, not all table columns.
- let equality_ids_vec = task.equality_ids.clone().unwrap();
+ let equality_ids_vec = task.equality_ids().unwrap().to_vec();
let evolved_stream = BasicDeleteFileLoader::evolve_schema(
basic_delete_file_loader
.parquet_to_batch_stream(
- &task.file_path,
- task.file_size_in_bytes,
- task.key_metadata.as_deref(),
+ task.file_path(),
+ task.file_size_in_bytes(),
+ task.key_metadata(),
)
.await?,
schema,
@@ -331,59 +331,47 @@ impl CachingDeleteFileLoader {
}
}
- /// Validates a deletion-vector task and returns what the read needs as typed values:
+ /// Extracts a deletion-vector task's metadata as the values needed by the read:
/// `(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`.
+ /// Both builder construction and deserialization validate the task's required fields.
+ /// The fallible extraction here converts the optional fields into values for the
+ /// byte-range read, preserving `DataInvalid` errors instead of unwrapping them.
fn validate_deletion_vector_task(
task: &FileScanTaskDeleteFile,
) -> Result<(u64, u64, String, u64)> {
- let content_offset = task.content_offset.ok_or_else(|| {
+ let content_offset = task.content_offset().ok_or_else(|| {
invalid_data!(
"deletion vector {} is missing content_offset",
- task.file_path
+ task.file_path()
)
})?;
- let content_size = task.content_size_in_bytes.ok_or_else(|| {
+ let content_size = task.content_size_in_bytes().ok_or_else(|| {
invalid_data!(
"deletion vector {} is missing content_size_in_bytes",
- task.file_path
- )
- })?;
- let data_file_path = task.referenced_data_file.clone().ok_or_else(|| {
- invalid_data!(
- "deletion vector {} is missing referenced_data_file",
- task.file_path
- )
- })?;
- let record_count = task.record_count.ok_or_else(|| {
- invalid_data!("deletion vector {} is missing record_count", task.file_path)
- })?;
-
- let start = u64::try_from(content_offset).map_err(|_| {
- invalid_data!(
- "deletion vector {} has negative content_offset {content_offset}",
- task.file_path
+ task.file_path()
)
})?;
- let len = u64::try_from(content_size).map_err(|_| {
+ let data_file_path = task
+ .referenced_data_file()
+ .map(ToOwned::to_owned)
+ .ok_or_else(|| {
+ invalid_data!(
+ "deletion vector {} is missing referenced_data_file",
+ task.file_path()
+ )
+ })?;
+ let record_count = task.record_count().ok_or_else(|| {
invalid_data!(
- "deletion vector {} has negative content_size_in_bytes {content_size}",
- task.file_path
+ "deletion vector {} is missing record_count",
+ task.file_path()
)
})?;
- Ok((start, len, data_file_path, record_count))
+ // content_offset / content_size_in_bytes are u64 on the task, and the
+ // manifest -> task conversion already rejected a negative, so there is
+ // nothing left to convert or re-check here.
+ Ok((content_offset, content_size, data_file_path, record_count))
}
/// Validates a decoded deletion vector's cardinality against the manifest entry's
@@ -420,8 +408,8 @@ impl CachingDeleteFileLoader {
let input_file = basic_delete_file_loader
.file_io()
- .new_input(&task.file_path)?;
- let blob = match task.key_metadata.as_deref() {
+ .new_input(task.file_path())?;
+ let blob = match task.key_metadata() {
Some(key_metadata) => {
let key_metadata = StandardKeyMetadata::decode(key_metadata)?;
EncryptedInputFile::new(input_file, key_metadata)
@@ -437,7 +425,7 @@ impl CachingDeleteFileLoader {
data_file_path,
blob,
record_count,
- dv_path: task.file_path.clone(),
+ dv_path: task.file_path().to_string(),
})
}
@@ -849,9 +837,29 @@ mod tests {
use super::*;
use crate::arrow::delete_filter::tests::setup;
use crate::scan::FileScanTaskDeleteFile;
- use crate::spec::{DataContentType, Schema};
+ use crate::spec::{DataContentType, DataFileFormat, Schema};
use crate::test_utils::encode_dv_blob;
+ #[test]
+ fn test_deserialized_deletion_vector_rejects_missing_fields() {
+ // A deletion vector that omits its required coordinates fails validation
+ // during deserialization (the try_from mirror routes through `build()`),
+ // before the loader ever sees the task.
+ let err = serde_json::from_value::<FileScanTaskDeleteFile>(serde_json::json!({
+ "file_path": "dv.puffin",
+ "file_size_in_bytes": 100,
+ "file_type": "PositionDeletes",
+ "file_format": "Puffin",
+ "partition_spec_id": 0
+ }))
+ .unwrap_err();
+
+ assert!(
+ err.to_string().contains("missing referenced_data_file"),
+ "expected a missing-field validation error, got `{err}`"
+ );
+ }
+
#[tokio::test]
async fn test_delete_file_loader_parse_equality_deletes() {
let tmp_dir = TempDir::new().unwrap();
@@ -1400,7 +1408,8 @@ mod tests {
.with_file_type(DataContentType::PositionDeletes)
.with_file_format(DataFileFormat::Parquet)
.with_partition_spec_id(0)
- .build();
+ .build()
+ .unwrap();
let eq_del = FileScanTaskDeleteFile::builder()
.with_file_path(eq_delete_path.clone())
@@ -1409,7 +1418,8 @@ mod tests {
.with_file_format(DataFileFormat::Parquet)
.with_partition_spec_id(0)
.with_equality_ids(Some(vec![2, 3])) // Only use field IDs that exist in both schemas
- .build();
+ .build()
+ .unwrap();
let file_scan_task = FileScanTask::builder()
.with_file_size_in_bytes(0)
@@ -1542,8 +1552,8 @@ mod tests {
dv_path: String,
file_size: u64,
data_file_path: String,
- content_offset: i64,
- content_size: i64,
+ content_offset: u64,
+ content_size: u64,
record_count: u64,
key_metadata: Option<Box<[u8]>>,
) -> FileScanTaskDeleteFile {
@@ -1559,6 +1569,7 @@ mod tests {
.with_record_count(Some(record_count))
.with_key_metadata(key_metadata)
.build()
+ .expect("deletion vector task should be valid")
}
#[tokio::test]
@@ -1571,8 +1582,8 @@ mod tests {
// 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 content_offset = 12u64;
+ let content_size = blob.len() as u64;
let mut file_bytes = vec![0u8; content_offset as usize];
file_bytes.extend_from_slice(&blob);
file_bytes.extend_from_slice(&[0u8; 8]);
@@ -1623,7 +1634,7 @@ mod tests {
.with_aad_prefix(b"test-aad-prefix!");
let blob = encode_dv_blob([2u64, 4]);
- let plaintext_size = blob.len() as i64;
+ let plaintext_size = blob.len() as u64;
let dv_path = format!("{table_location}/deletes.puffin");
let output = EncryptedOutputFile::new(file_io.new_output(&dv_path).unwrap(), key_metadata);
let file_metadata = output.write(Bytes::from(blob)).await.unwrap();
@@ -1683,7 +1694,7 @@ mod tests {
std::fs::metadata(&dv_path).unwrap().len(),
data_file_path,
0,
- blob.len() as i64,
+ blob.len() as u64,
2,
None,
);
@@ -1708,7 +1719,8 @@ 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.
+ // A well-formed deletion-vector task, for tests that then corrupt one field after builder
+ // validation.
fn valid_dv_task() -> FileScanTaskDeleteFile {
dv_task(
"deletes.puffin".to_string(),
@@ -1721,64 +1733,31 @@ mod tests {
)
}
+ // The coordinate fields are unsigned, so a negative byte offset is rejected during
+ // deserialization: a hand-written or corrupted scan plan fails before any loader
+ // code runs.
#[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);
+ fn test_deserializing_negative_content_offset_is_rejected() {
+ let mut task = serde_json::to_value(valid_dv_task()).unwrap();
+ task["content_offset"] = serde_json::json!(-1);
- let err = CachingDeleteFileLoader::validate_deletion_vector_task(&task).unwrap_err();
- assert_eq!(err.kind(), ErrorKind::DataInvalid);
- assert!(err.message().contains("negative content_offset"));
+ let err = serde_json::from_value::<FileScanTaskDeleteFile>(task).unwrap_err();
+ assert!(
+ err.to_string().contains("invalid value"),
+ "expected an out-of-range integer error, got `{err}`"
+ );
}
#[test]
- fn test_validate_deletion_vector_task_rejects_negative_content_size() {
- let mut task = valid_dv_task();
- task.content_size_in_bytes = Some(-1);
+ fn test_deserializing_negative_content_size_is_rejected() {
+ let mut task = serde_json::to_value(valid_dv_task()).unwrap();
+ task["content_size_in_bytes"] = serde_json::json!(-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"));
+ let err = serde_json::from_value::<FileScanTaskDeleteFile>(task).unwrap_err();
+ assert!(
+ err.to_string().contains("invalid value"),
+ "expected an out-of-range integer error, got `{err}`"
+ );
}
#[test]
diff --git a/crates/iceberg/src/arrow/delete_file_loader.rs b/crates/iceberg/src/arrow/delete_file_loader.rs
index 5852cf1ac..ec61710a4 100644
--- a/crates/iceberg/src/arrow/delete_file_loader.rs
+++ b/crates/iceberg/src/arrow/delete_file_loader.rs
@@ -124,16 +124,16 @@ impl DeleteFileLoader for BasicDeleteFileLoader {
) -> Result<ArrowRecordBatchStream> {
let raw_batch_stream = self
.parquet_to_batch_stream(
- &task.file_path,
- task.file_size_in_bytes,
- task.key_metadata.as_deref(),
+ task.file_path(),
+ task.file_size_in_bytes(),
+ task.key_metadata(),
)
.await?;
// For equality deletes, only evolve the equality_ids columns.
// For positional deletes (equality_ids is None), use all field IDs.
- let field_ids = match &task.equality_ids {
- Some(ids) => ids.clone(),
+ let field_ids = match task.equality_ids() {
+ Some(ids) => ids.to_vec(),
None => schema.field_id_to_name_map().keys().cloned().collect(),
};
@@ -230,19 +230,15 @@ mod tests {
.unwrap(),
);
- let task = FileScanTaskDeleteFile {
- 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())),
- referenced_data_file: None,
- content_offset: None,
- content_size_in_bytes: None,
- record_count: None,
- };
+ let task = FileScanTaskDeleteFile::builder()
+ .with_file_path(del_path.clone())
+ .with_file_size_in_bytes(std::fs::metadata(&del_path).unwrap().len())
+ .with_file_type(DataContentType::PositionDeletes)
+ .with_file_format(DataFileFormat::Parquet)
+ .with_partition_spec_id(0)
+ .with_key_metadata(Some(Box::from(key_metadata.as_ref())))
+ .build()
+ .unwrap();
let scan_metrics = ScanMetrics::new();
let delete_file_loader = BasicDeleteFileLoader::new(file_io, scan_metrics);
@@ -309,19 +305,16 @@ mod tests {
.unwrap(),
);
- let task = FileScanTaskDeleteFile {
- 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())),
- referenced_data_file: None,
- content_offset: None,
- content_size_in_bytes: None,
- record_count: None,
- };
+ let task = FileScanTaskDeleteFile::builder()
+ .with_file_path(del_path.clone())
+ .with_file_size_in_bytes(std::fs::metadata(&del_path).unwrap().len())
+ .with_file_type(DataContentType::EqualityDeletes)
+ .with_file_format(DataFileFormat::Parquet)
+ .with_partition_spec_id(0)
+ .with_equality_ids(Some(vec![1]))
+ .with_key_metadata(Some(Box::from(key_metadata.as_ref())))
+ .build()
+ .unwrap();
let scan_metrics = ScanMetrics::new();
let delete_file_loader = BasicDeleteFileLoader::new(file_io, scan_metrics);
diff --git a/crates/iceberg/src/arrow/delete_filter.rs b/crates/iceberg/src/arrow/delete_filter.rs
index 6f13be71d..29843295d 100644
--- a/crates/iceberg/src/arrow/delete_filter.rs
+++ b/crates/iceberg/src/arrow/delete_filter.rs
@@ -208,14 +208,14 @@ impl DeleteFilter {
}
let Some(predicate) = self
- .get_equality_delete_predicate_for_delete_file_path(&delete.file_path)
+ .get_equality_delete_predicate_for_delete_file_path(delete.file_path())
.await
else {
return Err(Error::new(
ErrorKind::Unexpected,
format!(
"Missing predicate for equality delete file '{}'",
- delete.file_path
+ delete.file_path()
),
));
};
@@ -279,7 +279,7 @@ impl DeleteFilter {
}
pub(crate) fn is_equality_delete(f: &FileScanTaskDeleteFile) -> bool {
- matches!(f.file_type, DataContentType::EqualityDeletes)
+ matches!(f.file_type(), DataContentType::EqualityDeletes)
}
#[cfg(test)]
@@ -448,7 +448,8 @@ pub(crate) mod tests {
.with_file_type(DataContentType::PositionDeletes)
.with_file_format(DataFileFormat::Parquet)
.with_partition_spec_id(0)
- .build();
+ .build()
+ .unwrap();
let pos_del_2 = FileScanTaskDeleteFile::builder()
.with_file_path(format!(
@@ -466,7 +467,8 @@ pub(crate) mod tests {
.with_file_type(DataContentType::PositionDeletes)
.with_file_format(DataFileFormat::Parquet)
.with_partition_spec_id(0)
- .build();
+ .build()
+ .unwrap();
let pos_del_3 = FileScanTaskDeleteFile::builder()
.with_file_path(format!(
@@ -484,7 +486,8 @@ pub(crate) mod tests {
.with_file_type(DataContentType::PositionDeletes)
.with_file_format(DataFileFormat::Parquet)
.with_partition_spec_id(0)
- .build();
+ .build()
+ .unwrap();
let file_scan_tasks = vec![
FileScanTask::builder()
@@ -561,7 +564,8 @@ pub(crate) mod tests {
.with_file_type(DataContentType::EqualityDeletes)
.with_file_format(DataFileFormat::Parquet)
.with_partition_spec_id(0)
- .build(),
+ .build()
+ .unwrap(),
])
.with_case_sensitive(true)
.build()
diff --git a/crates/iceberg/src/arrow/reader/pipeline.rs b/crates/iceberg/src/arrow/reader/pipeline.rs
index 11fa200ee..18ba42d3b 100644
--- a/crates/iceberg/src/arrow/reader/pipeline.rs
+++ b/crates/iceberg/src/arrow/reader/pipeline.rs
@@ -2486,11 +2486,12 @@ mod tests {
let delete = FileScanTaskDeleteFile::builder()
.with_file_path(del_path.clone())
- .with_file_format(DataFileFormat::Parquet)
.with_file_size_in_bytes(std::fs::metadata(&del_path).unwrap().len())
.with_file_type(DataContentType::PositionDeletes)
+ .with_file_format(DataFileFormat::Parquet)
.with_partition_spec_id(0)
- .build();
+ .build()
+ .unwrap();
let task = row_id_task_with_options(data_path, Some(100), 0, 0, vec![delete]);
let reader = ArrowReaderBuilder::new(FileIO::new_with_fs(), Runtime::current()).build();
diff --git a/crates/iceberg/src/arrow/reader/positional_deletes.rs b/crates/iceberg/src/arrow/reader/positional_deletes.rs
index 2915596da..a261bd536 100644
--- a/crates/iceberg/src/arrow/reader/positional_deletes.rs
+++ b/crates/iceberg/src/arrow/reader/positional_deletes.rs
@@ -452,7 +452,8 @@ mod tests {
.with_file_type(DataContentType::PositionDeletes)
.with_file_format(DataFileFormat::Parquet)
.with_partition_spec_id(0)
- .build(),
+ .build()
+ .unwrap(),
])
.with_case_sensitive(false)
.build()
@@ -672,7 +673,8 @@ mod tests {
.with_file_type(DataContentType::PositionDeletes)
.with_file_format(DataFileFormat::Parquet)
.with_partition_spec_id(0)
- .build(),
+ .build()
+ .unwrap(),
])
.with_case_sensitive(false)
.build()
@@ -886,7 +888,8 @@ mod tests {
.with_file_type(DataContentType::PositionDeletes)
.with_file_format(DataFileFormat::Parquet)
.with_partition_spec_id(0)
- .build(),
+ .build()
+ .unwrap(),
])
.with_case_sensitive(false)
.build()
@@ -975,8 +978,8 @@ mod tests {
// 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 content_offset = 12u64;
+ let content_size = blob.len() as u64;
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");
@@ -1005,12 +1008,14 @@ mod tests {
.with_content_offset(Some(content_offset))
.with_content_size_in_bytes(Some(content_size))
.with_record_count(Some(2))
- .build(),
+ .build()
+ .unwrap(),
])
.with_case_sensitive(false)
- .build();
+ .build()
+ .unwrap();
- let tasks = Box::pin(futures::stream::iter(vec![task])) as FileScanTaskStream;
+ let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream;
let result = reader
.read(tasks)
.unwrap()
@@ -1100,14 +1105,16 @@ mod tests {
.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_content_size_in_bytes(Some(blob.len() as u64))
.with_record_count(Some(5))
- .build(),
+ .build()
+ .unwrap(),
])
.with_case_sensitive(false)
- .build();
+ .build()
+ .unwrap();
- let tasks = Box::pin(futures::stream::iter(vec![task])) as FileScanTaskStream;
+ 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;
diff --git a/crates/iceberg/src/arrow/reader/row_filter.rs b/crates/iceberg/src/arrow/reader/row_filter.rs
index 34b739113..5d92f6bc7 100644
--- a/crates/iceberg/src/arrow/reader/row_filter.rs
+++ b/crates/iceberg/src/arrow/reader/row_filter.rs
@@ -1235,19 +1235,16 @@ mod tests {
.with_schema(iceberg_schema.clone())
.with_project_field_ids(vec![1, 2])
.with_predicate(Some(predicate_sub2))
- .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(),
- referenced_data_file: None,
- content_offset: None,
- content_size_in_bytes: None,
- record_count: None,
- key_metadata: None,
- }])
+ .with_deletes(vec![
+ FileScanTaskDeleteFile::builder()
+ .with_file_path(pos_del_path.clone())
+ .with_file_type(DataContentType::PositionDeletes)
+ .with_file_format(DataFileFormat::Parquet)
+ .with_partition_spec_id(0)
+ .with_file_size_in_bytes(std::fs::metadata(&pos_del_path).unwrap().len())
+ .build()
+ .unwrap(),
+ ])
.with_case_sensitive(false)
.build()
.unwrap();
diff --git a/crates/iceberg/src/cow_rewrite/plan.rs b/crates/iceberg/src/cow_rewrite/plan.rs
index b96aea635..8e96e9ab4 100644
--- a/crates/iceberg/src/cow_rewrite/plan.rs
+++ b/crates/iceberg/src/cow_rewrite/plan.rs
@@ -151,7 +151,7 @@ mod tests {
for file in files {
assert_eq!(file.scan_task.deletes().len(), 1);
assert_eq!(
- file.scan_task.deletes()[0].file_path,
+ file.scan_task.deletes()[0].file_path(),
format!("{}/del.parquet", fixture.table_location)
);
}
diff --git a/crates/iceberg/src/delete_file_index.rs b/crates/iceberg/src/delete_file_index.rs
index 692ab9315..8bae43941 100644
--- a/crates/iceberg/src/delete_file_index.rs
+++ b/crates/iceberg/src/delete_file_index.rs
@@ -215,25 +215,14 @@ impl PopulatedDeleteFileIndex {
// 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(invalid_data!(
+ // referenced_data_file is the map key the index needs; the offset/size/
+ // record_count invariants are enforced centrally when the task is built.
+ let path = data_file.referenced_data_file().ok_or_else(|| {
+ invalid_data!(
"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(invalid_data!(
- "deletion vector {} is missing content_offset or content_size_in_bytes",
- arc_ctx.manifest_entry.file_path()
- ));
- }
+ data_file.file_path()
+ )
+ })?;
if let Some(existing) =
dvs_by_referenced_data_file.insert(path.clone(), arc_ctx)
@@ -295,27 +284,25 @@ impl PopulatedDeleteFileIndex {
) -> Result<Vec<FileScanTaskDeleteFile>> {
let mut results = vec![];
- self.global_equality_deletes
- .iter()
+ for delete in self.global_equality_deletes.iter().filter(|&delete| {
// filter that returns true if the provided delete file's sequence number is **greater than** `seq_num`
- .filter(|&delete| {
- seq_num
- .map(|seq_num| delete.manifest_entry.sequence_number() > Some(seq_num))
- .unwrap_or_else(|| true)
- })
- .for_each(|delete| results.push(delete.as_ref().into()));
+ seq_num
+ .map(|seq_num| delete.manifest_entry.sequence_number() > Some(seq_num))
+ .unwrap_or_else(|| true)
+ }) {
+ results.push(FileScanTaskDeleteFile::try_from(delete.as_ref())?);
+ }
if let Some(deletes) = self.eq_deletes_by_partition.get(data_file.partition()) {
- deletes
- .iter()
+ for delete in deletes.iter().filter(|&delete| {
// filter that returns true if the provided delete file's sequence number is **greater than** `seq_num`
- .filter(|&delete| {
- seq_num
- .map(|seq_num| delete.manifest_entry.sequence_number() > Some(seq_num))
- .unwrap_or_else(|| true)
- && data_file.partition_spec_id == delete.partition_spec_id
- })
- .for_each(|delete| results.push(delete.as_ref().into()));
+ seq_num
+ .map(|seq_num| delete.manifest_entry.sequence_number() > Some(seq_num))
+ .unwrap_or_else(|| true)
+ && data_file.partition_spec_id == delete.partition_spec_id
+ }) {
+ results.push(FileScanTaskDeleteFile::try_from(delete.as_ref())?);
+ }
}
// A deletion vector supersedes all position delete files for its data file, per the spec:
@@ -351,12 +338,12 @@ impl PopulatedDeleteFileIndex {
));
}
}
- results.push(dv.as_ref().into());
+ results.push(FileScanTaskDeleteFile::try_from(dv.as_ref())?);
return Ok(results);
}
if let Some(deletes) = self.pos_deletes_by_partition.get(data_file.partition()) {
- deletes
+ for delete in deletes
.iter()
// filter that returns true if the provided delete file's sequence number is **greater than or equal to** `seq_num`
.filter(|&delete| {
@@ -365,14 +352,16 @@ impl PopulatedDeleteFileIndex {
.unwrap_or_else(|| true)
&& data_file.partition_spec_id == delete.partition_spec_id
})
- .for_each(|delete| results.push(delete.as_ref().into()));
+ {
+ results.push(FileScanTaskDeleteFile::try_from(delete.as_ref())?);
+ }
}
// Position deletes indexed by the exact path of the data file they reference.
// An exact path match is sufficient proof that the delete applies, so no
// partition spec id check is performed.
if let Some(deletes) = self.pos_deletes_by_path.get(data_file.file_path()) {
- deletes
+ for delete in deletes
.iter()
// filter that returns true if the provided delete file's sequence number is **greater than or equal to** `seq_num`
.filter(|&delete| {
@@ -380,7 +369,9 @@ impl PopulatedDeleteFileIndex {
.map(|seq_num| delete.manifest_entry.sequence_number() >= Some(seq_num))
.unwrap_or(true)
})
- .for_each(|delete| results.push(delete.as_ref().into()));
+ {
+ results.push(FileScanTaskDeleteFile::try_from(delete.as_ref())?);
+ }
}
Ok(results)
@@ -442,7 +433,7 @@ mod tests {
.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)
+ .map(|file| file.file_path().to_string())
.collect();
assert_eq!(
@@ -456,7 +447,7 @@ mod tests {
.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)
+ .map(|file| file.file_path().to_string())
.collect();
assert_eq!(
actual_paths_to_apply_for_seq_5,
@@ -469,7 +460,7 @@ mod tests {
.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)
+ .map(|file| file.file_path().to_string())
.collect();
assert_eq!(
actual_paths_to_apply_for_seq_6,
@@ -486,7 +477,7 @@ mod tests {
let actual_paths_to_apply_for_partitioned_file: Vec<String> =
delete_files_to_apply_for_partitioned_file
.into_iter()
- .map(|file| file.file_path)
+ .map(|file| file.file_path().to_string())
.collect();
assert_eq!(
actual_paths_to_apply_for_partitioned_file,
@@ -541,7 +532,7 @@ mod tests {
.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)
+ .map(|file| file.file_path().to_string())
.collect();
assert_eq!(
@@ -555,7 +546,7 @@ mod tests {
.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)
+ .map(|file| file.file_path().to_string())
.collect();
assert_eq!(
actual_paths_to_apply_for_seq_5,
@@ -568,7 +559,7 @@ mod tests {
.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)
+ .map(|file| file.file_path().to_string())
.collect();
assert_eq!(
actual_paths_to_apply_for_seq_6,
@@ -584,7 +575,7 @@ mod tests {
let actual_paths_to_apply_for_different_partition: Vec<String> =
delete_files_to_apply_for_different_partition
.into_iter()
- .map(|file| file.file_path)
+ .map(|file| file.file_path().to_string())
.collect();
assert!(actual_paths_to_apply_for_different_partition.is_empty());
@@ -596,7 +587,7 @@ mod tests {
let actual_paths_to_apply_for_different_spec: Vec<String> =
delete_files_to_apply_for_different_spec
.into_iter()
- .map(|file| file.file_path)
+ .map(|file| file.file_path().to_string())
.collect();
assert!(actual_paths_to_apply_for_different_spec.is_empty());
}
@@ -617,7 +608,7 @@ mod tests {
.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());
+ 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.
@@ -785,7 +776,7 @@ mod tests {
.get_deletes_for_data_file(&data_file, Some(0))
.unwrap()
.into_iter()
- .map(|delete| delete.file_path)
+ .map(|delete| delete.file_path().to_string())
.collect();
actual_paths.sort();
@@ -822,7 +813,7 @@ mod tests {
.get_deletes_for_data_file(&data_file, Some(0))
.unwrap()
.into_iter()
- .map(|delete| delete.file_path)
+ .map(|delete| delete.file_path().to_string())
.collect();
actual_paths.sort();
@@ -861,7 +852,10 @@ mod tests {
.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_eq!(
+ deletes_for_referenced[0].file_path(),
+ pos_delete.file_path()
+ );
assert!(
index
@@ -964,13 +958,13 @@ mod tests {
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.record_count, Some(3));
+ let task = FileScanTaskDeleteFile::try_from(&ctx).unwrap();
+ 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(),
+ task.referenced_data_file(),
Some("s3://bucket/data/part-0.parquet")
);
}
@@ -1001,7 +995,7 @@ mod tests {
.get_deletes_for_data_file(&data_file, Some(0))
.unwrap()
.into_iter()
- .map(|f| f.file_path)
+ .map(|f| f.file_path().to_string())
.collect();
// Only the deletion vector applies; the partition-scoped position delete file, which
@@ -1116,7 +1110,7 @@ mod tests {
.get_deletes_for_data_file(&data_file, Some(0))
.unwrap()
.into_iter()
- .map(|f| f.file_path)
+ .map(|f| f.file_path().to_string())
.collect();
// Only the deletion vector applies; the path-scoped position delete file, which would
@@ -1151,7 +1145,7 @@ mod tests {
.get_deletes_for_data_file(&data_file, Some(0))
.unwrap()
.into_iter()
- .map(|f| f.file_path)
+ .map(|f| f.file_path().to_string())
.collect();
applied.sort();
@@ -1162,7 +1156,7 @@ mod tests {
}
#[test]
- fn test_deletion_vector_missing_referenced_data_file_is_rejected() {
+ fn test_deletion_vector_index_rejects_missing_referenced_data_file() {
let malformed_dv = DataFileBuilder::default()
.file_path("deletes.puffin".to_string())
.file_format(DataFileFormat::Puffin)
@@ -1175,43 +1169,59 @@ mod tests {
.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() {
+ fn test_deletion_vector_index_rejects_missing_coordinates() {
+ // The missing-coordinate check fires in `get_deletes_for_data_file`, where the
+ // manifest entry is converted into a task. `PopulatedDeleteFileIndex::new` only
+ // needs `referenced_data_file` as its map key, so a DV that carries it but omits
+ // `content_offset` survives construction and is rejected when the data file is
+ // resolved against it.
+ let data_file = DataFileBuilder::default()
+ .file_path("target-data.parquet".to_string())
+ .file_format(DataFileFormat::Parquet)
+ .content(DataContentType::Data)
+ .record_count(100)
+ .partition(Struct::empty())
+ .partition_spec_id(0)
+ .file_size_in_bytes(100)
+ .build()
+ .unwrap();
+
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()))
+ .referenced_data_file(Some("target-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();
+ let delete_file_index = PopulatedDeleteFileIndex::new(contexts).unwrap();
+ let err = delete_file_index
+ .get_deletes_for_data_file(&data_file, Some(0))
+ .unwrap_err();
+
assert_eq!(err.kind(), ErrorKind::DataInvalid);
- assert!(
- err.message()
- .contains("missing content_offset or content_size_in_bytes")
- );
+ assert!(err.message().contains("missing content_offset"));
}
#[test]
diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs
index ccdb5c7d5..0c241125d 100644
--- a/crates/iceberg/src/scan/mod.rs
+++ b/crates/iceberg/src/scan/mod.rs
@@ -1881,7 +1881,8 @@ mod tests {
.with_content_size_in_bytes(Some(34))
.with_record_count(Some(5))
.with_key_metadata(Some(vec![4, 5, 6].into_boxed_slice()))
- .build(),
+ .build()
+ .unwrap(),
])
.with_partition(Some(Struct::from_iter([Some(Literal::long(42))])))
.with_partition_spec(Some(partition_spec))
@@ -2905,7 +2906,7 @@ mod tests {
"positional delete file should be planned into the task"
);
assert_eq!(
- tasks[0].deletes()[0].file_type,
+ tasks[0].deletes()[0].file_type(),
DataContentType::PositionDeletes
);
diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs
index 97ca7e518..45940b0df 100644
--- a/crates/iceberg/src/scan/task.rs
+++ b/crates/iceberg/src/scan/task.rs
@@ -21,13 +21,13 @@ use futures::stream::BoxStream;
use serde::{Deserialize, Serialize};
use typed_builder::TypedBuilder;
-use crate::Result;
use crate::error::invalid_data;
use crate::expr::BoundPredicate;
use crate::spec::{
DataContentType, DataFileFormat, ManifestEntryRef, NameMapping, PartitionSpec, Schema,
SchemaRef, SortOrderRef, Struct, StructType,
};
+use crate::{Error, Result};
/// A stream of [`FileScanTask`].
pub type FileScanTaskStream = BoxStream<'static, Result<FileScanTask>>;
@@ -309,8 +309,24 @@ pub(crate) struct DeleteFileContext {
pub(crate) partition_spec_id: i32,
}
-impl From<&DeleteFileContext> for FileScanTaskDeleteFile {
- fn from(ctx: &DeleteFileContext) -> Self {
+impl TryFrom<&DeleteFileContext> for FileScanTaskDeleteFile {
+ type Error = Error;
+
+ fn try_from(ctx: &DeleteFileContext) -> Result<Self> {
+ // The manifest stores these as i64. Convert here so a negative byte offset is
+ // rejected once at the boundary, for every delete-file kind rather than only
+ // deletion vectors, instead of being carried inward and re-checked per consumer.
+ let file_path = ctx.manifest_entry.file_path();
+ let to_offset = |value: Option<i64>, field: &str| -> Result<Option<u64>> {
+ value
+ .map(|value| {
+ u64::try_from(value).map_err(|_| {
+ invalid_data!("delete file {file_path} has negative {field} {value}")
+ })
+ })
+ .transpose()
+ };
+
FileScanTaskDeleteFile::builder()
.with_file_path(ctx.manifest_entry.file_path().to_string())
.with_file_size_in_bytes(ctx.manifest_entry.file_size_in_bytes())
@@ -319,8 +335,14 @@ impl From<&DeleteFileContext> for FileScanTaskDeleteFile {
.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_content_offset(to_offset(
+ ctx.manifest_entry.data_file.content_offset,
+ "content_offset",
+ )?)
+ .with_content_size_in_bytes(to_offset(
+ ctx.manifest_entry.data_file.content_size_in_bytes,
+ "content_size_in_bytes",
+ )?)
.with_record_count(Some(ctx.manifest_entry.record_count()))
.with_key_metadata(
ctx.manifest_entry
@@ -335,55 +357,62 @@ impl From<&DeleteFileContext> for FileScanTaskDeleteFile {
/// A task to scan part of file.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, TypedBuilder)]
-#[builder(field_defaults(setter(prefix = "with_")))]
+#[serde(try_from = "crate::scan::task::_serde::FileScanTaskDeleteFileSerde")]
+#[builder(
+ field_defaults(setter(prefix = "with_")),
+ build_method(into = Result<FileScanTaskDeleteFile>)
+)]
pub struct FileScanTaskDeleteFile {
/// The delete file path
- pub file_path: String,
+ file_path: String,
/// The total size of the delete file in bytes, from the manifest entry.
- pub file_size_in_bytes: u64,
+ file_size_in_bytes: u64,
/// delete file type
- pub file_type: DataContentType,
+ 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,
+ file_format: DataFileFormat,
/// partition id
- pub partition_spec_id: i32,
+ partition_spec_id: i32,
/// equality ids for equality deletes (null for anything other than equality-deletes)
#[builder(default)]
- pub equality_ids: Option<Vec<i32>>,
+ equality_ids: Option<Vec<i32>>,
/// For a deletion vector, the location of the data file whose rows it deletes. Required for
/// deletion vectors, and may also be set on a position delete file scoped to one data file.
#[serde(default)]
#[serde(skip_serializing_if = "Option::is_none")]
#[builder(default)]
- pub referenced_data_file: Option<String>,
+ referenced_data_file: Option<String>,
/// For a deletion vector, the offset of the blob within its Puffin file. Set only for
- /// deletion vectors, where it locates the blob for direct access.
+ /// deletion vectors, where it locates the blob for direct access. Unsigned because it is a
+ /// byte offset: a negative value is rejected once, when the task is built from a manifest
+ /// entry, rather than re-checked by every consumer.
#[serde(default)]
#[serde(skip_serializing_if = "Option::is_none")]
#[builder(default)]
- pub content_offset: Option<i64>,
+ content_offset: Option<u64>,
/// For a deletion vector, the length in bytes of the blob within its Puffin file.
/// Required together with `content_offset`; both are absent for non-DV delete files.
+ /// Unsigned for the same reason as `content_offset`.
#[serde(default)]
#[serde(skip_serializing_if = "Option::is_none")]
#[builder(default)]
- pub content_size_in_bytes: Option<i64>,
+ content_size_in_bytes: Option<u64>,
/// The number of records in the delete file, from the manifest entry; for a deletion vector,
/// the cardinality of its bitmap. `None` only for a task not built from a manifest entry.
#[serde(default)]
#[serde(skip_serializing_if = "Option::is_none")]
#[builder(default)]
- pub record_count: Option<u64>,
+ record_count: Option<u64>,
/// 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
@@ -397,7 +426,7 @@ pub struct FileScanTaskDeleteFile {
#[serde(default)]
#[serde(skip_serializing_if = "Option::is_none")]
#[builder(default)]
- pub key_metadata: Option<Box<[u8]>>,
+ key_metadata: Option<Box<[u8]>>,
}
mod _serde {
@@ -409,8 +438,8 @@ mod _serde {
use crate::error::invalid_data;
use crate::expr::BoundPredicate;
use crate::spec::{
- DataFileFormat, Literal, NameMapping, PartitionSpec, RawLiteral, SchemaRef, SortOrderRef,
- StructType, Type,
+ DataContentType, DataFileFormat, Literal, NameMapping, PartitionSpec, RawLiteral,
+ SchemaRef, SortOrderRef, StructType, Type,
};
use crate::{Error, Result};
@@ -579,13 +608,168 @@ mod _serde {
.build()
}
}
+
+ #[derive(Deserialize)]
+ pub(super) struct FileScanTaskDeleteFileSerde {
+ file_path: String,
+ file_size_in_bytes: u64,
+ file_type: DataContentType,
+ file_format: DataFileFormat,
+ partition_spec_id: i32,
+ equality_ids: Option<Vec<i32>>,
+ #[serde(default)]
+ referenced_data_file: Option<String>,
+ #[serde(default)]
+ content_offset: Option<u64>,
+ #[serde(default)]
+ content_size_in_bytes: Option<u64>,
+ #[serde(default)]
+ record_count: Option<u64>,
+ #[serde(default)]
+ key_metadata: Option<Box<[u8]>>,
+ }
+
+ impl TryFrom<FileScanTaskDeleteFileSerde> for FileScanTaskDeleteFile {
+ type Error = Error;
+
+ fn try_from(value: FileScanTaskDeleteFileSerde) -> Result<Self> {
+ Self::builder()
+ .with_file_path(value.file_path)
+ .with_file_size_in_bytes(value.file_size_in_bytes)
+ .with_file_type(value.file_type)
+ .with_file_format(value.file_format)
+ .with_partition_spec_id(value.partition_spec_id)
+ .with_equality_ids(value.equality_ids)
+ .with_referenced_data_file(value.referenced_data_file)
+ .with_content_offset(value.content_offset)
+ .with_content_size_in_bytes(value.content_size_in_bytes)
+ .with_record_count(value.record_count)
+ .with_key_metadata(value.key_metadata)
+ .build()
+ }
+ }
+}
+
+impl FileScanTaskDeleteFile {
+ /// Returns the delete file path.
+ pub fn file_path(&self) -> &str {
+ &self.file_path
+ }
+
+ /// Returns the total size of the delete file in bytes.
+ pub fn file_size_in_bytes(&self) -> u64 {
+ self.file_size_in_bytes
+ }
+
+ /// Returns the delete file content type.
+ pub fn file_type(&self) -> DataContentType {
+ self.file_type
+ }
+
+ /// Returns the delete file format.
+ pub fn file_format(&self) -> DataFileFormat {
+ self.file_format
+ }
+
+ /// Returns the partition spec id.
+ pub fn partition_spec_id(&self) -> i32 {
+ self.partition_spec_id
+ }
+
+ /// Returns the equality field ids for an equality delete file.
+ pub fn equality_ids(&self) -> Option<&[i32]> {
+ self.equality_ids.as_deref()
+ }
+
+ /// Returns the referenced data file path.
+ pub fn referenced_data_file(&self) -> Option<&str> {
+ self.referenced_data_file.as_deref()
+ }
+
+ /// Returns the deletion vector blob offset.
+ pub fn content_offset(&self) -> Option<u64> {
+ self.content_offset
+ }
+
+ /// Returns the deletion vector blob size in bytes.
+ pub fn content_size_in_bytes(&self) -> Option<u64> {
+ self.content_size_in_bytes
+ }
+
+ /// Returns the number of records in the delete file.
+ pub fn record_count(&self) -> Option<u64> {
+ self.record_count
+ }
+
+ /// Returns the key metadata for the encrypted delete file.
+ pub fn key_metadata(&self) -> Option<&[u8]> {
+ self.key_metadata.as_deref()
+ }
+
+ fn is_deletion_vector(&self) -> bool {
+ self.file_type == DataContentType::PositionDeletes
+ && self.file_format == DataFileFormat::Puffin
+ }
+
+ fn validate(&self) -> Result<()> {
+ // Negative offsets/sizes are unrepresentable: both fields are u64, and the
+ // conversion from the manifest entry's i64 rejects a negative there.
+
+ // Spec: `content_size_in_bytes` is required whenever `content_offset` is
+ // present, for every delete-file kind (not only deletion vectors).
+ if self.content_offset.is_some() && self.content_size_in_bytes.is_none() {
+ return Err(invalid_data!(
+ "delete file {} is missing content_size_in_bytes while content_offset is present",
+ self.file_path
+ ));
+ }
+
+ // Presence of the deletion-vector coordinates is required only for
+ // deletion vectors; other delete files legitimately carry none of these.
+ if !self.is_deletion_vector() {
+ return Ok(());
+ }
+
+ let missing = if self.referenced_data_file.is_none() {
+ Some("referenced_data_file")
+ } else if self.content_offset.is_none() {
+ Some("content_offset")
+ } else if self.content_size_in_bytes.is_none() {
+ Some("content_size_in_bytes")
+ } else if self.record_count.is_none() {
+ Some("record_count")
+ } else {
+ None
+ };
+
+ if let Some(field) = missing {
+ return Err(invalid_data!(
+ "deletion vector {} is missing {field}",
+ self.file_path
+ ));
+ }
+
+ Ok(())
+ }
+}
+
+impl From<FileScanTaskDeleteFile> for Result<FileScanTaskDeleteFile> {
+ fn from(task: FileScanTaskDeleteFile) -> Self {
+ task.validate()?;
+ Ok(task)
+ }
}
#[cfg(test)]
mod tests {
+ use std::collections::HashMap;
+
use super::*;
use crate::ErrorKind;
- use crate::spec::{Literal, NestedField, PrimitiveType, Transform, Type};
+ use crate::spec::{
+ DataFileBuilder, Literal, ManifestEntry, ManifestStatus, NestedField, PrimitiveType,
+ Transform, Type,
+ };
fn build_file_scan_task(
schema: SchemaRef,
@@ -630,6 +814,34 @@ mod tests {
(schema, partition_spec)
}
+ fn build_delete_file_task(
+ file_type: DataContentType,
+ file_format: DataFileFormat,
+ ) -> Result<FileScanTaskDeleteFile> {
+ FileScanTaskDeleteFile::builder()
+ .with_file_path("delete-file".to_string())
+ .with_file_size_in_bytes(100)
+ .with_file_type(file_type)
+ .with_file_format(file_format)
+ .with_partition_spec_id(0)
+ .build()
+ }
+
+ fn assert_delete_file_builder_error(
+ result: Result<FileScanTaskDeleteFile>,
+ expected_message: &str,
+ ) {
+ match result {
+ Ok(task) => panic!(
+ "expected delete file builder to fail with `{expected_message}`, but got Ok({task:?})"
+ ),
+ Err(err) => {
+ assert_eq!(err.kind(), ErrorKind::DataInvalid);
+ assert_eq!(err.message(), expected_message);
+ }
+ }
+ }
+
#[test]
fn test_file_scan_task_builder_rejects_non_empty_partition_without_spec() {
// Regression test for https://github.com/apache/iceberg-rust/issues/3130.
@@ -739,4 +951,218 @@ mod tests {
assert_eq!(err.kind(), ErrorKind::DataInvalid);
}
+
+ #[test]
+ fn test_delete_file_builder_accepts_valid_deletion_vector() {
+ let task = FileScanTaskDeleteFile::builder()
+ .with_file_path("dv.puffin".to_string())
+ .with_file_size_in_bytes(100)
+ .with_file_type(DataContentType::PositionDeletes)
+ .with_file_format(DataFileFormat::Puffin)
+ .with_partition_spec_id(7)
+ .with_referenced_data_file(Some("data.parquet".to_string()))
+ .with_content_offset(Some(11))
+ .with_content_size_in_bytes(Some(13))
+ .with_record_count(Some(3))
+ .with_key_metadata(Some(vec![17, 19].into_boxed_slice()))
+ .build()
+ .unwrap();
+
+ assert_eq!(task.file_path(), "dv.puffin");
+ assert_eq!(task.file_size_in_bytes(), 100);
+ assert_eq!(task.file_type(), DataContentType::PositionDeletes);
+ assert_eq!(task.file_format(), DataFileFormat::Puffin);
+ assert_eq!(task.partition_spec_id(), 7);
+ assert_eq!(task.equality_ids(), None);
+ assert_eq!(task.referenced_data_file(), Some("data.parquet"));
+ assert_eq!(task.content_offset(), Some(11));
+ assert_eq!(task.content_size_in_bytes(), Some(13));
+ assert_eq!(task.record_count(), Some(3));
+ assert_eq!(task.key_metadata(), Some([17, 19].as_slice()));
+ }
+
+ #[test]
+ fn test_delete_file_builder_rejects_dv_missing_referenced_data_file() {
+ assert_delete_file_builder_error(
+ FileScanTaskDeleteFile::builder()
+ .with_file_path("dv.puffin".to_string())
+ .with_file_size_in_bytes(100)
+ .with_file_type(DataContentType::PositionDeletes)
+ .with_file_format(DataFileFormat::Puffin)
+ .with_partition_spec_id(0)
+ .with_content_offset(Some(7))
+ .with_content_size_in_bytes(Some(11))
+ .with_record_count(Some(3))
+ .build(),
+ "deletion vector dv.puffin is missing referenced_data_file",
+ );
+ }
+
+ #[test]
+ fn test_delete_file_builder_rejects_dv_missing_content_offset() {
+ assert_delete_file_builder_error(
+ FileScanTaskDeleteFile::builder()
+ .with_file_path("dv.puffin".to_string())
+ .with_file_size_in_bytes(100)
+ .with_file_type(DataContentType::PositionDeletes)
+ .with_file_format(DataFileFormat::Puffin)
+ .with_partition_spec_id(0)
+ .with_referenced_data_file(Some("data.parquet".to_string()))
+ .with_content_size_in_bytes(Some(11))
+ .with_record_count(Some(3))
+ .build(),
+ "deletion vector dv.puffin is missing content_offset",
+ );
+ }
+
+ #[test]
+ fn test_delete_file_builder_rejects_dv_missing_content_size() {
+ assert_delete_file_builder_error(
+ FileScanTaskDeleteFile::builder()
+ .with_file_path("dv.puffin".to_string())
+ .with_file_size_in_bytes(100)
+ .with_file_type(DataContentType::PositionDeletes)
+ .with_file_format(DataFileFormat::Puffin)
+ .with_partition_spec_id(0)
+ .with_referenced_data_file(Some("data.parquet".to_string()))
+ .with_content_offset(Some(7))
+ .with_record_count(Some(3))
+ .build(),
+ "delete file dv.puffin is missing content_size_in_bytes while content_offset is present",
+ );
+ }
+
+ #[test]
+ fn test_delete_file_builder_rejects_dv_missing_record_count() {
+ assert_delete_file_builder_error(
+ FileScanTaskDeleteFile::builder()
+ .with_file_path("dv.puffin".to_string())
+ .with_file_size_in_bytes(100)
+ .with_file_type(DataContentType::PositionDeletes)
+ .with_file_format(DataFileFormat::Puffin)
+ .with_partition_spec_id(0)
+ .with_referenced_data_file(Some("data.parquet".to_string()))
+ .with_content_offset(Some(7))
+ .with_content_size_in_bytes(Some(11))
+ .build(),
+ "deletion vector dv.puffin is missing record_count",
+ );
+ }
+
+ #[test]
+ fn test_delete_file_builder_rejects_non_dv_offset_without_size() {
+ // Spec pairing rule applies to every delete-file kind: a Parquet
+ // position-delete file that carries a content_offset must also carry a
+ // content_size_in_bytes, even though it is not a deletion vector.
+ assert_delete_file_builder_error(
+ FileScanTaskDeleteFile::builder()
+ .with_file_path("position-deletes.parquet".to_string())
+ .with_file_size_in_bytes(100)
+ .with_file_type(DataContentType::PositionDeletes)
+ .with_file_format(DataFileFormat::Parquet)
+ .with_partition_spec_id(0)
+ .with_content_offset(Some(7))
+ .build(),
+ "delete file position-deletes.parquet is missing content_size_in_bytes while content_offset is present",
+ );
+ }
+
+ fn delete_manifest_entry(
+ file_path: &str,
+ file_format: DataFileFormat,
+ content_offset: Option<i64>,
+ content_size_in_bytes: Option<i64>,
+ ) -> DeleteFileContext {
+ let mut data_file = DataFileBuilder::default()
+ .content(DataContentType::PositionDeletes)
+ .file_path(file_path.to_string())
+ .file_format(file_format)
+ .partition(Struct::empty())
+ .record_count(3)
+ .file_size_in_bytes(100)
+ .column_sizes(HashMap::new())
+ .value_counts(HashMap::new())
+ .null_value_counts(HashMap::new())
+ .partition_spec_id(0)
+ .referenced_data_file(Some("data.parquet".to_string()))
+ .build()
+ .unwrap();
+ data_file.content_offset = content_offset;
+ data_file.content_size_in_bytes = content_size_in_bytes;
+
+ DeleteFileContext {
+ manifest_entry: Arc::new(
+ ManifestEntry::builder()
+ .status(ManifestStatus::Added)
+ .data_file(data_file)
+ .build(),
+ ),
+ partition_spec_id: 0,
+ }
+ }
+
+ fn assert_delete_context_error(ctx: &DeleteFileContext, expected_message: &str) {
+ match FileScanTaskDeleteFile::try_from(ctx) {
+ Ok(task) => panic!(
+ "expected conversion to fail with `{expected_message}`, but got Ok({task:?})"
+ ),
+ Err(error) => {
+ assert_eq!(error.kind(), ErrorKind::DataInvalid);
+ assert!(
+ error.message().contains(expected_message),
+ "expected `{expected_message}`, got `{}`",
+ error.message()
+ );
+ }
+ }
+ }
+
+ /// The manifest entry supplies `content_offset` / `content_size_in_bytes` as
+ /// `i64`; the conversion into a task rejects a negative value as it enters,
+ /// since the task fields are `u64`.
+ #[test]
+ fn test_delete_context_rejects_negative_dv_offset() {
+ assert_delete_context_error(
+ &delete_manifest_entry("dv.puffin", DataFileFormat::Puffin, Some(-1), Some(11)),
+ "delete file dv.puffin has negative content_offset -1",
+ );
+ }
+
+ #[test]
+ fn test_delete_context_rejects_negative_dv_size() {
+ assert_delete_context_error(
+ &delete_manifest_entry("dv.puffin", DataFileFormat::Puffin, Some(7), Some(-1)),
+ "delete file dv.puffin has negative content_size_in_bytes -1",
+ );
+ }
+
+ /// The negative-coordinate check applies to every delete-file kind, including
+ /// a Parquet position-delete file rather than only deletion vectors.
+ #[test]
+ fn test_delete_context_rejects_negative_coordinates_for_non_dv() {
+ assert_delete_context_error(
+ &delete_manifest_entry(
+ "position-deletes.parquet",
+ DataFileFormat::Parquet,
+ Some(-1),
+ None,
+ ),
+ "delete file position-deletes.parquet has negative content_offset -1",
+ );
+ assert_delete_context_error(
+ &delete_manifest_entry(
+ "position-deletes.parquet",
+ DataFileFormat::Parquet,
+ None,
+ Some(-1),
+ ),
+ "delete file position-deletes.parquet has negative content_size_in_bytes -1",
+ );
+ }
+
+ #[test]
+ fn test_delete_file_builder_accepts_non_dv_delete_without_dv_fields() {
+ build_delete_file_task(DataContentType::PositionDeletes, DataFileFormat::Parquet).unwrap();
+ build_delete_file_task(DataContentType::EqualityDeletes, DataFileFormat::Parquet).unwrap();
+ }
}