This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
The following commit(s) were added to refs/heads/main by this push:
new b0d8baaf9 fix(scan): validate deletion vector delete tasks (#3145)
b0d8baaf9 is described below
commit b0d8baaf966813f82363279db46d6688bf455d24
Author: hylin <[email protected]>
AuthorDate: Fri Oct 9 12:36:50 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 <[email protected]>
* 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 <[email protected]>
Co-authored-by: linhongyu510 <[email protected]>
Co-authored-by: linhongyu510 <mailto:[email protected]>
---
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();
+ }
}
