Support dynamic filters and work-table plans state reset

There is a reset_plan_states function which is intended to re-use an existing ExecutionPlan:

datafusion/datafusion/physical-plan/src/execution_plan.rs

Lines 1401 to 1423 in d90d074

    
           /// Make plan ready to be re-executed returning its clone with state reset for all nodes. 
        
           /// 
        
           /// Some plans will change their internal states after execution, making them unable to be executed again. 
        
           /// This function uses [`ExecutionPlan::reset_state`] to reset any internal state within the plan. 
        
           /// 
        
           /// An example is `CrossJoinExec`, which loads the left table into memory and stores it in the plan. 
        
           /// However, if the data of the left table is derived from the work table, it will become outdated 
        
           /// as the work table changes. When the next iteration executes this plan again, we must clear the left table. 
        
           /// 
        
           /// # Limitations 
        
           /// 
        
           /// While this function enables plan reuse, it does not allow the same plan to be executed if it (OR): 
        
           /// 
        
           /// * uses dynamic filters, 
        
           /// * represents a recursive query. 
        
           /// 
        
           pub fn reset_plan_states(plan: Arc<dyn ExecutionPlan>) -> Result<Arc<dyn ExecutionPlan>> { 
        
               plan.transform_up(|plan| { 
        
                   let new_plan = Arc::clone(&plan).reset_state()?; 
        
                   Ok(Transformed::yes(new_plan)) 
        
               }) 
        
               .data() 
        
           }

Currently, it does not support dynamic filters and work table of the recursive query, because they are injected into nodes once on the planning stage and then never updated. Actually, to re-use a plan, dynamic filters must be reset.

Possible implementation

Add ExecutionPlan::dynamic_filters(...) method which is intended to return current version of dynamic filters.
Introduce PushedDynamicFilters struct to pass into ExecutionPlan::with_new_state method.
Modify reset_plan_states to use the following algorithm:
- Create an empty PushedDynamicFilters
- Create an option work_table (None initially)
- Transform plan up to bottom
- When process a node: reset its state, use with_new_state(pushed_filters) and with_new_state(work_table) if work_table is Some. Then collect new dynamic filters from it and possibly a work table if it is a RecursiveQueryExec (successfull downcast).
For each node that polls dynamic filter support with_new_state where will be an attempt to down-cast state into PushedDynamicFilters.

take

@askalt

@askalt are you still working on this? I looked at the same area and would like to suggest a simpler model, based on what main does today.

What happens today in a recursive CTE (the recursive term is reset with reset_state before each iteration). The three dynamic-filter producers each do something different:

Producer	`reset_state`	Effect inside a recursive term
`AggregateExec` (MIN/MAX)	none, keeps the filter	wrong results from iteration 2 (#26054)
`HashJoinExec`	drops its filter (`hash_join/exec.rs`, `dynamic_filter = None`)	the scan keeps the old filter, which is never updated: no pruning
`SortExec` (TopK)	creates a new filter (`sort.rs`, `create_filter()`)	the scan keeps the old filter, which is never updated: no pruning

Repro for the hash join (main, 81824e8):

SET datafusion.execution.target_partitions = 1;
SET datafusion.execution.parquet.pushdown_filters = true;
COPY (SELECT value AS k FROM generate_series(1, 100)) TO '/tmp/t.parquet';
CREATE EXTERNAL TABLE t STORED AS PARQUET LOCATION '/tmp/t.parquet';

EXPLAIN ANALYZE WITH RECURSIVE w(n) AS (
  SELECT 1 UNION ALL SELECT t.k FROM w JOIN t ON t.k = w.n + 1 WHERE w.n < 5
) SELECT * FROM w ORDER BY n;
-- scan of t: DynamicFilter [ empty ], pushdown_rows_pruned=0

EXPLAIN ANALYZE SELECT t.k FROM (VALUES (1), (2)) w(n) JOIN t ON t.k = w.n + 1;
-- scan of t: DynamicFilter [ k@0 >= 2 AND k@0 <= 3 AND ... ], pushdown_rows_pruned=98

This matters for graph walks, the common use of recursive CTEs. On a 20M-row edge table sorted by src, one join step takes 3 ms with the dynamic filter (19 of 20 row groups pruned) and 24 ms without it; a 10-step recursive walk takes 0.90 s today.

Suggested model: a producer owns one filter and resets its value, never its identity. The scan already holds a reference to that filter from planning. If reset_state clears the value in place instead of dropping or replacing the filter, the scan sees the next iteration's values with no re-wiring, so the dynamic_filters() / PushedDynamicFilters plumbing proposed above would not be needed.

#26056 already does this for AggregateExec and adds DynamicFilterPhysicalExpr::reset(). After it lands, the remaining steps are small:

HashJoinExec::reset_state keeps its filter and calls reset() instead of setting it to None, plus a cte.slt case that checks pruning in a recursive join.
SortExec::reset_state does the same instead of create_filter(), plus a cte.slt case for TopK.

If you are no longer working on this, would it be OK to mark it help wanted with these steps?

@jayzhan211

This makes sense to me @jayzhan211

Suggested model: a producer owns one filter and resets its value, never its identity.

The initial intention was to give an ability to reuse an existing physical plan. This model allows to do it, however each.execute(...) must be called strictly after reset_plan_state and, hence, after the previous stream is exhausted. To execute the same query in parallel, we need to create a physical plan N times for it. It seems acceptable to me. It would be cool to reflect this in a doc.

If you are no longer working on this, would it be OK to mark it help wanted with these steps?

Yes, I am definitely won't be able to work on this issue the next 2 weeks, so it is OK to mark it help wanted.

Navigation Menu

Repository navigation

Support dynamic filters and work-table plans state reset #19905

Possible implementation

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	/// Make plan ready to be re-executed returning its clone with state reset for all nodes.
	///
	/// Some plans will change their internal states after execution, making them unable to be executed again.
	/// This function uses [`ExecutionPlan::reset_state`] to reset any internal state within the plan.
	///
	/// An example is `CrossJoinExec`, which loads the left table into memory and stores it in the plan.
	/// However, if the data of the left table is derived from the work table, it will become outdated
	/// as the work table changes. When the next iteration executes this plan again, we must clear the left table.
	///
	/// # Limitations
	///
	/// While this function enables plan reuse, it does not allow the same plan to be executed if it (OR):
	///
	/// * uses dynamic filters,
	/// * represents a recursive query.
	///
	pub fn reset_plan_states(plan: Arc<dyn ExecutionPlan>) -> Result<Arc<dyn ExecutionPlan>> {
	plan.transform_up(\|plan\| {
	let new_plan = Arc::clone(&plan).reset_state()?;
	Ok(Transformed::yes(new_plan))
	})
	.data()
	}

Support dynamic filters and work-table plans state reset #19905

Description

Possible implementation

Activity

askalt commented on Jan 20, 2026

jayzhan211 commented on Oct 16, 2026

adriangb commented on Oct 17, 2026

askalt commented on Oct 19, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions