fix(arrow): release a positional delete file whose load fails by JeonDaehong · Pull Request #3435 · apache/iceberg-rust

JeonDaehong · 2026-10-13T14:40:25Z

Which issue does this PR close?

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

What changes are included in this PR?

A positional delete file whose load fails no longer stays Loading forever.

PosDelLoadAction::Load now carries a PosDelLoadGuard. finish() marks the file loaded. Dropping the guard without finishing (open or parse error, or the load dropped because another delete file failed) removes the Loading marker and wakes the waiters.
Tasks waiting on a load that fails get its error back, so a missing file shared by many tasks is opened once. A load dropped without an error (another delete file failed and stopped load_deletes) wakes its waiters to try again. Either way the entry is removed, so a later call starts a new attempt.

This is the positional counterpart of #3275 and doesn't depend on it or on #3433. It merges cleanly with #3433 (1960 lib tests pass on the merge).

Are these changes tested?

Yes.

test_pos_del_load_can_be_retried_after_failure is the Failed positional-delete load leaves the file stuck in Loading, hanging later loads #3434 repro. On main it times out after 5s; here the retry loads the file and matches a fresh loader.
test_cancelled_pos_del_load_wakes_waiters_and_can_be_retried: dropping the guard wakes a waiter, and the next call gets Load.
test_pos_del_waiters_return_the_error_of_a_failed_load, test_pos_del_waiters_retry_after_cancelled_load, test_pos_del_load_dropped_by_a_failing_sibling_can_be_loaded_later and test_failed_pos_del_load_returns_its_error_to_waiters cover several waiters on one failed or cancelled load.

cargo test -p iceberg --lib --all-features (2091 passed), make check-fmt, check-clippy and check-doc are clean.

AI Disclosure

Claude Code wrote the fix and tests and ran them on Linux. I reproduced the hang on main and reviewed the code.

try_start_pos_del_load marks a positional delete file Loading, and only the success path marked it Loaded. If opening or parsing the file failed, or the load was dropped because another delete file failed, the entry stayed Loading: a retry on the same loader, a clone of the same ArrowReader, or a concurrent load of the same file waited for a notification that never came. The Load action now carries a PosDelLoadGuard. finish() marks the file loaded; dropping the guard without finishing removes the Loading marker and wakes the waiters. A woken waiter checks the state again and loads the file itself if it isn't loaded, instead of assuming it is. Closes apache#3434.

JeonDaehong · 2026-10-13T14:41:18Z

@mbutrovich this is the fix for the hang you found on #3433. @laskoviymishka could you take a look when you have time?

mbutrovich

Thanks @JeonDaehong for turning the hang from #3433 into #3434 and a fix so quickly.

mbutrovich · 2026-10-13T15:06:14Z

+                        PosDelLoadAction::WaitFor(notified) => {
+                            // Positional deletes are accessed synchronously by ArrowReader.
+                            // We must wait here to ensure the data is ready before returning,
+                            // otherwise ArrowReader might get an empty/partial result.
+                            // The wait also ends if that load fails, so check again.
+                            notified.await;
+                        }


Should a woken waiter retry the load, or return the failed load's error? On #2859, @laskoviymishka suggested an RAII guard with waiters that "re-check state and surface the error" (thread). This PR adds the guard and the re-check, but a waiter that finds the file released loads it itself.

As I read it, if K tasks wait on a delete file that keeps failing (corrupt, or a permission error), each one takes a turn at loading it, one after another. Each turn opens the file again, and for a parse error it also reads and parses the file again. load_deletes runs on a spawned task, so those attempts continue after ArrowReader has returned the first error. Retrying does help with the transient failure that #3433 adds. Surfacing the error would need the error stored in the state, because Notify carries no payload and Error isn't Clone. Which behavior do we want here?

Good point. I'd keep the retry. Given that #3433 involves potential transient failures (e.g., size stats), surfacing the first error would cause every waiter to fail on something a second attempt could fix.

You're right about the overhead for a file that keeps failing, as K waiters would retry sequentially. However, storing the error in the state would require a larger change since Error isn't Clone. I'd prefer to handle this as a follow-up. Happy to open an issue if you agree.

