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