Dandandan opened a new pull request, #24782:
URL: https://github.com/apache/datafusion/pull/24782
## Which issue does this PR close?
- Related to the "single join" of the unnesting paper; no existing issue.
## Rationale for this change
A correlated scalar subquery has to return at most one row for each set of
outer
values. DataFusion enforced that syntactically, in the analyzer: unless the
subquery had an aggregate on top, planning failed with
```
Correlated scalar subquery must be aggregated to return at most one row
```
So the most ordinary shape of correlated subquery there is — looking an
attribute up in a dimension table —
```sql
select o_orderkey, (select c_name from customer where c_custkey = o_custkey)
from orders
```
did not plan at all. To run it you had to wrap the column in an aggregate
(`min`, `max`, `any_value`) whose only purpose was to convince the analyzer
of
something the data already guaranteed, and then pay for a full hash
aggregation
over the subquery side at run time.
Neumann and Kemper's unnesting paper solves this with a *single join*: an
outer
join that emits exactly one row per row of its preserved side and raises an
error if a second row matches. With one, the subquery decorrelates into a
join
and nothing else — no aggregate, no grouping, no extra pass over the data.
## What changes are included in this PR?
- `JoinType::LeftSingle` and `JoinType::RightSingle`, behaving like their
`Left`/`Right` counterparts except that a second match for a row of the
preserved side fails with `Scalar subquery returned more than one row` —
the
same error `ScalarSubqueryExec` already raises for the uncorrelated case.
- `ScalarSubqueryToJoin` emits a single join for correlated scalar subqueries
that are not already known to return at most one row, and keeps the plain
`LEFT JOIN` for the ones that are -- either from an aggregate grouped only
by
columns the correlated predicate fixes, or from functional dependencies
showing the subquery unique on the join keys. The analyzer no longer
rejects
the former.
- `HashJoinExec` implements both directions, so `JoinSelection` can still
swap
the inputs to pick the build side. `NestedLoopJoinExec` implements both for
correlations with no equijoin key. In both operators the existing matched
bitmaps double as the duplicate detector, so the check is one
already-cached
bit test per matched row and covers matches spread over batches and
partitions. `SortMergeJoinExec` and `PiecewiseMergeJoin` reject single
joins,
and the physical planner routes them to an operator that implements them.
- Proto round-trip for the two new variants; the Substrait producer reports
them as unsupported, since Substrait has no equivalent.
Plans that worked before are unchanged: no query in TPC-H (22) or TPC-DS (99)
produces a single join, because every correlated scalar subquery in them
aggregates a real value rather than proving a row count.
## What this costs and what it buys
Measured on TPC-H SF10, comparing the form you had to write before against
the
form that now plans. Median of 7 runs, interleaved, repeated.
| | outer / subquery rows | forced aggregate | single join | |
|---|---|---|---|---|
| lookup in a projection | 15M / 1.5M | 174 ms | 141 ms | **1.24x faster** |
| lookup in a projection | 60M / 2M | 565 ms | 542 ms | 1.04x, within noise |
| lookup in a projection | 100K / 1.5M | 15.1 ms | 6.1 ms | **2.5x faster** |
| lookup in a filter | 15M / 1.5M | 102 ms | 153 ms | **1.5x slower** |
| lookup in a filter, non-pushable predicate | 15M / 1.5M | 90 ms | 139 ms |
**1.55x slower** |
In a projection the aggregate is pure overhead and dropping it is a pure win,
by the share of the work the subquery side represents.
In a filter it is not, and that is worth being precise about. A plain `LEFT
JOIN` lets the optimizer use a predicate on the subquery's output:
`eliminate_outer_join` turns the join inner because the predicate rejects
nulls, `extract_equijoin_predicate` folds a comparison against an outer
column
into a second join key, and `eliminate_join` finishes with a semi join. None
of
that is sound for a single join, because all three change how many rows
match,
which is the one thing the single join has to observe.
So the third commit makes the rule work harder to avoid needing a single join
at all: besides the aggregate it can see from the top of the subquery, it
asks
the decorrelated subquery's functional dependencies whether it is already
unique on the columns the join equates with the outer plan. That covers
uniqueness the subquery inherits rather than declares -- a key that survives
an
extra condition in the correlated predicate, a `GROUP BY` on the correlated
column with nothing to aggregate, uniqueness carried up through the
subquery's
own joins.
With `c_custkey` declared a primary key, the filter cases come out ahead of
the
form they replace, because they now get both the aggregate removal and every
rewrite the left join enables:
| | forced aggregate | single join path | |
|---|---|---|---|
| lookup in a filter | 87 ms | 55 ms | **1.6x faster** |
| lookup in a filter, non-pushable predicate | 97 ms | 85 ms | **1.15x
faster** |
| lookup in a projection | 180 ms | 152 ms | **1.2x faster** |
On a table with no declared key and nothing to infer uniqueness from, the
filter case keeps its ~1.5x cost. That is the price of the stricter
semantics,
and it is opt-in: the aggregated query still plans exactly as it did.
## Are these changes tested?
Yes.
- `subquery.slt` gains a single-join section: attribute lookup with and
without
a match, the duplicate-row error, the same in a filter rather than a
projection, the nested-loop path for a correlation with no equijoin key,
and
plan assertions for each way a subquery can prove it needs no single join
--
an aggregate, a declared key surviving an extra predicate, a bare `GROUP
BY`
-- plus one showing a key proves nothing under an inequality. Four existing
cases changed from "rejected by the analyzer" to running.
- `HashJoinExec` and `NestedLoopJoinExec` unit tests cover both directions
across partition modes and batch sizes, both the accepted and the rejected
case, plus the all-NULL build key path.
- The full sqllogictest suite (504 files) and the workspace test suite pass.
## Are there any user-facing changes?
Yes, all widening:
- Correlated scalar subqueries without an aggregate now plan and run. Where
the
data has a second matching row they fail at run time with `Scalar subquery
returned more than one row` instead of failing to plan.
- A `GROUP BY` in a correlated scalar subquery may now name columns the
correlated predicate does not fix.
- A `LIMIT` above the correlated predicate still cannot be decorrelated; that
case now says so directly (`Correlated scalar subquery with a LIMIT must be
limited to a single row`) rather than reporting a missing aggregate.
- `JoinType` gains two variants. Exhaustive matches on it in downstream code
will need a new arm, so this is an API change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
