From b2b9b32ddb38e232c8818ecdca4b380c185e4526 Mon Sep 17 00:00:00 2001 From: unikdahal Date: Mon, 5 Oct 2026 02:40:48 +0530 Subject: [PATCH] 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 set FileScanTaskDeleteFile::file_size_in_bytes to 0 when the size is unavailable, 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. 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. --- .../src/arrow/caching_delete_file_loader.rs | 315 ++++++++++++++++++ .../iceberg/src/arrow/delete_file_loader.rs | 143 ++++++-- crates/iceberg/src/scan/task.rs | 7 +- 3 files changed, 442 insertions(+), 23 deletions(-) diff --git a/crates/iceberg/src/arrow/caching_delete_file_loader.rs b/crates/iceberg/src/arrow/caching_delete_file_loader.rs index e1cb6de158..ada86fc1df 100644 --- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs +++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs @@ -1121,6 +1121,321 @@ 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, + })) + } + } + + #[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() + .cloned() + .map(|mut delete| { + delete.file_size_in_bytes = 0; + delete + }) + .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(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() + }; + 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(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(0)); + 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: String| { + let mut delete = task.deletes()[0].clone(); + delete.file_path = path; + delete.file_size_in_bytes = 0; + delete + }; + + // 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.clone()); + 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, + 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, 0, 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 5852cf1acd..cb6257dbfb 100644 --- a/crates/iceberg/src/arrow/delete_file_loader.rs +++ b/crates/iceberg/src/arrow/delete_file_loader.rs @@ -73,6 +73,37 @@ impl BasicDeleteFileLoader { Essentially a super-cut-down ArrowReader. We can't use ArrowReader directly as that introduces a circular dependency. */ + // A size of 0 means the task did not carry one (see + // `FileScanTaskDeleteFile`). Resolve it when the file is first loaded. + let file_size_in_bytes = if file_size_in_bytes == 0 { + 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 + } else { + file_size_in_bytes + }; + // 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,31 +261,34 @@ 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 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(); - - let batches: Vec<_> = result.try_collect().await.unwrap(); - assert_eq!(batches.len(), 1); - assert_eq!(batches[0].num_rows(), 4); + // A known size, and an unknown size (0) resolved before decryption. + for file_size_in_bytes in [std::fs::metadata(&del_path).unwrap().len(), 0] { + let task = FileScanTaskDeleteFile { + file_path: del_path.clone(), + file_size_in_bytes, + 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 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] @@ -335,4 +369,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(), 0, 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 [0, 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(), 0, 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 97ca7e5181..3f849ac22e 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -340,7 +340,12 @@ pub struct FileScanTaskDeleteFile { /// The delete file path pub file_path: String, - /// The total size of the delete file in bytes, from the manifest entry. + /// The total size of the delete file in bytes. + /// + /// Tasks planned from manifest entries carry the recorded file size. + /// Externally constructed tasks may use `0` 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. pub file_size_in_bytes: u64, /// delete file type