Add ExecutionPlan::reset_state by adriangb · Pull Request #17028 · apache/datafusion

adriangb · 2025-08-04T04:51:48Z

datafusion/sqllogictest/test_files/cte.slt

alamb

Thank you @adriangb -- this PR makes sense to me except for the expect / panic -- otherwise I think it is good to go

alamb · 2025-08-04T21:22:33Z

datafusion/physical-plan/src/sorts/sort.rs

+            "implementations have drifted and this is no longer safe even if `new()` still works, ",
+            "for example if `new()` now does something different than just calling `compute_properties(...).unwrap()`",
+            "\n",
+            "This is clearly a bug, please report it!"


I think we should return DatafusionError::Internal here rather than panic'ing as it is a better UX (if you want fail fast in debug builds, perhaps you could add a debug_assert)

I also recommend converting the explanation into comments and leaving the panic message like "Internal inconsistency in SortExec"

The rationale is that if a user sees this message it is not going to mean anything to them and they can't fix it, and this text will obscure the conclusion (this is a bug they can not do anything to fix). A developer will come to the source location and can read the comment.

@alamb this already had the possibility to panic in it because it called SortExec::new():

datafusion/datafusion/physical-plan/src/sorts/sort.rs

Lines 865 to 872 in eb2b8c0

impl SortExec {

/// Create a new sort execution plan that produces a single,

/// sorted output partition.

pub fn new(expr: LexOrdering, input: Arc<dyn ExecutionPlan>) -> Self {

let preserve_partitioning = false;

let (cache, sort_prefix) =

Self::compute_properties(&input, expr.clone(), preserve_partitioning)

.unwrap();

datafusion/datafusion/physical-plan/src/sorts/sort.rs

Lines 1115 to 1119 in eb2b8c0

fn with_new_children(

self: Arc<Self>,

children: Vec<Arc<dyn ExecutionPlan>>,

) -> Result<Arc<dyn ExecutionPlan>> {

let mut new_sort = SortExec::new(self.expr.clone(), Arc::clone(&children[0]))

Returning a Result would be a major breaking change to the ExecutionPlan::with_new_children API which I don't think we should do in this PR.

I do think ExecutionPlan::with_new_children returning a Result would be a good thing. In general I think trait methods should err on the side of returning a result in case some implementation needs to. If none of them do I'd expect compilation to make it pretty much a non issue for performance. But maybe let's do that as it's own PR if we really want to.

alamb · 2025-08-04T21:23:29Z

datafusion/physical-plan/src/execution_plan.rs

+    /// Note to implementers: unlike [`ExecutionPlan::with_new_children`] this method does not accept new children as an argument,
+    /// thus it is expected that any cached plan properties will remain valid after the reset.
+    ///
+    /// [`DynamicFilterPhysicalExpr`]: datafusion_physical_expr::expressions::DynamicFilterPhysicalExpr


I wonder if we should also add a note to DynamicFilterPhysicalExpr saying any ExecutionPlan that uses them should also implement reset_state

xudong963 · 2025-08-05T11:59:55Z

datafusion/physical-plan/src/sorts/sort.rs

+        let (cache, sort_prefix) = Self::compute_properties(
+            &new_sort.input,
+            new_sort.expr.clone(),
+            new_sort.preserve_partitioning,
+        )


Why don't we put the logic into reset_state if the changes are due to the reset_state?

The logic for this change is:

We need to do similar work as with_new_children (i.e. clone SortExec) but each method has slightly different requirements (with_new_children needs to reset cache while reset_state needs to reset filter.

To solve this I created the new SortExec::cloned which does neither of those two things and moved the resetting of cache into with_new_children and the resetting of filter into reset_state.

In other words, it doesn't make sense to put Self::compute_properties(...) in reset_state.

this function returns a Result though, right?

Ah good point yep I'll do that!

Sorry I missed your original point

xudong963

I recommend putting the PR into upgrading doc.

Co-authored-by: Robert Ream <[email protected]>

adriangb · 2025-08-05T16:58:49Z

I recommend putting the PR into upgrading doc.

Good suggestion, added!

adriangb · 2025-08-05T18:04:47Z

Are we concerned that reset_state will disable dynamic the dynamic filter optimization? I wasn't able to cook up an example query, I don't do much recursive queries myself, but I imagine that unless the optimizer runs again after reset_state to re-connect dynamic filters to a data source they can't possibly work. Right?

alamb · 2025-08-06T21:04:05Z

Are we concerned that reset_state will disable dynamic the dynamic filter optimization? I wasn't able to cook up an example query, I don't do much recursive queries myself, but I imagine that unless the optimizer runs again after reset_state to re-connect dynamic filters to a data source they can't possibly work. Right?

I suggest we file a ticket to look into this in the future for anyone who might be interested in that usecase, and not worry about it in this PR

adriangb · 2025-08-06T21:25:42Z

done! #17060

should we wait for anything else before merging this?

alamb · 2025-08-07T21:26:32Z

done! #17060

should we wait for anything else before merging this?

Nope, I think we are good -- let's merge it!

alamb · 2025-08-08T15:49:28Z

It seems like we should probably backport this to branch-49 for the 49.0.1 release

* Add ExecutionPlan::reset_state Co-authored-by: Robert Ream <[email protected]> * Update datafusion/sqllogictest/test_files/cte.slt * Add reference * fmt * add to upgrade guide * add explain plan, implement in more plans * fmt * only explain --------- Co-authored-by: Robert Ream <[email protected]>

* Add ExecutionPlan::reset_state * Update datafusion/sqllogictest/test_files/cte.slt * Add reference * fmt * add to upgrade guide * add explain plan, implement in more plans * fmt * only explain --------- Co-authored-by: Robert Ream <[email protected]>

* Add ExecutionPlan::reset_state Co-authored-by: Robert Ream <[email protected]> * Update datafusion/sqllogictest/test_files/cte.slt * Add reference * fmt * add to upgrade guide * add explain plan, implement in more plans * fmt * only explain --------- Co-authored-by: Robert Ream <[email protected]>

* Enable physical filter pushdown for hash joins (apache#16954) (cherry picked from commit b10f453) * Add ExecutionPlan::reset_state (apache#17028) * Add ExecutionPlan::reset_state Co-authored-by: Robert Ream <[email protected]> * Update datafusion/sqllogictest/test_files/cte.slt * Add reference * fmt * add to upgrade guide * add explain plan, implement in more plans * fmt * only explain --------- Co-authored-by: Robert Ream <[email protected]> * Add dynamic filter (bounds) pushdown to HashJoinExec (apache#16445) (cherry picked from commit ff77b70) * Push dynamic pushdown through CooperativeExec and ProjectionExec (apache#17238) (cherry picked from commit 4bc0696) * Fix dynamic filter pushdown in HashJoinExec (apache#17201) (cherry picked from commit 1d4d74b) * Fix HashJoinExec sideways information passing for partitioned queries (apache#17197) (cherry picked from commit 64bc58d) * disallow pushdown of volatile functions (apache#16861) * dissallow pushdown of volatile PhysicalExprs * fix * add FilteredVec helper to handle filter / remap pattern (#34) * checkpoint: Address PR feedback in https://github.com/apach... * add FilteredVec to consolidate handling of filter / remap pattern * lint * Add slt test for pushing volatile predicates down (#35) --------- Co-authored-by: Andrew Lamb <[email protected]> (cherry picked from commit 94e8548) * fix bounds accumulator reset in HashJoinExec dynamic filter pushdown (apache#17371) --------- Co-authored-by: Adrian Garcia Badaracco <[email protected]> Co-authored-by: Robert Ream <[email protected]> Co-authored-by: Jack Kleeman <[email protected]> Co-authored-by: Andrew Lamb <[email protected]>

github-actions bot added sqllogictest SQL Logic Tests (.slt) physical-plan Changes to the physical-plan crate labels Aug 4, 2025

adriangb mentioned this pull request Aug 4, 2025

Test and fix for issue #16998: SortExec shares DynamicFilterPhysicalExpr across multiple executions #17016

Closed

adriangb force-pushed the fix-bug branch 3 times, most recently from dbcd08c to f26b319 Compare August 4, 2025 05:06

adriangb requested a review from alamb August 4, 2025 05:35

nuno-faria reviewed Aug 4, 2025

View reviewed changes

datafusion/sqllogictest/test_files/cte.slt Outdated Show resolved Hide resolved

alamb approved these changes Aug 4, 2025

View reviewed changes

github-actions bot added the physical-expr Changes to the physical-expr crates label Aug 4, 2025

xudong963 reviewed Aug 5, 2025

View reviewed changes

adriangb and others added 5 commits August 5, 2025 11:53

Add ExecutionPlan::reset_state

6562cc7

Co-authored-by: Robert Ream <[email protected]>

Update datafusion/sqllogictest/test_files/cte.slt

c4838de

Add reference

9391713

fmt

5425c9b

add to upgrade guide

f50d737

adriangb force-pushed the fix-bug branch from 62d80d7 to f50d737 Compare August 5, 2025 16:58

github-actions bot added the documentation Improvements or additions to documentation label Aug 5, 2025

adriangb added 3 commits August 5, 2025 12:38

add explain plan, implement in more plans

5323b11

fmt

9073b6a

only explain

73b5053

adriangb mentioned this pull request Aug 6, 2025

Do dynamic filters (e.g. from SortExec) work with recursive queries? #17060

Open

alamb merged commit f9efba0 into apache:main Aug 7, 2025
28 checks passed

alamb mentioned this pull request Aug 8, 2025

Release DataFusion 49.0.1 (patch) #17036

Closed

5 tasks

adriangb deleted the fix-bug branch August 8, 2025 20:27

wkalt mentioned this pull request Sep 10, 2025

application of simple optimizer rule produces incorrect results (DF 49 regression) #17510

Closed

	impl SortExec {
	/// Create a new sort execution plan that produces a single,
	/// sorted output partition.
	pub fn new(expr: LexOrdering, input: Arc<dyn ExecutionPlan>) -> Self {
	let preserve_partitioning = false;
	let (cache, sort_prefix) =
	Self::compute_properties(&input, expr.clone(), preserve_partitioning)
	.unwrap();

	fn with_new_children(
	self: Arc<Self>,
	children: Vec<Arc<dyn ExecutionPlan>>,
	) -> Result<Arc<dyn ExecutionPlan>> {
	let mut new_sort = SortExec::new(self.expr.clone(), Arc::clone(&children[0]))

Conversation

adriangb commented Aug 4, 2025

Uh oh!

Uh oh!

alamb left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

adriangb Aug 4, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

xudong963 left a comment

Choose a reason for hiding this comment

Uh oh!

adriangb commented Aug 5, 2025

Uh oh!

adriangb commented Aug 5, 2025

Uh oh!

alamb commented Aug 6, 2025

Uh oh!

adriangb commented Aug 6, 2025

Uh oh!

alamb commented Aug 7, 2025

Uh oh!

Uh oh!

alamb commented Aug 8, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

adriangb Aug 4, 2025 •

edited

Loading