feat(reader): resolve unknown delete-file sizes lazily by JeonDaehong · Pull Request #3433 · apache/iceberg-rust

JeonDaehong · 2026-10-13T13:05:36Z

Which issue does this PR close?

Closes Support unknown delete-file sizes in externally constructed scan tasks #3344.

Continues #3345 by @unikdahal, who is away for a few weeks and offered it up in #3343. The original commit is credited with Co-authored-by.

What changes are included in this PR?

Same feature as #3345, rebased on #3145, with one change from the review there: an unknown size is None instead of a 0 sentinel.

FileScanTaskDeleteFile::file_size_in_bytes() now returns Option<u64>. Tasks planned from manifests always carry Some.
Builder: .with_file_size_in_bytes(n) is unchanged; leave it unset (or use .with_file_size_in_bytes_opt) when the size is unknown.
When the size is None, the Parquet delete-file loader resolves it with one metadata request on first load.
Deletion vectors don't use this size and are unchanged.

Are these changes tested?

Yes. The tests from #3345 are kept, adapted to the builder (private fields after #3145), plus a serde round trip for an unknown size.

cargo test -p iceberg --lib (1957 passed), cargo clippy --workspace --all-targets -- -D warnings and cargo fmt --check are clean. public-api.txt is updated for the getter.

AI Disclosure

I used Claude Code to help with the rebase and the Option<u64> change. I reviewed the code and ran the tests and lints above.

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 <unikdahal03@gmail.com>

JeonDaehong · 2026-10-13T13:07:38Z

@blackmwk @mbutrovich could you take a look when you have time? This picks up #3345 on top of #3145, which you both reviewed.

mbutrovich

Thanks @JeonDaehong for picking this up from #3345, and thanks @unikdahal for the original work. Comet stats every Parquet delete file up front today (fill_delete_file_sizes) because of apache/iceberg#12554, so I'd like to move it to this once it lands.

mbutrovich · 2026-10-13T13:52:10Z

+    /// 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,
+            }))
+        }
+    }


Could the test storage import LocalFsStorage, AtomicUsize, Ordering and the other crate::io types at the top of the test module, and hold one LocalFsStorage in a field? Each method currently builds a new crate::io::LocalFsStorage::new() behind a fully qualified path, which makes the 90 lines harder to scan for the two methods that differ (metadata and the constructors).

Done in b2981e2: the crate::io types are imported at the top and the storage holds a LocalFsStorage.

mbutrovich · 2026-10-13T13:52:17Z

+                            error.kind(),
+                            format!("Failed to stat delete file '{data_file_path}'"),
+                        )
+                        .with_retryable(error.retryable())


What should a caller do with a retryable error here? On the head commit, I loaded a position delete file of unknown size whose path didn't exist yet, so load_deletes failed with Failed to stat delete file. After creating the file, a second load_deletes for the same delete file on the same CachingDeleteFileLoader never completed (I stopped it after 3 seconds). A new loader read it fine.

As I read it, this predates the PR, and #3345's description listed it under "Not in this PR". try_start_pos_del_load marks the file Loading, and only the success path calls finish_pos_del_load, so a failed load leaves it Loading and later callers wait on a notification that never comes. I see the same hang with a known size when the open fails. Since ArrowReader clones share this state, retrying read on a clone of the same reader would hang too. #3274 tracks the equality-delete half, and #3275 fixes only that half. This PR adds a new retryable failure on the positional path, so could we file an issue for the positional half and link it here?

Reproduced on main at b154603 with a known size as well: the first load fails, a second load on the same loader after the file appears doesn't complete within 3s, and a new loader reads it fine. Filed #3434 for the positional half. I'm happy to take the fix there.

mbutrovich · 2026-10-13T13:52:42Z

+        // 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"
+                ),
+            ));
+        }


A size below the Parquet footer minimum, including Some(0), is rejected instead of dropping the file's deletes.

Which path reads a sub-footer size as a file with no deletes? Parquet already rejects a size below its footer in ParquetMetaDataReader. With this check disabled on the head commit, None on an empty file, Some(0) and Some(4) each fail with Failed to load Parquet metadata, source: EOF: file size of N is less than footer. A 10-byte file passes this check and then fails in Parquet with Corrupt footer, so the check doesn't define the minimum either.

What do you think about removing it and relying on Parquet's error? The footer minimum assertions in unknown_delete_size_still_reports_missing_and_empty_objects would then check for Parquet's error instead. If the DataInvalid kind matters to callers, could we compare against parquet::file::FOOTER_SIZE rather than the literal 8?

Agreed, removed in b2981e2. The test now checks Parquet's file size of N is less than footer error for None on an empty file, Some(0) and Some(4).

mbutrovich · 2026-10-13T13:52:53Z

+    /// 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; the Parquet delete-file loader then resolves it with one metadata request
+    /// when the file is first loaded. Deletion vectors do not use it.


Should this doc also cover a recorded size that can't be trusted? A task planned on the JVM always has fileSizeInBytes(), since file_size_in_bytes is required in every format version. The case I know of is apache/iceberg#12554: rewrite_table_path before Iceberg Java 1.12.0 left the old size in the manifest after rewriting position delete files, and an engine that wants to read those tables has to ignore the recorded size. Saying so here would tell engine authors when to leave it unset.

Added in b2981e2, with a link to apache/iceberg#12554.

mbutrovich · 2026-10-13T13:53:02Z

+        #[serde(default)]
+        file_size_in_bytes: Option<u64>,


Could we add a serde test for the unknown size? A serialized task without file_size_in_bytes used to fail to deserialize and now produces None, and skip_serializing_if drops the field on the way out. A round trip of a task with None, next to test_deserialized_deletion_vector_rejects_missing_fields, would cover both directions.

