feat(reader): resolve unknown delete-file sizes lazily

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<std::sync::atomic::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 crate::io::Storage for MetadataCountingStorage {
+        async fn exists(&self, path: &str) -> Result<bool> {
+            crate::io::LocalFsStorage::new().exists(path).await
+        }
+
+        async fn metadata(&self, path: &str) -> Result<crate::io::FileMetadata> {
+            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<Bytes> {
+            crate::io::LocalFsStorage::new().read(path).await
+        }
+
+        async fn reader(&self, path: &str) -> Result<Box<dyn crate::io::FileRead>> {
+            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<Box<dyn crate::io::FileWrite>> {
+            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<crate::io::InputFile> {
+            Ok(crate::io::InputFile::new(
+                Arc::new(self.clone()),
+                path.to_string(),
+            ))
+        }
+
+        fn new_output(&self, path: &str) -> Result<crate::io::OutputFile> {
+            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<std::sync::atomic::AtomicUsize>,
+        #[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<Arc<dyn crate::io::Storage>> {
+            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<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);
+    }
+
+    #[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();