Dandandan opened a new pull request, #24784:
URL: https://github.com/apache/datafusion/pull/24784
## Which issue does this PR close?
- Closes #.
## Rationale for this change
A correlated scalar subquery must return at most one row per set of outer
values. DataFusion enforced this in the analyzer by requiring an aggregate on
top of the subquery. Without one, planning failed with
```
Correlated scalar subquery must be aggregated to return at most one row
```
So a plain attribute lookup did not plan at all:
```sql
select o_orderkey, (select c_name from customer where c_custkey = o_custkey)
from orders
```
To run it you had to wrap the column in `min`, `max` or `any_value`. That
aggregate does no useful work; it only proves a row count that the data
already
guarantees, and it costs a full hash aggregation over the subquery side.
The single join from Neumann and Kemper's unnesting paper removes the need
for
it. It is an outer join that emits one row per row of its preserved side and
raises an error when a second row matches, so the subquery decorrelates into
a
join with no aggregate at all.
## What changes are included in this PR?
- `JoinType::LeftSingle` and `JoinType::RightSingle`. They behave like `Left`
and `Right` except that a second match for a row of the preserved side
fails
with `Scalar subquery returned more than one row`, the error
`ScalarSubqueryExec` already raises for uncorrelated scalar subqueries.
- `ScalarSubqueryToJoin` uses a single join only for subqueries not already
known to return at most one row. Subqueries that are known to, from an
aggregate grouped on correlated columns or from functional dependencies
showing the subquery unique on the join keys, keep their plain `LEFT JOIN`.
The analyzer no longer rejects the rest.
- `HashJoinExec` implements both directions, so `JoinSelection` can still
swap
the inputs to choose a build side. `NestedLoopJoinExec` implements both for
correlations with no equijoin key. Both reuse the matched bitmaps as the
duplicate detector, so the check costs one 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 new variants. The Substrait producer reports them
as
unsupported, since Substrait has no equivalent.
No query in TPC-H (22) or TPC-DS (99) produces a single join, so no plan in
either suite changes. Every correlated scalar subquery in them aggregates a
real value rather than proving a row count.
## Benchmarks
TPC-H SF10, comparing the form you had to write before against the form that
now plans. Median of 7 runs, interleaved and repeated.
| | outer / subquery rows | forced aggregate | single join | |
|---|---|---|---|---|
| lookup in a projection | 100K / 1.5M | 15.1 ms | 6.1 ms | 2.5x faster |
| lookup in a projection | 15M / 1.5M | 174 ms | 141 ms | 1.24x faster |
| lookup in a projection | 60M / 2M | 565 ms | 542 ms | within noise |
| 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 removing it saves the
share
of the work the subquery side represents.
In a filter it is not that simple. A plain `LEFT JOIN` lets the optimizer
use a
predicate on the subquery output: `eliminate_outer_join` makes 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`
turns the result into a semi join. None of these are sound for a single join,
because all three change how many rows match, which is what the single join
must observe.
So the rule tries harder to avoid needing a single join. Besides the
aggregate
it can see at 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,
and
uniqueness carried up through the subquery's own joins.
With `c_custkey` declared a primary key, the filter cases end up faster than
the form they replace, because they get both the aggregate removal and the
rewrites 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 no way to infer uniqueness, 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 before.
## Are these changes tested?
Yes.
- `subquery.slt` gains a single join section: attribute lookup with and
without
a match, the more-than-one-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 show 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, the accepted and the rejected case,
and the all-NULL build key path.
- The full sqllogictest suite (504 files) and the workspace test suite pass.
## Are there any user-facing changes?
- Correlated scalar subqueries without an aggregate now plan and run. When
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 reports `Correlated scalar subquery with a LIMIT must be limited
to
a single row` instead of reporting a missing aggregate.
- `JoinType` gains two variants, so exhaustive matches on it in downstream
code
need a new arm. 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]