Added test_delete_file_without_size_round_trips_through_serde next to that test in b2981e2. It checks that the field is left out when serializing and that a task without it reads back as None.

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.

JeonDaehong · 2026-10-13T14:32:31Z

@mbutrovich
Thank you for the review! I've made the requested changes and left a reply. :D

mbutrovich

Thanks @JeonDaehong for the quick turnaround, and for filing #3434. Another pass, mostly concerned with trimming tests.

mbutrovich · 2026-10-13T15:10:49Z

+    #[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(), None, 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 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}");
+        }
+
+        // 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(), None, None)
+            .await
+            .unwrap()
+            .try_collect()
+            .await
+            .unwrap();
+        assert_eq!(batches.iter().map(|b| b.num_rows()).sum::<usize>(), 3);
+    }


Could this test keep only the footer cases? As I read it, the other two parts are covered elsewhere in this PR. The missing-file case is in test_load_deletes_reports_unknown_size_failures, which also checks that a retryable storage error stays retryable. A local missing file can't show that, since both sides of the kind() and retryable() comparison come from the same LocalFsStorage error. The read of a present file with an unknown size is the None iteration in test_read_encrypted_positional_delete_file, which goes through the same parquet_to_batch_stream call. Dropping the copy also removes the inline arrow_schema::, arrow_array:: and parquet::arrow:: paths. The suggestion also adds the test_ prefix the other tests in this module use.

Suggested change

#[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(), None, 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 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}");

}

// 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(), None, None)

.await

.unwrap()

.try_collect()

.await

.unwrap();

assert_eq!(batches.iter().map(|b| b.num_rows()).sum::<usize>(), 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}");

}

}

Done in ebc0671, applied as suggested.

mbutrovich · 2026-10-13T15:11:35Z

+    #[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,
+                Some(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, None, 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);
+    }


Is this test needed alongside test_unknown_size_equality_deletes_are_resolved_once_and_cached? That test already compares the predicate from an equality delete file of unknown size with the one from the known size, through load_deletes. This one checks the same thing one layer down, with the same file. The wider eq_ids set here doesn't matter for this change, because the size only affects how the file is opened.

Agreed, removed in ebc0671.

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.

JeonDaehong · 2026-10-13T23:39:03Z

@mbutrovich addressed all comments, ready for another look.

JeonDaehong · 2026-10-21T12:33:27Z

@mbutrovich @laskoviymishka friendly ping. Your comments from last week are addressed (b2981e2, ebc0671), CI is green and it merges cleanly. Could you take another look when you have a moment?

laskoviymishka

this looks reasonable to me. My main ask is marking the file_size_in_bytes() change as breaking, see the inline on the getter.

Could we also open an issue for a scan option that ignores recorded delete-file sizes? plan_files always passes the manifest size, so on a table hit by apache/iceberg#12554 only tasks an engine builds itself get the lazy stat.

I'd like @mbutrovich or @blackmwk to vet the delete-loader path and the API change before this goes in.

laskoviymishka · 2026-10-23T18:21:38Z

+            .unwrap();
+        assert_eq!(calls.load(Ordering::SeqCst), unknown.len());
+        loader
+            .load_deletes(&unknown[1..], task.schema_ref())


Could this also run two load_deletes on the same unknown-size files concurrently, say with futures::join!, and check calls stays at unknown.len()? Here the second load starts after the first finishes, so it takes the AlreadyLoaded path. The only WaitFor test is state-level and doesn't go through load_deletes. A succeeding stat is enough, since a failing one hangs today (#3434).

laskoviymishka · 2026-10-23T18:21:38Z

-    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> {


Could we mark this as breaking, with feat(reader)!: in the title or a breaking-change note in the description, as #3263 did? On main this getter returns u64, so downstream callers that use it as one stop compiling. A task serialized with None also can't be read by an older build, since the base serde proxy requires the field.

JeonDaehong mentioned this pull request Oct 13, 2026

[EPIC] Execution-time adaptive pruning for Iceberg readers #3343

Open

21 tasks

mbutrovich mentioned this pull request Oct 13, 2026

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

Closed

mbutrovich reviewed Oct 13, 2026

View reviewed changes

JeonDaehong mentioned this pull request Oct 13, 2026

Failed positional-delete load leaves the file stuck in Loading, hanging later loads #3434

Open

JeonDaehong mentioned this pull request Oct 13, 2026

fix(arrow): release a positional delete file whose load fails #3435

Open

mbutrovich self-requested a review October 13, 2026 14:41

mbutrovich reviewed Oct 13, 2026

View reviewed changes

Merge branch 'main' into lazy-delete-file-sizes

aa189cc

JeonDaehong mentioned this pull request Oct 16, 2026

ci: Limit open, non-draft PRs to 3 for users without write access #3465

Merged

nssalian mentioned this pull request Oct 22, 2026

EPIC: v3 Support Tracking #2411

Open

22 of 64 tasks

laskoviymishka approved these changes Oct 23, 2026

View reviewed changes

Conversation

JeonDaehong commented Oct 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which issue does this PR close?

What changes are included in this PR?

Are these changes tested?

AI Disclosure

Uh oh!

JeonDaehong commented Oct 13, 2026

Uh oh!

mbutrovich left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

JeonDaehong Oct 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

JeonDaehong commented Oct 13, 2026

Uh oh!

mbutrovich left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

JeonDaehong commented Oct 13, 2026

Uh oh!

JeonDaehong commented Oct 21, 2026

Uh oh!

laskoviymishka left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

This branch has not been deployed

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

JeonDaehong commented Oct 13, 2026 •

edited

Loading

JeonDaehong Oct 13, 2026 •

edited

Loading