datafusion-contrib / datafusion-contrib/datafusion-distributed

Support for `ScalarSubqueryExec` and `ScalarSubqueryExpr`

Open
#471 0 comments 1 reaction 0 assignees View on GitHub

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, ScalarSubqueryExpr and DynamicFilterPhysicalExpr.
  • Do not have network boundaries between ScalarSubqueryExec and the node containing the ScalarSubqueryExpr.

cc @jayshrivastava @adriangb

Contributor guide

No contributing guide indexed for this repository

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.