This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-3433-5370b61638e205d7caf97889affe5da63aea9b16 in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit 1c4018e7afc0d4eb2f5ccfc1ccbd5c4d4649e1d4 Author: JeonDaehong <[email protected]> AuthorDate: Sat Nov 7 09:28:10 2026 +0000 feat(reader)!: resolve unknown delete-file sizes lazily (#3433) * 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<u64>. 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 <[email protected]> * 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. * 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. * test(reader): resolve unknown delete sizes once under concurrent loads * test(reader): yield in the counting storage so concurrent loads overlap * fix(scan): keep the delete file size required in the builder --------- Co-authored-by: unikdahal <[email protected]> --- crates/iceberg/public-api.txt | 2 +- .../src/arrow/caching_delete_file_loader.rs | 329 ++++++++++++++++++++- crates/iceberg/src/arrow/delete_file_loader.rs | 79 +++-- crates/iceberg/src/scan/task.rs | 25 +- 4 files changed, 406 insertions(+), 29 deletions(-) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 1c5af7cae..05fcde658 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1403,7 +1403,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<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 diff --git a/crates/iceberg/src/arrow/caching_delete_file_loader.rs b/crates/iceberg/src/arrow/caching_delete_file_loader.rs index 0711a58b3..62ce6350e 100644 --- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs +++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs @@ -840,12 +840,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; @@ -853,10 +855,34 @@ 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_size_in_bytes_opt(None) + .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 @@ -890,7 +916,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 @@ -1146,6 +1172,303 @@ 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)] + inner: LocalFsStorage, + #[serde(skip)] + calls: Arc<AtomicUsize>, + /// 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 Storage for MetadataCountingStorage { + async fn metadata(&self, path: &str) -> Result<FileMetadata> { + self.calls.fetch_add(1, Ordering::SeqCst); + // `LocalFsStorage::metadata` doesn't yield, so without this the first of two + // concurrent loads finishes before the second starts. + tokio::task::yield_now().await; + if self.fail_metadata { + return Err( + Error::new(ErrorKind::Unexpected, "metadata temporarily unavailable") + .with_retryable(true), + ); + } + self.inner.metadata(path).await + } + + fn new_input(&self, path: &str) -> Result<InputFile> { + Ok(InputFile::new(Arc::new(self.clone()), path.to_string())) + } + + fn new_output(&self, path: &str) -> Result<OutputFile> { + Ok(OutputFile::new(Arc::new(self.clone()), path.to_string())) + } + + async fn exists(&self, path: &str) -> Result<bool> { + self.inner.exists(path).await + } + + async fn read(&self, path: &str) -> Result<Bytes> { + self.inner.read(path).await + } + + async fn reader(&self, path: &str) -> Result<Box<dyn FileRead>> { + self.inner.reader(path).await + } + + async fn write(&self, path: &str, bs: Bytes) -> Result<()> { + self.inner.write(path, bs).await + } + + async fn writer(&self, path: &str) -> Result<Box<dyn FileWrite>> { + self.inner.writer(path).await + } + + async fn delete(&self, path: &str) -> Result<()> { + self.inner.delete(path).await + } + + 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<AtomicUsize>, + #[serde(skip)] + fail_metadata: bool, + } + + #[typetag::serde(name = "test_metadata_counting_storage_factory")] + impl StorageFactory for MetadataCountingStorageFactory { + fn build(&self, _config: &StorageConfig) -> Result<Arc<dyn Storage>> { + Ok(Arc::new(MetadataCountingStorage { + inner: LocalFsStorage, + 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_size_in_bytes_opt(None) + .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() { + 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 = FileIOBuilder::new(factory).build(); + let file_io_for_join = file_io.clone(); + 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<u64> { + 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); + + // Two concurrent loads of the same files on a fresh loader: the second waits on + // the first instead of requesting the sizes again. + let before = calls.load(Ordering::SeqCst); + let loader = CachingDeleteFileLoader::new(file_io_for_join, 10, Runtime::current()); + let (first, second) = futures::join!( + loader.load_deletes(&unknown, task.schema_ref()), + loader.load_deletes(&unknown, task.schema_ref()) + ); + let (first, second) = (first.unwrap().unwrap(), second.unwrap().unwrap()); + assert_eq!(calls.load(Ordering::SeqCst) - before, unknown.len()); + assert_eq!(positions(&first), expected); + assert_eq!(positions(&second), expected); + } + + #[tokio::test] + async fn test_unknown_size_equality_deletes_are_resolved_once_and_cached() { + 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 = 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 = 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_parse_positional_deletes_rejects_negative_positions() { let schema = crate::arrow::delete_filter::tests::create_pos_del_schema(); @@ -1329,7 +1652,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 @@ -1507,7 +1830,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 ec61710a4..756ce26f7 100644 --- a/crates/iceberg/src/arrow/delete_file_loader.rs +++ b/crates/iceberg/src/arrow/delete_file_loader.rs @@ -66,13 +66,34 @@ 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<u64>, key_metadata: Option<&[u8]>, ) -> Result<ArrowRecordBatchStream> { /* 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 + } + }; let parquet_read_options = ParquetReadOptions::builder().build(); let (parquet_file_reader, arrow_metadata) = ArrowReader::open_parquet_file( @@ -230,27 +251,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 +352,23 @@ mod tests { assert_eq!(batches.len(), 1); assert_eq!(batches[0].num_rows(), 3); } + + #[tokio::test] + 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()); + + // 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(); + 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!(error.to_string().contains("is less than footer"), "{error}"); + } + } } diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs index 45940b0df..29c6ed71a 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -366,8 +366,17 @@ 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 pass `None` when the size is + /// 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 + /// (<https://github.com/apache/iceberg/issues/12554>). 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(setter(strip_option(fallback = with_file_size_in_bytes_opt)))] + file_size_in_bytes: Option<u64>, /// delete file type file_type: DataContentType, @@ -612,7 +621,8 @@ mod _serde { #[derive(Deserialize)] pub(super) struct FileScanTaskDeleteFileSerde { file_path: String, - file_size_in_bytes: u64, + #[serde(default)] + file_size_in_bytes: Option<u64>, file_type: DataContentType, file_format: DataFileFormat, partition_spec_id: i32, @@ -635,7 +645,7 @@ mod _serde { 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_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 +666,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<u64> { self.file_size_in_bytes } @@ -969,7 +980,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);
