rgbuilds commented on code in PR #26534:
URL: https://github.com/apache/datafusion/pull/26534#discussion_r4343799740
##########
datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs:
##########
@@ -324,22 +336,48 @@ impl DynamicFilterPhysicalExpr {
new_expr,
)?;
- // Load the current inner, increment generation, and store the new one
let mut current = self.inner.write();
- let new_generation = current.generation + 1;
- *current = Inner {
- // Preserve the expression id across updates.
- expression_id: current.expression_id,
- generation: new_generation,
- expr: new_expr,
- is_complete: current.is_complete,
- };
- drop(current); // Release the lock before broadcasting
+ current.generation += 1;
+ current.expr = new_expr;
- // Broadcast the new state to all waiters
- let _ = self.state_watch.send(FilterState::InProgress {
- generation: new_generation,
- });
+ // Publish while holding the lock so concurrent updates and completion
+ // cannot reorder notifications. Retain the state for future
subscribers.
+ self.state_watch.send_replace(current.state());
+ Ok(())
+ }
+
+ /// Update the current expression from another
[`DynamicFilterPhysicalExpr`]
+ /// and notify all waiters.
+ ///
+ /// The source's [`PhysicalExpr::expression_id`] must match this filter's.
+ /// The original children must also match, but the filters may have
different
+ /// children remapped by [`PhysicalExpr::with_new_children`].
+ pub fn update_from(&self, source: &Self) -> Result<()> {
+ if Arc::ptr_eq(&self.inner, &source.inner) {
+ return Ok(());
+ }
+
+ // Release the source lock before acquiring the target lock via
clone().
+ let source_state = source.inner.read().clone();
+ if self.children != source.children {
+ return internal_err!(
+ "Cannot update dynamic filters with different original
children"
+ );
+ }
+
+ let mut current = self.inner.write();
+ if current.expression_id != source_state.expression_id {
+ return internal_err!(
+ "Cannot update dynamic filters with different expression IDs:
{} and {}",
+ current.expression_id,
+ source_state.expression_id
+ );
+ }
+
+ current.generation += 1;
+ current.expr = source_state.expr;
+ current.is_complete |= source_state.is_complete;
Review Comment:
Thanks for the fix. I re-reviewed the latest commit locally. The
completed-target guard prevents late incomplete, complete, and duplicate
snapshots from changing the expression, generation, or watch state.
I also restored the previous behavior locally and confirmed the strengthened
regression fails on the original post-completion expression change. This
concern is resolved.
--
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]
