fix(arrow): resolve eq-delete waiters when a delete file fails to load - #3275

Which issue does this PR close?

• Closes Failed equality-delete load leaves the entry stuck in Loading, hanging later reads #3274.

This case was raised during review of #3268.

What changes are included in this PR?

An equality-delete file that fails to load (missing, unreadable, or unparseable) can leave an entry permanently stuck in
the Loading state inside DeleteFilter. Any task that later waits on that entry blocks
forever on a notification that will never fire. This PR makes a failed load clean up
after itself and wake its waiters, so the failure surfaces as an error instead of a
hang.

Background: how equality-delete loading works

Equality-delete predicates are produced by one task and consumed by others through a small
state machine keyed by delete-file path:

• EqDelState::Loading(Notify) — a task is parsing this file; waiters hold the Notify.
• EqDelState::Loaded(Predicate) — the predicate is ready.

The producer side (insert_equality_delete) marks the entry Loading, then spawns a task
that waits on a oneshot channel for the parsed predicate, stores it as Loaded, and calls
notify_waiters(). The parsed predicate is sent from the delete-file loader
(caching_delete_file_loader.rs, the FreshEqDel branch).

The consumer side (get_equality_delete_predicate_for_delete_file_path) reads the entry; if
it is Loading, it waits on the Notify until the producer signals completion.

The bug

The spawned producer task did:

let eq_del = eq_del.await.unwrap();   // <-- panics if the sender was dropped
// ... insert Loaded ...
notify.notify_waiters();

The predicate is sent here (caching_delete_file_loader.rs):

let predicate =
    Self::parse_equality_deletes_record_batch_stream(batch_stream, equality_ids)
        .await?;               // <-- early return on parse error DROPS `sender`
sender.send(predicate) ...

If parsing the delete file fails, the ? returns early before sender.send(...),
dropping the sender. On the producer task, eq_del.await then resolves to Err(RecvError),
.unwrap() panics, and the two lines that follow — the Loaded insert and
notify_waiters() — never run. The entry is stranded in Loading, and its Notify is
never signalled.

Exact sequence of events

sequenceDiagram
    participant P as Producer task (insert_equality_delete)
    participant L as Loader (FreshEqDel branch)
    participant S as DeleteFilter state
    participant W as Waiter (get_equality_delete_predicate...)

    L->>S: insert(path, Loading(notify))
    L->>P: spawn task, awaiting oneshot rx
    W->>S: read(path) -> Loading(notify)
    W->>W: notify.notified().await   (parked)
    Note over L: parse_equality_deletes_...().await?  ❌ parse error
    L--xP: sender dropped (early return, never sent)
    P->>P: eq_del.await -> Err(RecvError)
    Note over P: .unwrap() PANICS
    Note over S: entry stays Loading forever,<br/>notify_waiters() never called
    Note over W: parked forever — HANG 🔒

State of the map entry:

                 success path                     failure path (this bug)
   Loading  ──────────────────► Loaded      Loading ──► (parse error) ──► Loading  🔒
              send + notify                            sender dropped,     forever
                                                       task panics

When it actually hangs

Within a single load_deletes call the parse error also propagates up the results stream
via ?, which tears down that call and cancels its sibling tasks — so the stuck entry is
usually not observed there. The hang bites when a DeleteFilter is shared across
concurrent or sequential loads (an ArrowReader is Clone and its clones share one
loader/filter, and many data files typically reference the same equality-delete file):

1. Loader task A claims delete file E, marks it Loading, and fails to parse it.
2. A panics; E is never signalled and never leaves Loading.
3. A later task B references E, sees it already claimed, and waits on its Notify.
4. B blocks forever.

Secondary effects: the runtime task panics on an ordinary I/O/parse error, and a reused
ArrowReader stays poisoned for that file across subsequent scans.

The fix

Handle the dropped sender instead of unwrapping it:

• On Ok(predicate): unchanged — store Loaded and notify_waiters().
• On Err(_): remove the stale Loading marker, then notify_waiters().

Waiters that wake to a missing entry now return None (the post-await match no longer
treats that as unreachable!). build_equality_delete_predicate turns that None into a
clear "Missing predicate for equality delete file '…'" error, so the failure is reported
rather than deadlocked. Removing the entry (rather than caching a permanent failure) also
lets a later, independent load of the same file try again.

sequenceDiagram
    participant P as Producer task
    participant S as DeleteFilter state
    participant W as Waiter

    Note over P: eq_del.await -> Err(RecvError)
    P->>S: remove(path)          (drop stale Loading)
    P->>W: notify.notify_waiters()
    W->>S: re-read(path) -> None
    W-->>W: return None  ➜  surfaced as "missing predicate" error ✅

Are these changes tested?

Each test is bounded by tokio::time::timeout. With the fix reverted, all three fail after 5s with a "hung" assertion instead of hanging the test suite.

• test_get_equality_delete_predicate_resolves_when_load_fails: a unit test on DeleteFilter. It drops the predicate sender without sending, which is what a failed parse does, and checks that the waiter gets None.
• test_cloned_readers_error_on_failed_equality_delete_load: builds one ArrowReader, clones it, and reads a task whose equality-delete file can't be read. Both the original and the clone must return an error. The clones share one DeleteFilter, so this tests the shared-reader case from the description directly.
• test_equality_delete_load_retries_after_failure: after a failed read, it restores the delete file and reads again with the same reader. The deletes must be applied ([1, 2, 3, 4, 5] minus 3), which shows that removing the stale entry lets the reader recover.

AI Disclosure

The code was written using Claude but was manually reviewed by me.