jayshrivastava commented on code in PR #26534:
URL: https://github.com/apache/datafusion/pull/26534#discussion_r4352754382
##########
datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs:
##########
@@ -324,22 +336,53 @@ 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`].
+ /// A compatible update is ignored if this filter is already complete.
+ 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 {
Review Comment:
Yeah that makes sense. The children should not be different. Deleted in
favor of just the ID check.
--
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]
