[DISCUSS] Representing Shared State / ExecutionPlan::reset_state

@Dandandan

Is your feature request related to a problem or challenge?

I brought this out into its own discussion so it wasn't lost on a PR comment

@Dandandan observes on #21351 (comment):

I don't like the additions in some recent PRs of reset_state on plans / sources etc.

In my opinion we should avoid setting mutable state on ExecutionPlans (and related data structures) & move that to the inner streams.

In an ideal scenario:

During any time we can clone a plan and share / run it without expecting any side effects (e.g. partially executed plans)
Concurrent runs of a query (ideallyexecutes) should result in the same result, even while a query is running.
In my PR #20481 I combined the shared state using TaskContext (which, in normal usages, will be created on each new query run), which I think still not is 100% the desired end state but avoids a lot of scenario's.

https://github.com/apache/datafusion/pull/20481/files#diff-e3975f93f1d4599fe6f19905e81f02574d65402e604e39c45b87448b3892d1adR70

Describe the solution you'd like

No response

Describe alternatives you've considered

No response

Additional context

No response

In my opinion we should avoid setting mutable state on ExecutionPlans (and related data structures) & move that to the inner streams.

I think actually most of the state is already on the inner streams. I believe the core usecase for mutable state on ExecutionPlans is making connections between the inner streams.

Sometimes this connection is for streams that come from the same ExecutionPlan -- e.g. the CollectLeft in joins

Sometimes the connection is for streams for different ExecutionPlans -- e.g dynamic filters need to source from a TopK operator

I combined the shared state using TaskContext (which, in normal usages, will be created on each new query run), which I think still not is 100% the desired end state but avoids a lot of scenario's.

I really like this idea of using TaskContext to store the shared state somehow. It maps quite nicely to existing patterns and I think would not be super disruptive. I think if we got the general purpose API right, that could also solve a lot of challenges (like how to serialize dynamic filter references, for example) -- e.g. the dynamic filters could get some sort of id and then the actual runtime connection could be made by looking up in the TaskContext

Here are some examples (found with codex, reviewed by me) of this shared state:

Details

Shared recursive CTE work table

This seems like the clearest example of explicit shared mutable operator/runtime state.

WorkTable itself is mutable shared state (Mutex<Option<ReservedBatches>>)
WorkTableExec stores Arc<WorkTable>

Hash join shared build-side state, especially `CollectLeft`

HashJoinExec stores a shared left_fut: Arc<OnceAsync<JoinLeftData>> and optional dynamic_filter
JoinLeftData contains shared runtime artifacts: hash map, visited bitmap, bounds, membership, and atomics
In execute, PartitionMode::CollectLeft shares one left_fut across all probe partitions

Dynamic filter information on joins

HashJoinExecDynamicFilter holds the shared DynamicFilterPhysicalExpr plus OnceLock<Arc<SharedBuildAccumulator>>
The dynamic filter is also exposed as part of the plan expressions in apply_expressions
In execute, the dynamic filter path lazily creates a shared SharedBuildAccumulator
SharedBuildAccumulator is the cross-partition shared state used to accumulate bounds / membership and update the filter
Filter updates happen in SharedBuildAccumulator when all build partitions report in

Other join operators with shared build-side state

These follow the same general pattern: plan owns a OnceAsync<_> that multiple output partitions/streams wait on.

`CrossJoinExec`

CrossJoinExec has shared left_fut: OnceAsync<JoinLeftData>
reset_state explicitly recreates left_fut
execute shares the buffered left side through left_fut.try_once(...)

`NestedLoopJoinExec`

NestedLoopJoinExec has shared build_side_data: OnceAsync<JoinLeftData>

`PiecewiseMergeJoinExec`

PiecewiseMergeJoinExec has shared buffered_fut: OnceAsync<BufferedSideData>

`SortExec`

SortExec::reset_state recreates the TopK dynamic filter and resets metrics
The TopK self-filter is pushed down from operator-owned state in gather_filters_for_pushdown

`AggregateExec`

AggregateExec::init_dynamic_filter can attach a dynamic_filter owned by the plan

@askalt

I think @askalt @adriangb @gabotechs @xudong963 @geoffreyclaude @gene-bordegaray and others may have some thoughts about this topic as well

@askalt

I think @askalt @adriangb @gabotechs @xudong963 @geoffreyclaude @gene-bordegaray and others may have some thoughts about this topic as well

The main reason I worked on this was datafusion-tracing: recursive queries broke once plans were wrapped, because the runtime wiring depended on concrete WorkTableExec handling. That led to with_new_state in #16469 and the follow-up in #18803.

So I’m sympathetic to moving shared state off plans, but whatever replaces it should still support wrapper/custom ExecutionPlans without concrete type checks. TaskContext seems like a plausible direction if it preserves that.

In general, not baking shared state into execution plans is greatly beneficial to any system the relies on plan [de]serialization to work, for example FFI-based wrappers (https://github.com/apache/datafusion-python) or distributed engines (https://github.com/apache/datafusion-ballista or https://github.com/datafusion-contrib/datafusion-distributed).

Indeed, one of the main challenges for integrating dynamic filters with a distributed system is the collision with the assumption that data can simply be shared through Arc-based shared references.

I actually gave it a try some time ago in gabotechs#7, but abandoned the project as a #18739 was a prerequisite but was not very well received.

I can imagine how not relying on shared state inside can add some complexity if not done with care, but IMO it's something worth pursuing.

@alamb

It sounds like the architectural debt that will compound — every new feature that needs shared state (dynamic filters for more join types, CTE optimizations, plan caching) will make the problem worse.

Let me ensure that I totally understand the issue (correct me @alamb )

Physical plan — a declarative description of what to compute. Its lifecycle is session-level: created once, potentially reused, cached, or serialized.
Execution state — the runtime artifacts produced while computing: hash tables, buffers, coordination signals. Its lifecycle is per-query-run: created fresh each time, discarded after.

DataFusion currently binds execution state to plan objects. HashJoinExec holds Arc<OnceAsync>, CrossJoinExec holds OnceAsync, WorkTableExec holds Arc, and dynamic filters thread Arc references between join and scan nodes. This lifecycle mismatch is the root cause of three concrete problems:

Plans are not reusable. OnceAsync is consumed after one run. Re-execution requires reset_state() — which is itself an admission that state is in the wrong place.
Plans are not serializable. Arc pointers are memory addresses. They cannot be written to protobuf and sent to another machine. This blocks distributed execution (Ballista), FFI wrappers (datafusion-python), and plan caching.
Plans are not safely concurrent. Two concurrent runs of the same plan race on the same OnceAsync. The second run may observe residual state from the first.

If so, I agree with the TaskContext direction:

Add a SharedStateRegistry to TaskContext — a HashMap<SharedStateId, Arc<dyn Any + Send + Sync>> where SharedStateId is a lightweight identifier (probably a u64 or a newtype around usize).
Plan nodes carry SharedStateId instead of Arc<State> — this is serializable, cloneable, and doesn't carry execution state. The physical planner assigns IDs when creating plans.
State initialization happens in a new prepare_execution phase — before the first execute() call, walk the plan tree and let each node register its shared state in the TaskContext. This replaces the current pattern where state is lazily initialized in execute().

TaskContext indeed sounds like a natural place for holding the shared state. I can imagine some pending things that can probably be easily addressed:

Some entities need to access shared state but have no access to TaskContext. For example, DynamicFilterPhysicalExpr, as a PhysicalExpr implementation, has no access to TaskContext (see PhysicalExpr::evaluate)
TaskContext is not mutable, so representing shared state with structs like a HashMap implies that all entities that require shared state are registered prior to building the TaskContext itself, rather than lazily at runtime. This might not be a problem at all.

@gabotechs

@gabotechs yes

And taskContext satisfies all four constraints:

A new TaskContext is created for each query run → State is naturally isolated
Passed via Arc to all execute(partition, context) calls within the same plan → Shared within the same run
Does not belong to any plan node → Shared across nodes
Plan carries no execution state — becomes serializable

@xudong963

Our DF based DBMS project assumes that an execution plan can be reused, so for some time we struggled with the presence of a state in the plans (and currently we forcely rebuild the physical plan if dynamic filters are enabled or it is a recursive query). In our DF fork we stored a state in TaskContext for some time: this solution is described in details here #19351 and it is seems similar to @xudong963 suggestion:

Add a SharedStateRegistry to TaskContext — a HashMap<SharedStateId, Arc<dyn Any + Send + Sync>> where SharedStateId is a lightweight identifier (probably a u64 or a newtype around usize).

Additionally it defines the plan state trait:

/// Generic plan state.
pub trait PlanState: std::fmt::Debug + std::any::Any + Send + Sync {
    /// Represent this state as any.
    fn as_any(&self) -> &dyn std::any::Any;
    
    /// Return plan metrics if some. 
    fn metrics(&self) -> Option<crate::metrics::ExecutionPlanMetricsSet>;
}

We used ExecutionPlan raw pointer addresses as SharedStateId as actually we need to establish the correspondence between plan nodes and its state after an execution is finished: e.g. to recover node metrics.

State initialization happens in a new prepare_execution phase — before the first execute() call

We need to figure out a solution to keep it backward compatible, as currently any DF based code expects that a pure .execute(...) call is enough to launch a plan properly. e.g. keep a state in both plan and task context for some time (until statefull plans are deprecated) and decide which one to use based on the TaskContext attached value.

@alamb

Let me ensure that I totally understand the issue (correct me @alamb )

Yes that is a good summary @xudong963

State initialization happens in a new prepare_execution phase — before the first execute() call, walk the plan tree and let each node register its shared state in the TaskContext. This replaces the current pattern where state is lazily initialized in execute().

That is an interesting idea -- another thing we could do would be create it on demand -- something like

impl TaskContext {
  fn shared_state(&self, id: SharedStateId, F: InitailizerFunc) -> ... 
}

but maybe that would get too tricky 🤔

We need to figure out a solution to keep it backward compatible, as currently any DF based code expects that a pure .execute(...) call is enough to launch a plan properly. e.g. keep a state in both plan and task context for some time (until statefull plans are deprecated) and decide which one to use based on the TaskContext attached value.

@askalt one thing we could do would be to change reset_state to clear out the TaskContext's shared stae (and then deprecate reset_state). I think that woudl be backwards compatible

Nice to hear the concern is shared (both from a idealogical / architecture standpoint as with some concrete technical examples).

I will add that I also think the state via TaskContext should share as little as possible:

Only keep it local to each returned stream when it is only for each partition
Shared among the streams across partitions for a single operator (so it is dropped when all streams are done)
Only bound to TaskContext when it has to be shared during the full query (and will be dropped when TaskContext is dropped <- perhaps this can be avoided in majority of cases

@neilconway

BTW I think @neilconway was adding shared state while implementing subquery execution

perf: Implement physical execution of uncorrelated scalar subqueries #21240

@neilconway

Interestingly, @neilconway 's approach in

perf: Implement physical execution of uncorrelated scalar subqueries #21240

Was to attach subquery state to the ExecutionProps -- not the TaskContext

Navigation Menu

Repository navigation

[DISCUSS] Representing Shared State / `ExecutionPlan::reset_state` #21650

Is your feature request related to a problem or challenge?

Describe the solution you'd like

Describe alternatives you've considered

Additional context

Shared recursive CTE work table

Hash join shared build-side state, especially `CollectLeft`

Dynamic filter information on joins

Other join operators with shared build-side state

`CrossJoinExec`

`NestedLoopJoinExec`

`PiecewiseMergeJoinExec`

`SortExec`

`AggregateExec`

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

[DISCUSS] Representing Shared State / ExecutionPlan::reset_state #21650

Description

Is your feature request related to a problem or challenge?

Describe the solution you'd like

Describe alternatives you've considered

Additional context

Activity

alamb commented on Apr 15, 2026

alamb commented on Apr 15, 2026

alamb commented on Apr 15, 2026

Shared recursive CTE work table

Hash join shared build-side state, especially CollectLeft

Dynamic filter information on joins

Other join operators with shared build-side state

CrossJoinExec

NestedLoopJoinExec

PiecewiseMergeJoinExec

SortExec

AggregateExec

alamb commented on Apr 15, 2026

geoffreyclaude commented on Apr 15, 2026

gabotechs commented on Apr 15, 2026

xudong963 commented on Apr 16, 2026

gabotechs commented on Apr 16, 2026

xudong963 commented on Apr 16, 2026

askalt commented on Apr 16, 2026

alamb commented on Apr 16, 2026

Dandandan commented on Apr 16, 2026

alamb commented on Apr 19, 2026

alamb commented on Apr 27, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions

[DISCUSS] Representing Shared State / `ExecutionPlan::reset_state` #21650

Hash join shared build-side state, especially `CollectLeft`

`CrossJoinExec`

`NestedLoopJoinExec`

`PiecewiseMergeJoinExec`

`SortExec`

`AggregateExec`