From 25a34eba06547d38184579f50bdb299dba2a19c6 Mon Sep 17 00:00:00 2001 From: JeonDaehong Date: Tue, 13 Oct 2026 21:53:27 +0900 Subject: [PATCH 1/3] feat(reader): resolve unknown delete-file sizes lazily Execution engines that build scan tasks themselves do not always carry a delete file's size. Tasks planned from manifest entries keep the recorded size; externally constructed tasks may now leave FileScanTaskDeleteFile's file size unset, and the Parquet delete-file loader resolves it with one metadata request when the file is first loaded. The caching loader loads each delete file once per reader, so the request is made once per delete file even when several data files reference it. FileScanTaskDeleteFile::file_size_in_bytes now returns Option. None means unknown, so a size of 0 is no longer a sentinel and is rejected like any other size below the Parquet footer minimum. A failed metadata request is reported with the delete-file path while keeping the storage error's kind and retryability. A size below the Parquet footer minimum, whether resolved or supplied by the task, fails the read rather than dropping the file's deletes. Deletion vectors do not use this size and are unchanged. Co-authored-by: unikdahal --- crates/iceberg/public-api.txt | 2 +- .../src/arrow/caching_delete_file_loader.rs | 330 +++++++++++++++++- .../iceberg/src/arrow/delete_file_loader.rs | 136 +++++++- crates/iceberg/src/scan/task.rs | 22 +- 4 files changed, 461 insertions(+), 29 deletions(-) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 4666147573..8d2030c5c0 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1373,7 +1373,7 @@ pub fn iceberg::scan::FileScanTaskDeleteFile::content_size_in_bytes(&self) -> co 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_size_in_bytes(&self) -> core::option::Option 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 diff --git a/crates/iceberg/src/arrow/caching_delete_file_loader.rs b/crates/iceberg/src/arrow/caching_delete_file_loader.rs index 5582e4397f..2a4a83f12f 100644 --- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs +++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs @@ -873,7 +873,7 @@ mod tests { let record_batch_stream = basic_delete_file_loader .parquet_to_batch_stream( &eq_delete_file_path, - std::fs::metadata(&eq_delete_file_path).unwrap().len(), + Some(std::fs::metadata(&eq_delete_file_path).unwrap().len()), None, ) .await @@ -1129,6 +1129,330 @@ mod tests { assert!(result.is_none()); // no pos dels for file 3 } + /// Local filesystem storage that counts metadata requests. + #[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)] + struct MetadataCountingStorage { + #[serde(skip)] + calls: Arc, + /// Fail every metadata request with a retryable error. + #[serde(skip)] + fail_metadata: bool, + } + + #[async_trait::async_trait] + #[typetag::serde(name = "test_metadata_counting_storage")] + impl crate::io::Storage for MetadataCountingStorage { + async fn exists(&self, path: &str) -> Result { + crate::io::LocalFsStorage::new().exists(path).await + } + + async fn metadata(&self, path: &str) -> Result { + self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + if self.fail_metadata { + return Err( + Error::new(ErrorKind::Unexpected, "metadata temporarily unavailable") + .with_retryable(true), + ); + } + crate::io::LocalFsStorage::new().metadata(path).await + } + + async fn read(&self, path: &str) -> Result { + crate::io::LocalFsStorage::new().read(path).await + } + + async fn reader(&self, path: &str) -> Result> { + crate::io::LocalFsStorage::new().reader(path).await + } + + async fn write(&self, path: &str, bs: Bytes) -> Result<()> { + crate::io::LocalFsStorage::new().write(path, bs).await + } + + async fn writer(&self, path: &str) -> Result> { + crate::io::LocalFsStorage::new().writer(path).await + } + + async fn delete(&self, path: &str) -> Result<()> { + crate::io::LocalFsStorage::new().delete(path).await + } + + async fn delete_prefix(&self, path: &str) -> Result<()> { + crate::io::LocalFsStorage::new().delete_prefix(path).await + } + + async fn delete_stream( + &self, + paths: futures::stream::BoxStream<'static, String>, + ) -> Result<()> { + crate::io::LocalFsStorage::new().delete_stream(paths).await + } + + fn new_input(&self, path: &str) -> Result { + Ok(crate::io::InputFile::new( + Arc::new(self.clone()), + path.to_string(), + )) + } + + fn new_output(&self, path: &str) -> Result { + Ok(crate::io::OutputFile::new( + Arc::new(self.clone()), + path.to_string(), + )) + } + } + + #[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)] + struct MetadataCountingStorageFactory { + #[serde(skip)] + calls: Arc, + #[serde(skip)] + fail_metadata: bool, + } + + #[typetag::serde(name = "test_metadata_counting_storage_factory")] + impl crate::io::StorageFactory for MetadataCountingStorageFactory { + fn build(&self, _config: &crate::io::StorageConfig) -> Result> { + Ok(Arc::new(MetadataCountingStorage { + calls: Arc::clone(&self.calls), + fail_metadata: self.fail_metadata, + })) + } + } + + /// Rebuilds `delete` at `path` without a file size. + fn without_size(delete: &FileScanTaskDeleteFile, path: &str) -> FileScanTaskDeleteFile { + FileScanTaskDeleteFile::builder() + .with_file_path(path.to_string()) + .with_file_type(delete.file_type()) + .with_file_format(delete.file_format()) + .with_partition_spec_id(delete.partition_spec_id()) + .with_equality_ids(delete.equality_ids().map(<[i32]>::to_vec)) + .with_referenced_data_file(delete.referenced_data_file().map(str::to_string)) + .with_content_offset(delete.content_offset()) + .with_content_size_in_bytes(delete.content_size_in_bytes()) + .with_record_count(delete.record_count()) + .with_key_metadata(delete.key_metadata().map(Box::from)) + .build() + .unwrap() + } + + #[tokio::test] + async fn test_unknown_delete_sizes_are_resolved_once_per_file() { + use std::sync::atomic::Ordering; + + let tmp_dir = TempDir::new().unwrap(); + let file_scan_tasks = setup(tmp_dir.path()); + let factory = Arc::new(MetadataCountingStorageFactory::default()); + let calls = Arc::clone(&factory.calls); + let file_io = crate::io::FileIOBuilder::new(factory).build(); + let task = &file_scan_tasks[0]; + let unknown: Vec<_> = task + .deletes() + .iter() + .map(|delete| without_size(delete, delete.file_path())) + .collect(); + let positions = |filter: &DeleteFilter| -> Vec { + filter + .get_delete_vector(task) + .unwrap() + .lock() + .unwrap() + .iter() + .collect() + }; + + // Known sizes never request metadata. + let known = CachingDeleteFileLoader::new(file_io.clone(), 10, Runtime::current()) + .load_deletes(task.deletes(), task.schema_ref()) + .await + .unwrap() + .unwrap(); + assert_eq!(calls.load(Ordering::SeqCst), 0); + + // Unknown sizes: one request per delete file, and none again when + // another task references an already loaded delete file. + let loader = CachingDeleteFileLoader::new(file_io, 10, Runtime::current()); + let lazy = loader + .load_deletes(&unknown, task.schema_ref()) + .await + .unwrap() + .unwrap(); + assert_eq!(calls.load(Ordering::SeqCst), unknown.len()); + loader + .load_deletes(&unknown[1..], task.schema_ref()) + .await + .unwrap() + .unwrap(); + assert_eq!(calls.load(Ordering::SeqCst), unknown.len()); + + // The same positions are deleted either way. + let expected = positions(&known); + assert_eq!(expected.len(), 12); + assert_eq!(positions(&lazy), expected); + } + + #[tokio::test] + async fn test_unknown_size_equality_deletes_are_resolved_once_and_cached() { + use std::sync::atomic::Ordering; + + use crate::scan::FileScanTask; + use crate::spec::DataFileFormat; + + let tmp_dir = TempDir::new().unwrap(); + let table_location = tmp_dir.path().to_str().unwrap(); + let eq_delete_path = setup_write_equality_delete_file_1(table_location); + let schema = Arc::new( + Schema::builder() + .with_fields(vec![ + NestedField::optional(2, "y", Type::Primitive(PrimitiveType::Long)).into(), + NestedField::optional(3, "z", Type::Primitive(PrimitiveType::Long)).into(), + ]) + .build() + .unwrap(), + ); + let equality_delete = |file_size_in_bytes| { + FileScanTaskDeleteFile::builder() + .with_file_path(eq_delete_path.clone()) + .with_file_size_in_bytes_opt(file_size_in_bytes) + .with_file_type(DataContentType::EqualityDeletes) + .with_file_format(DataFileFormat::Parquet) + .with_partition_spec_id(0) + .with_equality_ids(Some(vec![2, 3])) + .build() + .unwrap() + }; + let data_task = |name: &str, delete| { + FileScanTask::builder() + .with_file_size_in_bytes(0) + .with_start(0) + .with_length(0) + .with_data_file_path(format!("{table_location}/{name}")) + .with_data_file_format(DataFileFormat::Parquet) + .with_schema(Arc::clone(&schema)) + .with_project_field_ids(vec![2, 3]) + .with_deletes(vec![delete]) + .with_case_sensitive(false) + .build() + .unwrap() + }; + let known_size = std::fs::metadata(&eq_delete_path).unwrap().len(); + + let factory = Arc::new(MetadataCountingStorageFactory::default()); + let calls = Arc::clone(&factory.calls); + let file_io = crate::io::FileIOBuilder::new(factory).build(); + + // Known size: no metadata request. + let known_task = data_task("known.parquet", equality_delete(Some(known_size))); + let known = CachingDeleteFileLoader::new(file_io.clone(), 10, Runtime::current()) + .load_deletes(known_task.deletes(), known_task.schema_ref()) + .await + .unwrap() + .unwrap() + .build_equality_delete_predicate(&known_task) + .await + .unwrap(); + assert!(known.is_some()); + assert_eq!(calls.load(Ordering::SeqCst), 0); + + // Unknown size: one metadata request, and none when a second data + // file referencing the same equality delete file is loaded. + let loader = CachingDeleteFileLoader::new(file_io, 10, Runtime::current()); + for (name, expected_calls) in [("first.parquet", 1), ("second.parquet", 1)] { + let task = data_task(name, equality_delete(None)); + let predicate = loader + .load_deletes(task.deletes(), task.schema_ref()) + .await + .unwrap() + .unwrap() + .build_equality_delete_predicate(&task) + .await + .unwrap(); + assert_eq!(predicate, known, "{name}"); + assert_eq!(calls.load(Ordering::SeqCst), expected_calls, "{name}"); + } + } + + #[tokio::test] + async fn test_load_deletes_reports_unknown_size_failures() { + let tmp_dir = TempDir::new().unwrap(); + let file_scan_tasks = setup(tmp_dir.path()); + let task = &file_scan_tasks[0]; + let unknown_size = |path: &str| without_size(&task.deletes()[0], path); + + // A missing delete file fails the load instead of dropping its deletes. + let missing = unknown_size(&format!( + "{}/missing-pos-del.parquet", + tmp_dir.path().to_str().unwrap() + )); + let error = CachingDeleteFileLoader::new(FileIO::new_with_fs(), 10, Runtime::current()) + .load_deletes(&[missing], task.schema_ref()) + .await + .unwrap() + .err() + .unwrap(); + assert!( + error.to_string().contains("Failed to stat delete file"), + "{error}" + ); + + // A retryable storage failure stays retryable through the loader. + let factory = Arc::new(MetadataCountingStorageFactory { + fail_metadata: true, + ..Default::default() + }); + let file_io = crate::io::FileIOBuilder::new(factory).build(); + let present = unknown_size(task.deletes()[0].file_path()); + let error = CachingDeleteFileLoader::new(file_io, 10, Runtime::current()) + .load_deletes(&[present], task.schema_ref()) + .await + .unwrap() + .err() + .unwrap(); + assert!( + error.to_string().contains("Failed to stat delete file"), + "{error}" + ); + assert_eq!(error.kind(), ErrorKind::Unexpected); + assert!(error.retryable()); + } + + #[tokio::test] + async fn test_delete_file_loader_parses_equality_deletes_of_unknown_size() { + let tmp_dir = TempDir::new().unwrap(); + let table_location = tmp_dir.path().as_os_str().to_str().unwrap(); + let eq_delete_file_path = setup_write_equality_delete_file_1(table_location); + + let basic_delete_file_loader = + BasicDeleteFileLoader::new(FileIO::new_with_fs(), ScanMetrics::new()); + let known = basic_delete_file_loader + .parquet_to_batch_stream( + &eq_delete_file_path, + Some(std::fs::metadata(&eq_delete_file_path).unwrap().len()), + None, + ) + .await + .unwrap(); + let unknown = basic_delete_file_loader + .parquet_to_batch_stream(&eq_delete_file_path, None, None) + .await + .unwrap(); + + let eq_ids = HashSet::from_iter(vec![2, 3, 4, 6, 8]); + let known = CachingDeleteFileLoader::parse_equality_deletes_record_batch_stream( + known, + eq_ids.clone(), + ) + .await + .unwrap(); + let unknown = + CachingDeleteFileLoader::parse_equality_deletes_record_batch_stream(unknown, eq_ids) + .await + .unwrap(); + assert_eq!(unknown, known); + } + #[tokio::test] async fn test_parse_positional_deletes_rejects_negative_positions() { let schema = crate::arrow::delete_filter::tests::create_pos_del_schema(); @@ -1312,7 +1636,7 @@ mod tests { let batch_stream = basic_delete_file_loader .parquet_to_batch_stream( &delete_file_path, - std::fs::metadata(&delete_file_path).unwrap().len(), + Some(std::fs::metadata(&delete_file_path).unwrap().len()), None, ) .await @@ -1490,7 +1814,7 @@ mod tests { let basic_delete_file_loader = BasicDeleteFileLoader::new(file_io.clone(), ScanMetrics::new()); let record_batch_stream = basic_delete_file_loader - .parquet_to_batch_stream(&path, std::fs::metadata(&path).unwrap().len(), None) + .parquet_to_batch_stream(&path, Some(std::fs::metadata(&path).unwrap().len()), None) .await .expect("could not get batch stream"); diff --git a/crates/iceberg/src/arrow/delete_file_loader.rs b/crates/iceberg/src/arrow/delete_file_loader.rs index ec61710a42..0befe5e075 100644 --- a/crates/iceberg/src/arrow/delete_file_loader.rs +++ b/crates/iceberg/src/arrow/delete_file_loader.rs @@ -66,13 +66,45 @@ impl BasicDeleteFileLoader { pub(crate) async fn parquet_to_batch_stream( &self, data_file_path: &str, - file_size_in_bytes: u64, + file_size_in_bytes: Option, key_metadata: Option<&[u8]>, ) -> Result { /* Essentially a super-cut-down ArrowReader. We can't use ArrowReader directly as that introduces a circular dependency. */ + // `None` means the task did not carry a size (see `FileScanTaskDeleteFile`). + // Resolve it when the file is first loaded. + let file_size_in_bytes = match file_size_in_bytes { + Some(file_size_in_bytes) => file_size_in_bytes, + None => { + self.file_io + .new_input(data_file_path)? + .metadata() + .await + .map_err(|error| { + // Keep the storage error's kind and retryability. + Error::new( + error.kind(), + format!("Failed to stat delete file '{data_file_path}'"), + ) + .with_retryable(error.retryable()) + .with_source(error) + })? + .size + } + }; + // Reading a Parquet footer needs at least its last eight bytes. Reject + // a smaller size rather than read the file as if it had no deletes. + if file_size_in_bytes < 8 { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "Delete file '{data_file_path}' has a size of {file_size_in_bytes} bytes, \ + below the Parquet footer minimum" + ), + )); + } let parquet_read_options = ParquetReadOptions::builder().build(); let (parquet_file_reader, arrow_metadata) = ArrowReader::open_parquet_file( @@ -230,27 +262,30 @@ mod tests { .unwrap(), ); - 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); - let result = delete_file_loader - .read_delete_file(&task, schema) - .await - .unwrap(); + // A known size, and an unknown size resolved before decryption. + for file_size_in_bytes in [Some(std::fs::metadata(&del_path).unwrap().len()), None] { + let task = FileScanTaskDeleteFile::builder() + .with_file_path(del_path.clone()) + .with_file_size_in_bytes_opt(file_size_in_bytes) + .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 batches: Vec<_> = result.try_collect().await.unwrap(); - assert_eq!(batches.len(), 1); - assert_eq!(batches[0].num_rows(), 4); + let result = delete_file_loader + .read_delete_file(&task, Arc::clone(&schema)) + .await + .unwrap(); + + let batches: Vec<_> = result.try_collect().await.unwrap(); + assert_eq!(batches.len(), 1); + assert_eq!(batches[0].num_rows(), 4); + } } #[tokio::test] @@ -328,4 +363,69 @@ mod tests { assert_eq!(batches.len(), 1); assert_eq!(batches[0].num_rows(), 3); } + + #[tokio::test] + async fn unknown_delete_size_still_reports_missing_and_empty_objects() { + let temp = TempDir::new().unwrap(); + let loader = BasicDeleteFileLoader::new(FileIO::new_with_fs(), ScanMetrics::new()); + let missing = temp.path().join("missing.parquet"); + let error = loader + .parquet_to_batch_stream(missing.to_str().unwrap(), None, None) + .await + .err() + .unwrap(); + assert!(error.to_string().contains("Failed to stat delete file")); + // The storage error's kind and retryability are kept exactly. + let storage_error = FileIO::new_with_fs() + .new_input(missing.to_str().unwrap()) + .unwrap() + .metadata() + .await + .err() + .unwrap(); + assert_eq!(error.kind(), storage_error.kind()); + assert_eq!(error.retryable(), storage_error.retryable()); + + // A size below the Parquet footer minimum is rejected whether it was + // resolved from storage or supplied by the task. + let empty = temp.path().join("empty.parquet"); + std::fs::write(&empty, []).unwrap(); + for size in [None, Some(0), Some(4)] { + let error = loader + .parquet_to_batch_stream(empty.to_str().unwrap(), size, None) + .await + .err() + .unwrap(); + assert_eq!(error.kind(), ErrorKind::DataInvalid); + assert!(error.to_string().contains("footer minimum"), "{error}"); + } + + // A real file with an unknown size is sized lazily and read in full. + let present = temp.path().join("present.parquet"); + let schema = Arc::new(arrow_schema::Schema::new(vec![arrow_schema::Field::new( + "pos", + arrow_schema::DataType::Int64, + false, + )])); + let batch = arrow_array::RecordBatch::try_new(Arc::clone(&schema), vec![Arc::new( + arrow_array::Int64Array::from(vec![1, 2, 3]), + )]) + .unwrap(); + let mut writer = parquet::arrow::ArrowWriter::try_new( + std::fs::File::create(&present).unwrap(), + schema, + None, + ) + .unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + let batches: Vec<_> = loader + .parquet_to_batch_stream(present.to_str().unwrap(), None, None) + .await + .unwrap() + .try_collect() + .await + .unwrap(); + assert_eq!(batches.iter().map(|b| b.num_rows()).sum::(), 3); + } } diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs index 45940b0dfd..9a64a59f9d 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -366,8 +366,14 @@ pub struct FileScanTaskDeleteFile { /// The delete file path file_path: String, - /// The total size of the delete file in bytes, from the manifest entry. - file_size_in_bytes: u64, + /// The total size of the delete file in bytes. Tasks planned from manifest entries carry + /// the recorded size. An externally constructed task may leave it unset when the size is + /// unavailable; the Parquet delete-file loader then resolves it with one metadata request + /// when the file is first loaded. Deletion vectors do not use it. + #[serde(default)] + #[serde(skip_serializing_if = "Option::is_none")] + #[builder(default, setter(strip_option(fallback = with_file_size_in_bytes_opt)))] + file_size_in_bytes: Option, /// delete file type file_type: DataContentType, @@ -612,7 +618,8 @@ mod _serde { #[derive(Deserialize)] pub(super) struct FileScanTaskDeleteFileSerde { file_path: String, - file_size_in_bytes: u64, + #[serde(default)] + file_size_in_bytes: Option, file_type: DataContentType, file_format: DataFileFormat, partition_spec_id: i32, @@ -635,7 +642,7 @@ mod _serde { fn try_from(value: FileScanTaskDeleteFileSerde) -> Result { Self::builder() .with_file_path(value.file_path) - .with_file_size_in_bytes(value.file_size_in_bytes) + .with_file_size_in_bytes_opt(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) @@ -656,8 +663,9 @@ impl FileScanTaskDeleteFile { &self.file_path } - /// Returns the total size of the delete file in bytes. - pub fn file_size_in_bytes(&self) -> u64 { + /// Returns the total size of the delete file in bytes, or `None` if the task does not + /// carry it. + pub fn file_size_in_bytes(&self) -> Option { self.file_size_in_bytes } @@ -969,7 +977,7 @@ mod tests { .unwrap(); assert_eq!(task.file_path(), "dv.puffin"); - assert_eq!(task.file_size_in_bytes(), 100); + assert_eq!(task.file_size_in_bytes(), Some(100)); assert_eq!(task.file_type(), DataContentType::PositionDeletes); assert_eq!(task.file_format(), DataFileFormat::Puffin); assert_eq!(task.partition_spec_id(), 7); From b2981e2ead534ade399992edfd88d4cb27df7fd5 Mon Sep 17 00:00:00 2001 From: JeonDaehong Date: Tue, 13 Oct 2026 23:10:04 +0900 Subject: [PATCH 2/3] fix(reader): rely on Parquet's footer check for delete-file sizes Parquet already rejects a size below its footer, so drop the separate 8-byte check and assert Parquet's error in the test instead. Also document leaving the size unset when the recorded one can't be trusted (apache/iceberg#12554), add a serde round trip for a delete file without a size, and tidy the metadata-counting test storage. --- .../src/arrow/caching_delete_file_loader.rs | 107 ++++++++++-------- .../iceberg/src/arrow/delete_file_loader.rs | 16 +-- crates/iceberg/src/scan/task.rs | 7 +- 3 files changed, 68 insertions(+), 62 deletions(-) diff --git a/crates/iceberg/src/arrow/caching_delete_file_loader.rs b/crates/iceberg/src/arrow/caching_delete_file_loader.rs index 2a4a83f12f..25763d145b 100644 --- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs +++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs @@ -823,12 +823,14 @@ mod tests { use std::collections::HashMap; use std::fs::File; use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; use arrow_array::cast::AsArray; use arrow_array::{ ArrayRef, BinaryArray, Int32Array, Int64Array, RecordBatch, StringArray, StructArray, }; use arrow_schema::{DataType, Field, Fields}; + use futures::stream::BoxStream; use parquet::arrow::{ArrowWriter, PARQUET_FIELD_ID_META_KEY}; use parquet::basic::Compression; use parquet::file::properties::WriterProperties; @@ -836,10 +838,33 @@ mod tests { use super::*; use crate::arrow::delete_filter::tests::setup; + use crate::io::{ + FileIOBuilder, FileMetadata, FileRead, FileWrite, InputFile, LocalFsStorage, OutputFile, + Storage, StorageConfig, StorageFactory, + }; use crate::scan::FileScanTaskDeleteFile; use crate::spec::{DataContentType, DataFileFormat, Schema}; use crate::test_utils::encode_dv_blob; + #[test] + fn test_delete_file_without_size_round_trips_through_serde() { + let task = FileScanTaskDeleteFile::builder() + .with_file_path("pos-del.parquet".to_string()) + .with_file_type(DataContentType::PositionDeletes) + .with_file_format(DataFileFormat::Parquet) + .with_partition_spec_id(0) + .build() + .unwrap(); + assert_eq!(task.file_size_in_bytes(), None); + + // The unknown size is left out when serializing, and a task without the + // field deserializes to `None`. + let json = serde_json::to_value(&task).unwrap(); + assert!(json.get("file_size_in_bytes").is_none(), "{json}"); + let round_tripped: FileScanTaskDeleteFile = serde_json::from_value(json).unwrap(); + assert_eq!(round_tripped, task); + } + #[test] fn test_deserialized_deletion_vector_rejects_missing_fields() { // A deletion vector that omits its required coordinates fails validation @@ -1133,7 +1158,9 @@ mod tests { #[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)] struct MetadataCountingStorage { #[serde(skip)] - calls: Arc, + inner: LocalFsStorage, + #[serde(skip)] + calls: Arc, /// Fail every metadata request with a retryable error. #[serde(skip)] fail_metadata: bool, @@ -1141,80 +1168,72 @@ mod tests { #[async_trait::async_trait] #[typetag::serde(name = "test_metadata_counting_storage")] - impl crate::io::Storage for MetadataCountingStorage { - async fn exists(&self, path: &str) -> Result { - crate::io::LocalFsStorage::new().exists(path).await - } - - async fn metadata(&self, path: &str) -> Result { - self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + impl Storage for MetadataCountingStorage { + async fn metadata(&self, path: &str) -> Result { + self.calls.fetch_add(1, Ordering::SeqCst); if self.fail_metadata { return Err( Error::new(ErrorKind::Unexpected, "metadata temporarily unavailable") .with_retryable(true), ); } - crate::io::LocalFsStorage::new().metadata(path).await + self.inner.metadata(path).await } - async fn read(&self, path: &str) -> Result { - crate::io::LocalFsStorage::new().read(path).await + fn new_input(&self, path: &str) -> Result { + Ok(InputFile::new(Arc::new(self.clone()), path.to_string())) } - async fn reader(&self, path: &str) -> Result> { - crate::io::LocalFsStorage::new().reader(path).await + fn new_output(&self, path: &str) -> Result { + Ok(OutputFile::new(Arc::new(self.clone()), path.to_string())) } - async fn write(&self, path: &str, bs: Bytes) -> Result<()> { - crate::io::LocalFsStorage::new().write(path, bs).await + async fn exists(&self, path: &str) -> Result { + self.inner.exists(path).await } - async fn writer(&self, path: &str) -> Result> { - crate::io::LocalFsStorage::new().writer(path).await + async fn read(&self, path: &str) -> Result { + self.inner.read(path).await } - async fn delete(&self, path: &str) -> Result<()> { - crate::io::LocalFsStorage::new().delete(path).await + async fn reader(&self, path: &str) -> Result> { + self.inner.reader(path).await } - async fn delete_prefix(&self, path: &str) -> Result<()> { - crate::io::LocalFsStorage::new().delete_prefix(path).await + async fn write(&self, path: &str, bs: Bytes) -> Result<()> { + self.inner.write(path, bs).await } - async fn delete_stream( - &self, - paths: futures::stream::BoxStream<'static, String>, - ) -> Result<()> { - crate::io::LocalFsStorage::new().delete_stream(paths).await + async fn writer(&self, path: &str) -> Result> { + self.inner.writer(path).await } - fn new_input(&self, path: &str) -> Result { - Ok(crate::io::InputFile::new( - Arc::new(self.clone()), - path.to_string(), - )) + async fn delete(&self, path: &str) -> Result<()> { + self.inner.delete(path).await } - fn new_output(&self, path: &str) -> Result { - Ok(crate::io::OutputFile::new( - Arc::new(self.clone()), - path.to_string(), - )) + async fn delete_prefix(&self, path: &str) -> Result<()> { + self.inner.delete_prefix(path).await + } + + async fn delete_stream(&self, paths: BoxStream<'static, String>) -> Result<()> { + self.inner.delete_stream(paths).await } } #[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)] struct MetadataCountingStorageFactory { #[serde(skip)] - calls: Arc, + calls: Arc, #[serde(skip)] fail_metadata: bool, } #[typetag::serde(name = "test_metadata_counting_storage_factory")] - impl crate::io::StorageFactory for MetadataCountingStorageFactory { - fn build(&self, _config: &crate::io::StorageConfig) -> Result> { + impl StorageFactory for MetadataCountingStorageFactory { + fn build(&self, _config: &StorageConfig) -> Result> { Ok(Arc::new(MetadataCountingStorage { + inner: LocalFsStorage, calls: Arc::clone(&self.calls), fail_metadata: self.fail_metadata, })) @@ -1240,13 +1259,11 @@ mod tests { #[tokio::test] async fn test_unknown_delete_sizes_are_resolved_once_per_file() { - use std::sync::atomic::Ordering; - let tmp_dir = TempDir::new().unwrap(); let file_scan_tasks = setup(tmp_dir.path()); let factory = Arc::new(MetadataCountingStorageFactory::default()); let calls = Arc::clone(&factory.calls); - let file_io = crate::io::FileIOBuilder::new(factory).build(); + let file_io = FileIOBuilder::new(factory).build(); let task = &file_scan_tasks[0]; let unknown: Vec<_> = task .deletes() @@ -1295,8 +1312,6 @@ mod tests { #[tokio::test] async fn test_unknown_size_equality_deletes_are_resolved_once_and_cached() { - use std::sync::atomic::Ordering; - use crate::scan::FileScanTask; use crate::spec::DataFileFormat; @@ -1341,7 +1356,7 @@ mod tests { let factory = Arc::new(MetadataCountingStorageFactory::default()); let calls = Arc::clone(&factory.calls); - let file_io = crate::io::FileIOBuilder::new(factory).build(); + let file_io = FileIOBuilder::new(factory).build(); // Known size: no metadata request. let known_task = data_task("known.parquet", equality_delete(Some(known_size))); @@ -1402,7 +1417,7 @@ mod tests { fail_metadata: true, ..Default::default() }); - let file_io = crate::io::FileIOBuilder::new(factory).build(); + let file_io = FileIOBuilder::new(factory).build(); let present = unknown_size(task.deletes()[0].file_path()); let error = CachingDeleteFileLoader::new(file_io, 10, Runtime::current()) .load_deletes(&[present], task.schema_ref()) diff --git a/crates/iceberg/src/arrow/delete_file_loader.rs b/crates/iceberg/src/arrow/delete_file_loader.rs index 0befe5e075..1f0901f93e 100644 --- a/crates/iceberg/src/arrow/delete_file_loader.rs +++ b/crates/iceberg/src/arrow/delete_file_loader.rs @@ -94,17 +94,6 @@ impl BasicDeleteFileLoader { .size } }; - // Reading a Parquet footer needs at least its last eight bytes. Reject - // a smaller size rather than read the file as if it had no deletes. - if file_size_in_bytes < 8 { - return Err(Error::new( - ErrorKind::DataInvalid, - format!( - "Delete file '{data_file_path}' has a size of {file_size_in_bytes} bytes, \ - below the Parquet footer minimum" - ), - )); - } let parquet_read_options = ParquetReadOptions::builder().build(); let (parquet_file_reader, arrow_metadata) = ArrowReader::open_parquet_file( @@ -386,7 +375,7 @@ mod tests { assert_eq!(error.kind(), storage_error.kind()); assert_eq!(error.retryable(), storage_error.retryable()); - // A size below the Parquet footer minimum is rejected whether it was + // A size too small for a Parquet footer fails the read, whether it was // resolved from storage or supplied by the task. let empty = temp.path().join("empty.parquet"); std::fs::write(&empty, []).unwrap(); @@ -396,8 +385,7 @@ mod tests { .await .err() .unwrap(); - assert_eq!(error.kind(), ErrorKind::DataInvalid); - assert!(error.to_string().contains("footer minimum"), "{error}"); + assert!(error.to_string().contains("is less than footer"), "{error}"); } // A real file with an unknown size is sized lazily and read in full. diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs index 9a64a59f9d..60dabc6324 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -368,8 +368,11 @@ pub struct FileScanTaskDeleteFile { /// The total size of the delete file in bytes. Tasks planned from manifest entries carry /// the recorded size. An externally constructed task may leave it unset when the size is - /// unavailable; the Parquet delete-file loader then resolves it with one metadata request - /// when the file is first loaded. Deletion vectors do not use it. + /// unavailable or the recorded size can't be trusted, such as position delete files + /// rewritten by `rewrite_table_path` before Iceberg Java 1.12.0 + /// (). The Parquet delete-file loader then + /// resolves it with one metadata request when the file is first loaded. Deletion vectors + /// do not use it. #[serde(default)] #[serde(skip_serializing_if = "Option::is_none")] #[builder(default, setter(strip_option(fallback = with_file_size_in_bytes_opt)))] From ebc067140ee0689bc04d9aaa79fb2ff8f9eb1caa Mon Sep 17 00:00:00 2001 From: JeonDaehong Date: Wed, 14 Oct 2026 08:18:27 +0900 Subject: [PATCH 3/3] test(reader): drop delete-size tests covered elsewhere Keep only the footer cases in the delete_file_loader test; the missing file and lazy read are covered by test_load_deletes_reports_unknown_size_failures and test_read_encrypted_positional_delete_file. Remove the equality-delete parse test, which test_unknown_size_equality_deletes_are_resolved_once_and_cached already covers through load_deletes. --- .../src/arrow/caching_delete_file_loader.rs | 35 -------------- .../iceberg/src/arrow/delete_file_loader.rs | 47 +------------------ 2 files changed, 1 insertion(+), 81 deletions(-) diff --git a/crates/iceberg/src/arrow/caching_delete_file_loader.rs b/crates/iceberg/src/arrow/caching_delete_file_loader.rs index 25763d145b..e8b4cef4b5 100644 --- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs +++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs @@ -1433,41 +1433,6 @@ mod tests { assert!(error.retryable()); } - #[tokio::test] - async fn test_delete_file_loader_parses_equality_deletes_of_unknown_size() { - let tmp_dir = TempDir::new().unwrap(); - let table_location = tmp_dir.path().as_os_str().to_str().unwrap(); - let eq_delete_file_path = setup_write_equality_delete_file_1(table_location); - - let basic_delete_file_loader = - BasicDeleteFileLoader::new(FileIO::new_with_fs(), ScanMetrics::new()); - let known = basic_delete_file_loader - .parquet_to_batch_stream( - &eq_delete_file_path, - Some(std::fs::metadata(&eq_delete_file_path).unwrap().len()), - None, - ) - .await - .unwrap(); - let unknown = basic_delete_file_loader - .parquet_to_batch_stream(&eq_delete_file_path, None, None) - .await - .unwrap(); - - let eq_ids = HashSet::from_iter(vec![2, 3, 4, 6, 8]); - let known = CachingDeleteFileLoader::parse_equality_deletes_record_batch_stream( - known, - eq_ids.clone(), - ) - .await - .unwrap(); - let unknown = - CachingDeleteFileLoader::parse_equality_deletes_record_batch_stream(unknown, eq_ids) - .await - .unwrap(); - assert_eq!(unknown, known); - } - #[tokio::test] async fn test_parse_positional_deletes_rejects_negative_positions() { let schema = crate::arrow::delete_filter::tests::create_pos_del_schema(); diff --git a/crates/iceberg/src/arrow/delete_file_loader.rs b/crates/iceberg/src/arrow/delete_file_loader.rs index 1f0901f93e..756ce26f73 100644 --- a/crates/iceberg/src/arrow/delete_file_loader.rs +++ b/crates/iceberg/src/arrow/delete_file_loader.rs @@ -354,26 +354,9 @@ mod tests { } #[tokio::test] - async fn unknown_delete_size_still_reports_missing_and_empty_objects() { + async fn test_delete_file_too_small_for_footer_fails() { let temp = TempDir::new().unwrap(); let loader = BasicDeleteFileLoader::new(FileIO::new_with_fs(), ScanMetrics::new()); - let missing = temp.path().join("missing.parquet"); - let error = loader - .parquet_to_batch_stream(missing.to_str().unwrap(), None, None) - .await - .err() - .unwrap(); - assert!(error.to_string().contains("Failed to stat delete file")); - // The storage error's kind and retryability are kept exactly. - let storage_error = FileIO::new_with_fs() - .new_input(missing.to_str().unwrap()) - .unwrap() - .metadata() - .await - .err() - .unwrap(); - assert_eq!(error.kind(), storage_error.kind()); - assert_eq!(error.retryable(), storage_error.retryable()); // A size too small for a Parquet footer fails the read, whether it was // resolved from storage or supplied by the task. @@ -387,33 +370,5 @@ mod tests { .unwrap(); assert!(error.to_string().contains("is less than footer"), "{error}"); } - - // A real file with an unknown size is sized lazily and read in full. - let present = temp.path().join("present.parquet"); - let schema = Arc::new(arrow_schema::Schema::new(vec![arrow_schema::Field::new( - "pos", - arrow_schema::DataType::Int64, - false, - )])); - let batch = arrow_array::RecordBatch::try_new(Arc::clone(&schema), vec![Arc::new( - arrow_array::Int64Array::from(vec![1, 2, 3]), - )]) - .unwrap(); - let mut writer = parquet::arrow::ArrowWriter::try_new( - std::fs::File::create(&present).unwrap(), - schema, - None, - ) - .unwrap(); - writer.write(&batch).unwrap(); - writer.close().unwrap(); - let batches: Vec<_> = loader - .parquet_to_batch_stream(present.to_str().unwrap(), None, None) - .await - .unwrap() - .try_collect() - .await - .unwrap(); - assert_eq!(batches.iter().map(|b| b.num_rows()).sum::(), 3); } }