diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 00040053a6..5926489bd6 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1395,7 +1395,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..e8b4cef4b5 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 @@ -873,7 +898,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 +1154,285 @@ 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, + /// 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 { + self.calls.fetch_add(1, Ordering::SeqCst); + 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 { + Ok(InputFile::new(Arc::new(self.clone()), path.to_string())) + } + + fn new_output(&self, path: &str) -> Result { + Ok(OutputFile::new(Arc::new(self.clone()), path.to_string())) + } + + async fn exists(&self, path: &str) -> Result { + self.inner.exists(path).await + } + + async fn read(&self, path: &str) -> Result { + self.inner.read(path).await + } + + async fn reader(&self, path: &str) -> Result> { + 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> { + 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, + #[serde(skip)] + fail_metadata: bool, + } + + #[typetag::serde(name = "test_metadata_counting_storage_factory")] + 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, + })) + } + } + + /// 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() { + 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 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 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(); @@ -1312,7 +1616,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 +1794,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..756ce26f73 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, 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 + } + }; 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 45940b0dfd..60dabc6324 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 leave it unset 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 + /// (). 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 +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, file_type: DataContentType, file_format: DataFileFormat, partition_spec_id: i32, @@ -635,7 +645,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 +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 { 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);