Reapply "Add ExecutionPlan::apply_expressions() (apache#20337)" (apache#22437) - #24018
Reapply "Add ExecutionPlan::apply_expressions() (apache#20337)" (apache#22437)#24018jayshrivastava wants to merge 7 commits into
Conversation
github-actions
Bot
added
documentation
jayshrivastava
changed the title
|
Thank you for opening this pull request! Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch). Details |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #24018 +/- ##
==========================================
- Coverage 81.05% 80.98% -0.08%
==========================================
Files 1106 1106
Lines 382210 383149 +939
Branches 382210 383149 +939
==========================================
+ Hits 309812 310286 +474
- Misses 54091 54529 +438
- Partials 18307 18334 +27 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
b8794d2 to
5514194
Compare
| /// `apply_expressions` + `downcast_ref::<DynamicFilterPhysicalExpr>` and | ||
| /// counting nodes. Neither API is observable from SQL. | ||
| #[tokio::test] | ||
| async fn test_discover_dynamic_filters_via_expressions_api() { |
There was a problem hiding this comment.
This test explicitly covers that dynamic filters are discoverable post-pushdown. This is what we want in datafusion-distributed and I assume other projects want this as well
| aggregate.with_dynamic_filter_expr(dynamic_filter)? | ||
| } else { | ||
| let mut aggregate = aggregate; | ||
| aggregate.dynamic_filter = None; |
There was a problem hiding this comment.
Semi-related. I think this was a bug. We kept the default dynamic filter created by try_new around even there's no pushed down filter. Now, we explicitly remove it. The test I added in datafusion/proto/tests/cases/roundtrip_physical_plan.rs covers this.
|
I see the code coverage is isn't very high. I don't see any good candidates to call |
gene-bordegaray
left a comment
gene-bordegaray
left a comment
There was a problem hiding this comment.
I had some questions about the exact contract but overall I think this almost there, just some clarification and looking very clean 👍
| let mut found = false; | ||
| plan.apply(|node| { | ||
| node.apply_expressions(&mut |root| { | ||
| root.apply(|expr| { |
There was a problem hiding this comment.
is there any way to eliminate this triple nesting we do.
From my understanding its:
- walking the execution plan tree
- Walk the expressions roots in one plan node
- Walks inside the one expression
Maybe something like a apply_plan_expressions would make this easier to read?
There was a problem hiding this comment.
I think this pattern is expected. I followed these docs for apply_expressions on logical plans to implement this for physical plans:
datafusion/datafusion/expr/src/logical_plan/plan.rs
Lines 135 to 144 in 5514194
| &self, | ||
| _f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>, | ||
| ) -> Result<TreeNodeRecursion> { | ||
| Ok(TreeNodeRecursion::Continue) |
There was a problem hiding this comment.
Since we already have apply_expression_roots to abstract "walk these expressions" case, would it be worth a small sibling helper for the "I have none" case? Right now that's just Ok(TreeNodeRecursion::Continue) repeated at every leaf/no-op node. Wondering if it would make sense to have something like apply_expression_continue() living next to it, just a plain function.
I think something like apply_expression_roots(std::iter::empty(), f) does the same thing, but since we're trying to abstract away TreeNodeRecursion, wondering if it would still make sense to have the dedicated helper
There was a problem hiding this comment.
apply_expression_roots(std::iter::empty(), f) is nice, but you have to use type annotations in practice:
apply_expression_roots(
std::iter::empty::<&Arc<dyn PhysicalExpr>>(),
f,
)
This is a bit less clean so
I added a helper apply_no_expressions instead.
There was a problem hiding this comment.
Jay (or anyone), feel free to disagree. I suggested this because imo it’s clearer to me, but whatever is clearest when upgrading is best.
| // So here, we try to use ref count to determine if the dynamic filter | ||
| // has actually be pushed down. | ||
| // Issue: <https://github.com/apache/datafusion/issues/18856> | ||
| let child_accepts_dyn_filter = Arc::strong_count(dyn_filter) > 1; |
There was a problem hiding this comment.
Even if the rewrite to use apply_expressions looks like it adds a bunch of code, I think this is a nice use case for it, relying on the strong count of the dynamic filter's inner/outer Arc has caused several headaches before. Maybe we could do something similar in HashJoinExec to determine whether the dynamic filter has a consumer? Right now that's done via is_used, which relies on strong count. IIRC it's only used to check for consumers, so maybe we could remove is_used entirely in favor of this approach?
There was a problem hiding this comment.
This is a good idea! I'll do it rn
|
This PR as it stands is in conflict with #23494. I think it may need a pass over it to align with the |
7550339 to
db63637
Compare
|
@cetra3 Imo, apply_expressions isn't leaking internal state, at least not to the extent prior to #23494. It's just providing iteration over expressions in a plan. It also mirrors the API for logical plan nodes datafusion/datafusion/expr/src/logical_plan/tree_node.rs Lines 418 to 421 in db63637 |
gene-bordegaray
left a comment
gene-bordegaray
left a comment
There was a problem hiding this comment.
last thing, other than that I think this is good 👍
There was a problem hiding this comment.
I believe should we apply expressions to this op because of this as well?
There was a problem hiding this comment.
it is done at line 315 it seems 🤔
There was a problem hiding this comment.
yes but is using apply_no_expressions(f)
I was thinking that since this method can add the ordering to the cached self.cache.eq_properties we would traverse that in the apply_expressions() method here.
Maybe I am still unclear about what needs to be traversed and not @jayshrivastava ?
alamb
left a comment
alamb
left a comment
There was a problem hiding this comment.
Thank you @jayshrivastava and @gene-bordegaray -- I think this is looking quite nice and ready to go
Since I think we are under some time pressure, I took the liberty of merging up from main and resolving conflicts, and tweaking some documentation
I will also make a proposal PR to reduce the
datafusion_physical_plan::apply_expression_roots(
self.required_input_ordering
.iter()
.flatten()
.map(|sort_expr| &sort_expr.expr),
f,
)Boiler plate code, but we can also do that as a follow on
We might also be able to make this pattern easier to read
datafusion_physical_plan::apply_expression_roots(
self.projection
.source
.iter()
.map(|proj_expr| &proj_expr.expr),
f,
)By changing apply_expression_roots to take AsRef<Arc<PhysicalExpr>> and then implementing AsRef for ProjectionExpr
I think it could be collapsed into
datafusion_physical_plan::apply_expression_roots(
self.projection
.source
.iter()
f,
)| slice of the last non-empty input batch) and pass it to window expressions | ||
| via `WindowEvalContext::with_most_recent_row` instead of copying it into | ||
| each partition's state. | ||
| Add `apply_expressions` to your implementation. Call `f` on each top-level |
There was a problem hiding this comment.
This is fine, though it is technically redundant with the (very nice docs) on apply_expressions
There was a problem hiding this comment.
Specifically the examples are redundant I think
|
Sorry hit submit too early -- I am finishing my review now |
|
@alamb Thanks. I'll let you finish the review. Also I can fix up the commits! I figured I'd wait for a review first |
alamb
left a comment
alamb
left a comment
There was a problem hiding this comment.
TLDR looks good to me. I had some small simplification suggestions but I don't think they are required
As I mentioned, I did merge up from main and tweak some comments, but otherwise. think this is ready to go.
Thanks again (and thanks to @LiaCastaneda for starting this work!)
| /// | ||
| /// See [`ExecutionPlan::apply_expressions`] for more details and implementation examples. | ||
| /// | ||
| /// [`ExecutionPlan::apply_expressions`]: datafusion_physical_plan::ExecutionPlan::apply_expressions |
There was a problem hiding this comment.
This is good to direct readers to ExecutionPlan::apply_expressions and then document that heavily
| snapshot: snapshot_fn_wrapper, | ||
| snapshot_generation: snapshot_generation_fn_wrapper, | ||
| is_volatile_node: is_volatile_node_fn_wrapper, | ||
| expression_id: expression_id_fn_wrapper, |
There was a problem hiding this comment.
this appears to be an unrelated change, but a good one
There was a problem hiding this comment.
it is done at line 315 it seems 🤔
| @@ -852,6 +911,65 @@ pub trait ExecutionPlan: Any + Debug + DisplayAs + Send + Sync { | |||
| } | |||
| } | |||
|
|
|||
| /// Implements [`ExecutionPlan::apply_expressions`] for a node with no expressions. | |||
| pub fn apply_no_expressions( | |||
There was a problem hiding this comment.
I would suggest you remove this function (as it is pretty clearly documented what to do when there are no expressions, so it is just one more function that adds complexity but not much help I don't think).
I do realize there is a nice symmetry to have a helper for each case, but in this case I am not sure it is needed
| slice of the last non-empty input batch) and pass it to window expressions | ||
| via `WindowEvalContext::with_most_recent_row` instead of copying it into | ||
| each partition's state. | ||
| Add `apply_expressions` to your implementation. Call `f` on each top-level |
There was a problem hiding this comment.
Specifically the examples are redundant I think
| /// should not be called in any child expressions nor in any expressions of child nodes. | ||
| /// | ||
| /// The closure can return [`TreeNodeRecursion::Stop`] to stop iteration, otherwise | ||
| /// iteration should continue. ([`TreeNodeRecursion::Jump`] and [`TreeNodeRecursion::Jump`] |
There was a problem hiding this comment.
I think this means to say "continue" and "jump" are equivalent. I will push a fix for this
Thanks @jayshrivastava -- I am done, though it appears I (claude) botched the merge. Will fix |
| partition_count, | ||
| generator_count | ||
| ); | ||
| Arc::make_mut(&mut self.cache).partitioning = partitioning; |
There was a problem hiding this comment.
would this also be in the same boat? we would need to traverse these?
6 participants
Which issue does this PR close?
ExecutionPlannodes discoverable #23814This change does not close the above issues because it does not implement a way to tell if a node is a producer dynamic filters.
Rationale for this change
See #23814 and datafusion-contrib/datafusion-distributed#553.
To send dynamic filter updates across the network, there needs to be a way to get access to
PhysicalExprfromExecutionPlan. As discussed in #23814, the cleanest way to do this is to addExecutionPlan::apply_expressions, which mirrors a similar method for logical plan nodes.What changes are included in this PR?
There's 3 commits in this PR:
Firstly, commit 1 re-applies the changes in #20337 (reverted in #22437).
Some of the reasons for why the original PR was reverted include
(a)
apply_expressionsis too complicated to implement and there's no concrete need to justify this complexity(b) there was no usage of
apply_expressionsinside this repoTo address (a)
ExecutionPlannodes discoverable #23814apply_expression_rootsandapply_no_expressionswhich abstract away theTreeNodeRecursioncomplexity from implementors. Now,apply_expressionsvery trivial to implement ex.&Arc<dyn PhysicalExpr>rather than&dyn PhysicalExprto reduce complexity around lifetimesTo address (b):
apply_expressionsinphysical-plan/src/aggregates/mod.rs. Previously, there was a hack that checked if a filter was pushed down usingArc::strong_count(dyn_filter) > 1. Now it usesapply_expressionsis_usedfrom dynamic filters which used to check Arc references counts to see if a filter was pushed down. Now, the hash join usesapply_expressionsto find pushed down filters.Are these changes tested?
Yes.
Are there any user-facing changes?
There's a new mandatory method
ExecutionPlan::apply_expressions(). See the upgrading guide and documentation for details.