feat: support stateless physical plans #19462
Open
+3,232
−910
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This patch introduces the stateless physical plan feature. Currently, the physical-plan crate is fully supported. This feature allows for the reuse of physical plans and their concurrent execution. The feature is implemented by adding a separate Cargo feature named "stateless_plan". The implementation consists of several parts: * State tree. With the "stateless_plan" feature enabled, the plans themselves do not store state. The state is stored in a separate tree composed of PlanStateNodes, which is built lazily during plan execution. Each node of the tree stores not only the shared state of the plan but also its metrics. The shape of the state tree matches the shape of the execution plan tree. * Metrics Metrics are stored in the nodes of the state tree and can be accessed after plan execution. Support is provided for performing EXPLAIN using the state. * Dynamic Filters In the case of stateless plans, dynamic filters cannot simply be stored inside the plans, as the same plan can be executed concurrently. To overcome this, a dynamic filter is split into two parts: a planning-time version and an execution-time version. The plans contain the planning-time version, which is transformed into the execution version during the execution phase and then passed from parent nodes to child nodes using the state tree. * WorkTable Instead of explicitly injecting the WorkTable into nodes, RecursiveExec exposes the WorkTable in the state stored within the State Tree. Then, a node interested in obtaining the WorkTable traverses up the State Tree and thus retrieves the current WorkTable. Planned following work: - Support stateless plan for all other DataFusion crates. - Enable running tests with this feature in CI. - Deprecate stateful plans to eventually transition completely to the stateless version. - Add `fmt_as_with_state` to allow plans to include state-specific details in the EXPLAIN output, such as dynamic filters. Closes apache#19351
github-actions
bot
added
physical-expr
Contributor
Author
|
The following diagram explains the implementantion: 
Where:
|
Contributor
Author
|
@alamb Could you please help to review this proof of concept? There is a rather large diff because all physical plan states have been moved from the plan itself. However, the core idea can be grasped quickly—for example, by looking into the changes in https://github.com/apache/datafusion/blob/7e009ff2db1b2fbd2760923497f18b2a7c46557e/datafusion/physical-plan/src/state.rs |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Closes #19351
What changes are included in this PR?
This patch introduces the stateless physical plan feature. Currently, the physical-plan crate is fully supported. This feature allows for the reuse of physical plans and their concurrent execution.
The feature is implemented by adding a separate Cargo feature named "stateless_plan". The implementation consists of several parts:
State tree
With the "stateless_plan" feature enabled, the plans themselves do not store state. The state is stored in a separate tree composed of PlanStateNodes, which is built lazily during plan execution. Each node of the tree stores not only the shared state of the plan but also its metrics. The shape of the state tree matches the shape of the execution plan tree.
Metrics
Metrics are stored in the nodes of the state tree and can be accessed after plan execution. Support is provided for performing EXPLAIN using the state.
Dynamic Filters
In the case of stateless plans, dynamic filters cannot simply be stored inside the plans, as the same plan can be executed concurrently. To overcome this, a dynamic filter is split into two parts: a planning-time version and an execution-time version. The plans contain the planning-time version, which is transformed into the execution version during the execution phase and then passed from parent nodes to child nodes using the state tree.
WorkTable
Instead of explicitly injecting the WorkTable into nodes, RecursiveExec exposes the WorkTable in the state stored within the State Tree. Then, a node interested in obtaining the WorkTable traverses up the State Tree and thus retrieves the current WorkTable.
Are these changes tested?
Currently only locally as the patch introduces a new isolated feature which is not tested in CI yet.
Following work
fmt_as_with_stateto allow plans to include state-specific details in the EXPLAIN output, such as dynamic filters.