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.
A woken waiter checks the state again and loads the file itself if it still isn't loaded, instead of assuming it is.

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_abandoned_pos_del_load_wakes_waiters_and_can_be_retried: dropping the guard wakes a waiter, and the next call gets Load.
Checked by hand: two concurrent load_deletes of a missing file on one loader hang on main; here both return the error.

cargo test -p iceberg --lib (1954 passed), cargo clippy --workspace --all-targets -- -D warnings and cargo fmt --check are clean.

AI Disclosure

I used Claude Code to help write the fix and tests. I reproduced the hang on main, reviewed the code, and ran the tests and lints above.

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.

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 _.

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.

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

+        let removed = self
+            .state
+            .write()
+            .unwrap()


Could this recover from a poisoned lock instead of unwrapping? get_delete_vector_for_path already tolerates poisoning with .read().ok(). This now runs from Drop. Say the lock was poisoned by a panic inside upsert_delete_vector in the DelVecs arm. The guard then drops during that unwind, and the second panic aborts the process. On main that chain ends as a task panic. It takes two panics in a row, so this is minor.

Suggested change

.unwrap()

.unwrap_or_else(std::sync::PoisonError::into_inner)

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

Open

	.unwrap()
	.unwrap_or_else(std::sync::PoisonError::into_inner)

Conversation

JeonDaehong commented Oct 13, 2026

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!

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!

This branch has not been deployed

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

mbutrovich left a comment •

edited

Loading

JeonDaehong Oct 13, 2026 •

edited

Loading