laskoviymishka commented on code in PR #2859:
URL: https://github.com/apache/iceberg-rust/pull/2859#discussion_r3629462968
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -241,11 +241,11 @@ impl CachingDeleteFileLoader {
DataContentType::PositionDeletes => {
match del_filter.try_start_pos_del_load(&task.file_path) {
PosDelLoadAction::AlreadyLoaded =>
Ok(DeleteFileContext::ExistingPosDel),
- PosDelLoadAction::WaitFor(notify) => {
+ 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.
- notify.notified().await;
+ notified.await;
Review Comment:
One thing this fix makes easier to hit: waiters now reliably park on this
await, but if the load errors before `finish_pos_del_load` runs (e.g.
`parse_positional_deletes_record_batch_stream` yields an `Err` and `item?`
propagates), the state stays `Loading` and everyone parked here hangs forever.
Pre-existing, so not a blocker — but I'd want a `fail_pos_del_load` (or an
RAII guard) that flips to a failed state and still calls `notify_waiters()`,
with waiters re-checking state and surfacing the error. wdyt?
##########
crates/iceberg/src/arrow/delete_filter.rs:
##########
@@ -67,9 +68,11 @@ pub(crate) enum PosDelLoadAction {
/// The file is already loaded, nothing to do.
AlreadyLoaded,
/// The file is currently being loaded by another task.
- /// The caller *must* wait for this notifier to ensure data availability
- /// before returning, as subsequent access (get_delete_vector) is
synchronous.
- WaitFor(Arc<Notify>),
+ /// The caller *must* await this future to ensure data availability before
+ /// returning, as subsequent access (get_delete_vector) is synchronous. The
+ /// future is created under the state lock so it cannot miss the loader's
Review Comment:
I'd tweak this to name the actual load-bearing invariant. What makes it safe
isn't "created under the lock" in the abstract — `notified_owned()` snapshots
the `notify_waiters_calls` counter at construction and resolves on first poll
if that counter advanced. The safety is that the snapshot and the state
transition are guarded by the same lock.
Worth spelling out, because a future refactor that moves `notified_owned()`
off the lock (or switches `try_start_pos_del_load` to a read lock) reads as
harmless but silently reintroduces the race. wdyt?
##########
crates/iceberg/src/arrow/delete_filter.rs:
##########
@@ -127,7 +130,9 @@ impl DeleteFilter {
if let Some(state) = state.positional_deletes.get(file_path) {
match state {
PosDelState::Loaded => return PosDelLoadAction::AlreadyLoaded,
- PosDelState::Loading(notify) => return
PosDelLoadAction::WaitFor(notify.clone()),
+ PosDelState::Loading(notify) => {
+ return
PosDelLoadAction::WaitFor(notify.clone().notified_owned());
Review Comment:
While we're here — the equality-delete path has this exact bug unfixed.
`get_equality_delete_predicate_for_delete_file_path` clones the `Arc<Notify>`
under the read lock, drops it, then does `notifier.notified().await` outside,
so a `notify_waiters()` from `insert_equality_delete` in that window is lost —
same hang.
There's also a second race there: `insert_equality_delete` creates a fresh
`Notify` and overwrites the entry from `try_start_eq_del_load`, so a waiter can
end up parked on a different notifier than the one that gets signalled. Out of
scope for this PR (looks tied to #2696), but the same
`notified_owned()`-under-lock fix would close it. Thoughts?
##########
crates/iceberg/src/arrow/delete_filter.rs:
##########
@@ -296,6 +301,38 @@ pub(crate) mod tests {
const FIELD_ID_POSITIONAL_DELETE_FILE_PATH: u64 = 2147483546;
const FIELD_ID_POSITIONAL_DELETE_POS: u64 = 2147483545;
+ // Regression test for the positional-delete lost-wakeup hang.
+ //
+ // Drives the real API through the losing interleaving: the loader fires
+ // `notify_waiters()` (via `finish_pos_del_load`) *before* the waiter
awaits the notifier
+ // handed back by `WaitFor`. `notify_waiters()` stores no permit, so this
only completes if
+ // the waiter's `Notified` was created before the signal. Because
`WaitFor` now carries an
+ // `OwnedNotified` created under the lock in `try_start_pos_del_load`, it
is; on the old
+ // `WaitFor(Arc<Notify>)` contract the waiter created its `Notified` too
late and hung.
+ #[tokio::test]
+ async fn test_wait_for_completes_when_load_finishes_before_await() {
Review Comment:
This sequences finish-before-await, which is exactly the losing order, so
it's a valid check of the counter-snapshot semantics — good.
It doesn't exercise real concurrency though (single-threaded, serial, no
`spawn`), so it won't catch a future break that only shows up under concurrent
scheduling. I'd either add a second test that spawns the waiter and forces the
interleaving with a barrier / `yield_now`, or drop a line in the comment noting
this validates the mechanism rather than scheduling. Either's fine.
##########
crates/iceberg/src/arrow/delete_filter.rs:
##########
@@ -296,6 +301,38 @@ pub(crate) mod tests {
const FIELD_ID_POSITIONAL_DELETE_FILE_PATH: u64 = 2147483546;
const FIELD_ID_POSITIONAL_DELETE_POS: u64 = 2147483545;
+ // Regression test for the positional-delete lost-wakeup hang.
+ //
+ // Drives the real API through the losing interleaving: the loader fires
+ // `notify_waiters()` (via `finish_pos_del_load`) *before* the waiter
awaits the notifier
+ // handed back by `WaitFor`. `notify_waiters()` stores no permit, so this
only completes if
+ // the waiter's `Notified` was created before the signal. Because
`WaitFor` now carries an
+ // `OwnedNotified` created under the lock in `try_start_pos_del_load`, it
is; on the old
+ // `WaitFor(Arc<Notify>)` contract the waiter created its `Notified` too
late and hung.
+ #[tokio::test]
+ async fn test_wait_for_completes_when_load_finishes_before_await() {
+ let filter = DeleteFilter::new(Runtime::current());
+ let path = "s3://bucket/pos-delete.parquet";
+
+ assert!(matches!(
+ filter.try_start_pos_del_load(path),
+ PosDelLoadAction::Load
+ ));
+
+ let PosDelLoadAction::WaitFor(notified) =
filter.try_start_pos_del_load(path) else {
+ panic!("expected WaitFor for an in-progress load");
+ };
+
+ // Loader completes and signals before the waiter awaits.
+ filter.finish_pos_del_load(path);
+
+ let waited = tokio::time::timeout(std::time::Duration::from_secs(5),
notified).await;
Review Comment:
Two small things on this line. This test needs tokio's `time` (and `macros`
for `#[tokio::test]`), but the workspace pin only declares `sync` +
`rt-multi-thread` — it builds today purely via feature unification from
transitive deps, so if the dep tree ever gets slimmed this fails to compile
with a fairly opaque error. I'd declare them explicitly (workspace tokio
features, or a crate-level `tokio = { workspace = true, features = ["time"] }`).
Also, 5s is a lot here — on the fixed path the first poll resolves
synchronously and a real regression hangs indefinitely, so this only ever burns
CI wall-clock when something's broken. I'd drop it to 100–500ms.
--
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]
