mbutrovich commented on code in PR #3435:
URL: https://github.com/apache/iceberg-rust/pull/3435#discussion_r4260367658
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -271,25 +271,31 @@ impl CachingDeleteFileLoader {
return Self::load_deletion_vector(task,
basic_delete_file_loader).await;
}
- match del_filter.try_start_pos_del_load(task.file_path()) {
- PosDelLoadAction::AlreadyLoaded =>
Ok(DeleteFileContext::ExistingPosDel),
- 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.
- notified.await;
- Ok(DeleteFileContext::ExistingPosDel)
+ loop {
+ match del_filter.try_start_pos_del_load(task.file_path()) {
+ PosDelLoadAction::AlreadyLoaded => {
+ return Ok(DeleteFileContext::ExistingPosDel);
+ }
+ 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;
+ }
Review Comment:
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](https://github.com/apache/iceberg-rust/pull/2859#discussion_r3629462968)).
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](https://github.com/apache/iceberg-rust/blob/deff41be05833040d4aeb39e78a5f5dd8a0c1553/crates/iceberg/src/arrow/caching_delete_file_loader.rs#L207),
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`](https://github.com/apache/iceberg-rust/blob/deff41be05833040d4aeb39e78a5f5dd8a0c1553/crates/iceberg/src/error.rs#L134-L144)
isn't `Clone`. Which behavior do we want here?
##########
crates/iceberg/src/arrow/delete_filter.rs:
##########
@@ -163,6 +200,29 @@ impl DeleteFilter {
}
}
+ /// Removes the `Loading` marker of a positional delete file whose load
failed or was
+ /// dropped, and wakes any waiting tasks so they can try the load again.
+ 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();
+ }
+ }
Review Comment:
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?
```suggestion
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();
}
}
```
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -1791,4 +1797,55 @@ mod tests {
assert_eq!(err.kind(), ErrorKind::DataInvalid);
assert!(err.message().contains("expected 2 from record_count"));
}
+
+ #[tokio::test]
+ async fn test_pos_del_load_can_be_retried_after_failure() {
Review Comment:
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](https://github.com/apache/iceberg-rust/pull/2859#discussion_r3618445925)).
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).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