I don't feel strongly either way, but I'd prefer surfacing the error, having more insights in problem always helpful, as I suggested on #2859.

But If there's a good reason to keep the retry, I can be convinced.

For example, say one delete file is shared by 200 scan tasks and that file is missing or corrupt. With retry, we can end up reopening the same object ~200 times serially before the last task finally fails. If we surface the error, it’s one open and all waiting tasks fail fast.

Maybe a middle ground is to return the error only to the tasks waiting on that attempt, but still clear the entry so later calls can retry, similar to Go’s singleflight.

went with the singleflight version. a failed load now keeps its error and the tasks waiting on that attempt return it, so in your 200-task example the file is opened once. the entry is removed either way, so a later call starts a new attempt. a load dropped without an error (a sibling's error ending load_deletes) still lets its waiters retry, since nothing failed for that file.

mbutrovich · 2026-10-13T15:06:26Z

    }
+
+    #[tokio::test]
+    async fn test_pos_del_load_can_be_retried_after_failure() {


Could we add tests that go through the new loop in load_file_for_task? This test takes the Load branch twice in sequence, and test_abandoned_pos_del_load_wakes_waiters_and_can_be_retried stops at try_start_pos_del_load. As I read it, no test takes the WaitFor branch and then comes back around to Load, which is the new behavior here.

On #2859, load_file_for_task was left untested because a deterministic test seemed to need a test-only hook (thread). The guard removes that need. A test can take the Load guard from loader.delete_filter, poll load_file_for_task for the same file to Pending with futures::poll!, drop the guard, and then check that the waiter returns DeleteFileContext::PosDels. The test in #2630 uses the same polling pattern. A variant with the file missing should return the open error.

A second case is a positional delete file that opens but fails to parse (a row with a negative pos), then is rewritten with valid rows and loaded again on the same loader. Here the guard is dropped in parse_file_content_for_task instead of in load_file_for_task, so the open-error test doesn't cover it. On main the second load hangs (I stopped it after 5 seconds).

Added in 6757eb2 :
test_pos_del_waiter_loads_file_after_abandoned_load, test_pos_del_waiter_returns_open_error_after_abandoned_load (both poll the waiter to Pending while holding the guard), and test_pos_del_load_can_be_retried_after_parse_failure. All three time out with the guard's Drop disabled.

mbutrovich · 2026-10-13T15:06:44Z

+    fn abandon_pos_del_load(&self, file_path: &str) {
+        let notify = {
+            let mut state = self.state.write().unwrap();
+            match state.positional_deletes.remove(file_path) {
+                Some(PosDelState::Loading(notify)) => Some(notify),
+                // Not ours to remove: keep whatever is there.
+                Some(other) => {
+                    state
+                        .positional_deletes
+                        .insert(file_path.to_string(), other);
+                    None
+                }
+                None => None,
+            }
+        };
+
+        if let Some(notify) = notify {
+            notify.notify_waiters();
+        }
+    }


When can this entry be anything other than Loading here? As I read it, only try_start_pos_del_load inserts into positional_deletes, and only when the path is absent. After that, only this guard's finish or drop changes the entry. So while the guard is alive the entry is the Loading it created, and the Some(other) and None arms can't run. The "Not ours to remove" comment suggests that another task can own the entry, which could mislead the next reader of this code. What do you think about asserting the invariant instead?

Suggested change

fn abandon_pos_del_load(&self, file_path: &str) {

let notify = {

let mut state = self.state.write().unwrap();

match state.positional_deletes.remove(file_path) {

Some(PosDelState::Loading(notify)) => Some(notify),

// Not ours to remove: keep whatever is there.

Some(other) => {

state

.positional_deletes

.insert(file_path.to_string(), other);

None

}

None => None,

}

};

if let Some(notify) = notify {

notify.notify_waiters();

}

}

fn abandon_pos_del_load(&self, file_path: &str) {

let removed = self

.state

.write()

.unwrap()

.positional_deletes

.remove(file_path);

debug_assert!(matches!(removed, Some(PosDelState::Loading(_))));

if let Some(PosDelState::Loading(notify)) = removed {

notify.notify_waiters();

}

}

You're right, only the guard touches the entry once it's inserted. Applied in 6532513.

Only try_start_pos_del_load inserts into positional_deletes, and only the guard that it returns changes the entry afterwards, so the entry is always the guard's Loading marker when it is abandoned.

Hold the Load guard, poll load_file_for_task for the same file until it waits, then drop the guard: the waiter loads the file itself, or returns the open error when the file is gone. Also load a positional delete file that opens but fails to parse, then rewrite it and load it again on the same loader, which hangs without the guard.

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

@mbutrovich addressed all comments, ready for another look.

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

@mbutrovich @laskoviymishka friendly ping on this one too: the fix for the positional-delete retry hang you found on #3433 (#3434). The three review items are in 6532513 and 6757eb2, CI is green and it merges cleanly.

laskoviymishka

Thanks @JeonDaehong, the poll-based waiter tests make the new loop easy to follow. Overall this looks good to me, and what I have are questions rather than blockers. The main one is retry vs surface. I've replied on @mbutrovich's thread, and since that's his open question, I'd like him to sign off on that part as well.

laskoviymishka · 2026-10-23T18:18:48Z

+    ) -> (
+        PosDelLoadGuard,
+        std::pin::Pin<Box<impl Future<Output = Result<DeleteFileContext>>>>,
+        String,


nit: could wait_on_pos_del_load return a small struct instead of the 3-tuple? Only the open-error test uses the path. With named fields, the other caller could skip it with .. instead of _.

done, it returns a PendingPosDelLoad now and takes the number of waiters.

laskoviymishka · 2026-10-23T18:18:48Z

+    }
+
+    #[tokio::test]
+    async fn test_pos_del_waiter_loads_file_after_abandoned_load() {


Could the description's case of two concurrent load_deletes calls go in as a test? These tests have one waiter per abandoned load. Nothing covers several waiters waking on one abandon, or a sibling's guard dropped by the item? early return in load_deletes. One way: add a second polled waiter to wait_on_pos_del_load, remove the file before the drop, and both waiters should return the open error.

added in d51812c: two waiters on a load that fails while the file is still there both return the load's error (so neither reopened it), two waiters on a cancelled load each return their own open error, and a load dropped by a failing sibling's item? loads later on the same loader.

correction on the sibling test: on the current-thread test runtime it didn't actually overlap the two loads, the valid file only loaded on the second call, same thing @mbutrovich found on #3433. 5fa6051 adds a test storage that yields before reads and holds gated.parquet until the test opens it. with a probe the first call now goes Load invalid, Load gated, fail invalid, abandon gated, and the second call loads gated.

try_start_pos_del_load marks a positional delete file Loading, and only the success path marked it Loaded. If opening or parsing the file failed, or the load was dropped because another delete file failed, the entry stayed Loading: a retry on the same loader, a clone of the same ArrowReader, or a concurrent load of the same file waited for a notification that never came. The Load action now carries a PosDelLoadGuard. finish() marks the file loaded; dropping the guard without finishing removes the Loading marker and wakes the waiters. A woken waiter checks the state again and loads the file itself if it isn't loaded, instead of assuming it is. Closes apache#3434. [Upstream PR apache#3435, open on 2026-10-25.] (cherry picked from commit deff41b)

Only try_start_pos_del_load inserts into positional_deletes, and only the guard that it returns changes the entry afterwards, so the entry is always the guard's Loading marker when it is abandoned. [Upstream PR apache#3435, open on 2026-10-25.] (cherry picked from commit 6532513)

Hold the Load guard, poll load_file_for_task for the same file until it waits, then drop the guard: the waiter loads the file itself, or returns the open error when the file is gone. Also load a positional delete file that opens but fails to parse, then rewrite it and load it again on the same loader, which hangs without the guard. [Upstream PR apache#3435, open on 2026-10-25.] (cherry picked from commit 6757eb2)

…iled Tasks that share a delete filter load each equality delete file once; the others wait for that load. When the load failed, the spawned task that stores the set panicked on the dropped sender and left the file marked as loading. Every task that waited for, or later needed, that file then waited forever, and a reader that shares the filter across tasks hung after the first error instead of failing. A failed equality delete load now removes the file's state and wakes the waiters, which return the existing "Missing equality delete set" error. A later task loads the file again. The spawned task and the drop path of a positional delete load also take a poisoned state lock: a panic there would leave the waiters waiting forever, and in the drop path a panic during unwinding would abort the process. Upstream issue apache#3274 reports this hang. PR apache#3275 fixed it the same way and was closed without merging on 2026-10-20. The positional counterpart is issue apache#3434 and PR apache#3435, cherry-picked before this commit. One more test here checks that a task waiting for a positional delete load that a failing sibling cancelled loads the file itself, with all its deletes. Review: `insert_equality_delete` and the end of `get_equality_delete_set_for_delete_file_path` in delete_filter.rs. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

try_start_pos_del_load marks a positional delete file Loading, and only the success path marked it Loaded. If opening or parsing the file failed, or the load was dropped because another delete file failed, the entry stayed Loading: a retry on the same loader, a clone of the same ArrowReader, or a concurrent load of the same file waited for a notification that never came. The Load action now carries a PosDelLoadGuard. finish() marks the file loaded; dropping the guard without finishing removes the Loading marker and wakes the waiters. A woken waiter checks the state again and loads the file itself if it isn't loaded, instead of assuming it is. Closes apache#3434. [Upstream PR apache#3435, open on 2026-10-25.] (cherry picked from commit deff41b)

Only try_start_pos_del_load inserts into positional_deletes, and only the guard that it returns changes the entry afterwards, so the entry is always the guard's Loading marker when it is abandoned. [Upstream PR apache#3435, open on 2026-10-25.] (cherry picked from commit 6532513)

Hold the Load guard, poll load_file_for_task for the same file until it waits, then drop the guard: the waiter loads the file itself, or returns the open error when the file is gone. Also load a positional delete file that opens but fails to parse, then rewrite it and load it again on the same loader, which hangs without the guard. [Upstream PR apache#3435, open on 2026-10-25.] (cherry picked from commit 6757eb2)

…iled Tasks that share a delete filter load each equality delete file once; the others wait for that load. When the load failed, the spawned task that stores the set panicked on the dropped sender and left the file marked as loading. Every task that waited for, or later needed, that file then waited forever, and a reader that shares the filter across tasks hung after the first error instead of failing. A failed equality delete load now removes the file's state and wakes the waiters, which return the existing "Missing equality delete set" error. A later task loads the file again. The spawned task and the drop path of a positional delete load also take a poisoned state lock: a panic there would leave the waiters waiting forever, and in the drop path a panic during unwinding would abort the process. Upstream issue apache#3274 reports this hang. PR apache#3275 fixed it the same way and was closed without merging on 2026-10-20. The positional counterpart is issue apache#3434 and PR apache#3435, cherry-picked before this commit. One more test here checks that a task waiting for a positional delete load that a failing sibling cancelled loads the file itself, with all its deletes. Review: `insert_equality_delete` and the end of `get_equality_delete_set_for_delete_file_path` in delete_filter.rs. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

…ters Waiters on a failed load used to retry it one after another, so a missing file shared by many tasks was opened once per task. The load now records its error and the tasks waiting on that attempt return it. A load dropped without an error, when a sibling's error stops load_deletes, still lets its waiters retry, and later calls start a new attempt either way. Abandoning a load also tolerates a poisoned lock, since it runs from Drop.

JeonDaehong · 2026-10-29T11:27:29Z

@laskoviymishka addressed all four in d51812c. @mbutrovich this changes the retry behavior from your thread, could you take another look?

…le fails On the current-thread test runtime the local filesystem never yields, so the valid file was only loaded on the second call and its guard was never dropped mid-load. A test storage now yields before reads and holds one file until the test opens a gate.

mbutrovich

Another round, thanks for sticking with this @JeonDaehong!

mbutrovich · 2026-11-02T15:27:06Z

+    pub(crate) fn fail(self, error: &Error) {
+        let _ = self
+            .attempt
+            .failure
+            .set((error.kind(), error.message().to_string()));
+    }


Should the waiters get the whole error, including its context and source? fail keeps only kind and message, so a waiter's error drops both. With the OpenDAL storage, the cause of an I/O failure is in the source: from_opendal_error sets the message to "Failure in doing io operation" and puts the OpenDAL error in with_source. I failed a guard with an error built that way and printed what the leader and a waiter return:

leader: Unexpected, context: { path: s3://bucket/pos-delete.parquet } => Failure in doing io operation, source: 403 Forbidden waiter: Unexpected => Failure in doing io operation

In the 200-task example from the earlier thread, 199 tasks return the second line. With more than one data file in flight, the scan buffers tasks unordered, so as I read it the error a user sees can be a waiter's, with no path and no cause.

Keeping the Display output as the message carries both. The waiter's error then starts with the kind twice (DataInvalid => DataInvalid => bad file), and the two tests that compare err.message() exactly would check with contains instead. Could we also add a source to the error in test_failed_pos_del_load_returns_its_error_to_waiters and check that the waiter's error mentions it?

Suggested change

pub(crate) fn fail(self, error: &Error) {

let _ = self

.attempt

.failure

.set((error.kind(), error.message().to_string()));

}

pub(crate) fn fail(self, error: &Error) {

let _ = self

.attempt

.failure

.set((error.kind(), error.to_string()));

}

mbutrovich · 2026-11-02T15:27:33Z

+        let write = |name: &str, positions: Vec<i64>| {
+            let path = tmp_dir.path().join(name);
+            let path = path.to_str().unwrap().to_string();
+            let batch = RecordBatch::try_new(
+                crate::arrow::delete_filter::tests::create_pos_del_schema(),
+                vec![
+                    Arc::new(StringArray::from_iter_values(vec![
+                        "data.parquet";
+                        positions.len()
+                    ])),
+                    Arc::new(Int64Array::from_iter_values(positions)),
+                ],
+            )
+            .unwrap();
+            let mut writer =
+                ArrowWriter::try_new(File::create(&path).unwrap(), batch.schema(), None).unwrap();
+            writer.write(&batch).unwrap();
+            writer.close().unwrap();
+            FileScanTaskDeleteFile::builder()
+                .with_file_path(path.clone())
+                .with_file_size_in_bytes(std::fs::metadata(&path).unwrap().len())
+                .with_file_type(DataContentType::PositionDeletes)
+                .with_file_format(DataFileFormat::Parquet)
+                .with_partition_spec_id(0)
+                .build()
+                .unwrap()
+        };


Could this closure and the one in test_pos_del_load_can_be_retried_after_parse_failure (lines 2145-2169) become one test helper, such as fn write_pos_del_file(path: &str, positions: Vec<i64>) -> FileScanTaskDeleteFile? They write the same batch and build the same FileScanTaskDeleteFile, and differ only in how the path is formed. One helper keeps the two tests' fixtures from drifting apart.

mbutrovich reviewed Oct 13, 2026

View reviewed changes

JeonDaehong added 2 commits October 14, 2026 08:25

refactor(arrow): assert the Loading invariant when abandoning a load

6532513

Only try_start_pos_del_load inserts into positional_deletes, and only the guard that it returns changes the entry afterwards, so the entry is always the guard's Loading marker when it is abandoned.

Merge branch 'main' into fix-pos-del-load-retry

faa71d7

JeonDaehong mentioned this pull request Oct 16, 2026

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

Merged

laskoviymishka approved these changes Oct 23, 2026

View reviewed changes

c-thiel mentioned this pull request Oct 25, 2026

fix: partition names Avro does not allow, truncated statistics, main retention, failed delete loads, and upstream cherry-picks lakekeeper/iceberg-rust#4

Merged

JanKaul mentioned this pull request Oct 26, 2026

fix: partition names Avro does not allow, truncated statistics, main retention, failed delete loads, and upstream cherry-picks lakekeeper/iceberg-rust#6

Merged

JeonDaehong added 2 commits October 29, 2026 19:43

Merge branch 'main' into fix-pos-del-load-retry

339fff4

mbutrovich reviewed Nov 2, 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 • 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!

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!

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!

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!

Uh oh!

JeonDaehong commented Oct 29, 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!

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

mbutrovich left a comment •

edited

Loading

JeonDaehong Oct 13, 2026 •

edited

Loading