timsaucer commented on code in PR #26534:
URL: https://github.com/apache/datafusion/pull/26534#discussion_r4379410496
##########
datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs:
##########
@@ -372,20 +409,14 @@ impl DynamicFilterPhysicalExpr {
/// Waiters using [`Self::wait_complete`] will be notified.
pub fn mark_complete(&self) {
let mut current = self.inner.write();
- let current_generation = current.generation;
current.is_complete = true;
- drop(current);
-
- // Broadcast completion to all waiters
- let _ = self.state_watch.send(FilterState::Complete {
- generation: current_generation,
- });
+ self.state_watch.send_replace(current.state());
Review Comment:
Switching `mark_complete()` (and `update()`) to `send_replace` means the
watch channel now always stores `Complete`, even when there are no receivers.
But `reset()` (merged in from main, a few lines above) still uses `let _ =
self.state_watch.send(...)` after dropping the lock. `watch::Sender::send`
returns an error **without storing the value** when there are no receivers,
which is the common case since `new()`/`from_parts()` drop the initial receiver.
Sequence:
1. `mark_complete()` → `send_replace` stores `Complete`
2. `reset()` with no subscribers → `send` fails, channel still holds
`Complete`
3. `wait_complete()` → `inner.is_complete == false`, subscribes, sees stale
`Complete`, returns immediately on a reset filter
Repro (passes on main, fails on this branch):
```rust
#[test]
fn reset_after_complete_without_receivers() {
let f = DynamicFilterPhysicalExpr::new(vec![], lit(true));
f.update(lit(false)).unwrap();
f.mark_complete();
f.reset();
let mut cx = Context::from_waker(Waker::noop());
let mut w = pin!(f.wait_complete());
assert!(w.as_mut().poll(&mut cx).is_pending());
}
```
This affects the TopK filter re-armed across recursive CTE iterations.
Suggested fix, matching the new pattern:
```rust
pub fn reset(&self) {
let mut current = self.inner.write();
current.generation += 1;
current.expr = lit(true);
current.is_complete = false;
self.state_watch.send_replace(current.state());
}
```
It would be worth adding the repro above as a regression test too.
--
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]
