datafusion-contrib / datafusion-contrib/datafusion-distributed
Support for `ScalarSubqueryExec` and `ScalarSubqueryExpr`
Nobody has claimed this yet.
- Dominant language
- Rust
- Stars
- 139
- Forks
- 67
- Avg merge
- 3d 1h
- Merged PRs (30d)
- 35
Description
PR https://github.com/apache/datafusion/pull/21240 introduced a node to execute subqueries, ScalarSubqueryExec which structure holds a shared state:
pub struct ScalarSubqueryExec {
...
results: ScalarSubqueryResults,
}
pub struct ScalarSubqueryResults {
slots: Arc<Vec<Mutex<Option<ScalarValue>>>>,
}
where the a given subquery has an index to fill it's corresponding slot at runtime:
pub struct ScalarSubqueryExpr {
...
results: ScalarSubqueryResults,
}
If ScalarSubqueryExec and the node holding the ScalarSubqueryExpr are in different stages, the query fails during deserialization with the following error Internal("ScalarSubqueryExpr can only be deserialized as part of a surrounding ScalarSubqueryExec") . Even if we edited the decoder to create a fresh ScalarSubqueryResults on the worker, the result wouldn't be correct I think, the worker's slots are a different Arc from the coordinator, so the writer would never fill them. This looks like a similar problem to Dynamic Filtering (where the HashJoinExec holds the dynamic filter in a shared Arc with the probe/consumer). Note that DynamicFilterPhysicalExpr does not fail on serialization and we will never see correctness issues since it's an optimization.
For now we will be gating the new node behind a session property https://github.com/apache/datafusion/pull/22530. This change will probably stay to a couple of releases more (until distributed execution engines adapt), so we need to add support for this eventually.
Some raw ideas:
- Implement a communication mechanism that works for both use cases:
ScalarSubqueryExec,ScalarSubqueryExprandDynamicFilterPhysicalExpr. - Do not have network boundaries between
ScalarSubqueryExecand the node containing theScalarSubqueryExpr.
cc @jayshrivastava @adriangb
Contributor guide
No contributing guide indexed for this repository
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
Start by tracing serialization and deserialization for ScalarSubqueryExec, ScalarSubqueryExpr, and DynamicFilterPhysicalExpr, focusing on how shared Arc state crosses stages. Compare the existing dynamic-filter situation with the scalar-subquery failure. Done means distributed execution can serialize these nodes without losing the shared results state or producing incorrect query results.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust
- Domain
- distributed-systems
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Quiet
- Clarity
- Needs clarification
- Newbie friendliness
- 30/100