spark-test commented on code in PR #60091:
URL: https://github.com/apache/spark/pull/60091#discussion_r4486457776
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ValidateSubqueryExpression.scala:
##########
@@ -239,9 +250,37 @@ object ValidateSubqueryExpression
}
}
+ /**
+ * Returns the [[CTERelationDef]]s of `plan`'s [[WithCTE]]s that a
[[CTERelationRef]] in `plan`
+ * references, keyed by CTE id. The single-pass resolver validates a
subquery before its CTEs
+ * are inlined, so validation follows these references instead. References
inside subquery
+ * expressions count only if `includeReferencesInSubqueries` is set.
+ */
+ private def referencedCteDefinitions(
+ plan: LogicalPlan,
+ includeReferencesInSubqueries: Boolean = true): Map[Long,
CTERelationDef] = {
+ if (!plan.containsPattern(CTE)) {
+ Map.empty
+ } else {
+ val collectReferencedId: PartialFunction[LogicalPlan, Long] = {
+ case ref: CTERelationRef => ref.cteId
+ }
+ val referencedIds = if (includeReferencesInSubqueries) {
+ plan.collectWithSubqueries(collectReferencedId).toSet
+ } else {
+ plan.collect(collectReferencedId).toSet
+ }
+ plan.collect { case withCte: WithCTE => withCte.cteDefs }.flatten
Review Comment:
**Non-blocking (P2):** Definitions come only from `WithCTE` nodes inside the
plan being validated. The single-pass resolver validates a subquery as soon as
its hosting operator resolves, before it places the `WithCTE` of the enclosing
subquery or query, so a reference to a CTE defined in an enclosing scope is
still an opaque leaf here. The enclosing subquery's own correlation check
doesn't count nested references either, so it checks such a CTE only in place.
Fixed point validates after inlining, where the body sits at the reference.
That leaves the failure this PR fixes in place one scope out. Single pass
accepts
```sql
select *, (with cte as (select * from y where y1 = x1)
select count(*) from z
where exists (select * from z zz left join cte on zz.z1 =
cte.y1)) from x
```
which fixed point rejects with
`ACCESSING_OUTER_QUERY_COLUMN_IS_NOT_ALLOWED`. With
`spark.sql.optimizer.scalarSubqueryUseSingleJoin=false`, single pass also still
rejects `with c as (select * from y where y2 = 0) select *, (select count(*)
from c where c.y1 = x1 group by c.y1, c.y2) from x` with
`NON_CORRELATED_COLUMNS_IN_GROUP_BY`, while fixed point accepts it. The new
nested-EXISTS golden case can't tell the two behaviors apart, because its CTE
body already fails in place. This affects only the single-pass resolver.
**Recommended change:** Pass the enclosing-scope definitions that the
subquery's references resolve to from the single-pass resolver into
ValidateSubqueryExpression, and add them to its definition maps. Also count
references from nested subqueries in the correlation check, so a definition
used only inside a nested subquery is validated where that subquery references
it rather than in place. Add golden cases for a correlated CTE used on the
null-supplying side of a LEFT JOIN inside a nested EXISTS, and, with the single
join disabled, for a grouped scalar subquery over an outer-query CTE that pins
a grouping column to a constant.
**Why this works:** With the resolved definitions available, a nested
subquery's validation follows its reference into the enclosing body under its
own context, which is exactly where fixed point checks the inlined body. The
enclosing subquery then no longer checks that definition at its own WithCTE.
**Scope:** Supply enclosing-scope CTE definitions to single-pass subquery
validation, and validate definitions used only from nested subqueries where
those subqueries reference them.
**Risks:** Supplying a definition that a reference does not resolve to, for
example across a view boundary or behind a shadowing inner WITH, would validate
the wrong body; the lookup must use each reference's definition id.
**Constraints:** The fixed-point call site supplies no extra definitions, so
fixed-point behavior is unchanged.
**Success:** Single pass rejects a correlated CTE used on the null-supplying
side of an outer join inside a nested subquery, as fixed point does. With the
single join disabled, single pass accepts a grouped scalar subquery whose
grouped column is pinned inside an outer-query CTE, as fixed point does.
**Verification:**
- **Regression:** A scalar subquery whose correlated CTE is used only inside
a nested EXISTS, on the right side of a LEFT JOIN, fails in dual run with
ACCESSING_OUTER_QUERY_COLUMN_IS_NOT_ALLOWED and both analyzers agreeing.
- **Regression:** With the single join disabled, dual run accepts, with both
analyzers agreeing, a grouped scalar subquery over an outer-query CTE that pins
one grouping column to a constant.
- **Compatibility:** The nested-subquery golden case this pull request adds
and the existing cte-nested.sql cases keep their results.
##########
sql/core/src/test/resources/sql-tests/inputs/subquery/scalar-subquery/scalar-subquery-group-by.sql:
##########
@@ -44,3 +44,21 @@ set
spark.sql.legacy.scalarSubqueryAllowGroupByNonEqualityCorrelatedPredicate =
set spark.sql.optimizer.scalarSubqueryUseSingleJoin = false;
select * from x where (select count(*) from y where y1 > x1 group by y1) = 1;
reset
spark.sql.legacy.scalarSubqueryAllowGroupByNonEqualityCorrelatedPredicate;
+
+-- Scalar subqueries with a WITH clause are validated with their CTEs inlined.
+set spark.sql.legacy.scalarSubqueryAllowGroupByNonEqualityCorrelatedPredicate
= false;
Review Comment:
**Nit (P3):** This block sets and resets only
`spark.sql.legacy.scalarSubqueryAllowGroupByNonEqualityCorrelatedPredicate`. It
inherits `spark.sql.optimizer.scalarSubqueryUseSingleJoin = false` from the
legacy-behavior block above (line 44), which never resets it, and the shape and
group-by checks this PR changes run only under that setting. If that block
later gains a reset or moves below this one, the legal cases pass even with the
CTE look-through reverted, the two `MUST_AGGREGATE_CORRELATED_SCALAR_SUBQUERY`
cases return rows, and the `y1 > x1` case fails only at runtime with
`SCALAR_SUBQUERY_TOO_MANY_ROWS`. A regenerated golden would record those
outcomes, and the block would stop testing the new code.
See **Shared repair plan 2** in the review body.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ValidateSubqueryExpression.scala:
##########
@@ -38,6 +39,8 @@ object ValidateSubqueryExpression
plan: LogicalPlan,
expr: SubqueryExpression,
skipOuterReferenceValidationInFilter: Boolean = false): Unit = {
+ lazy val referencedCtes = referencedCteDefinitions(expr.plan)
Review Comment:
**Nit (P3):** This map counts references inside nested subquery expressions,
while the correlation check at line 421 doesn't. That lets the shape check skip
a `WithCTE` whose CTEs are used only from nested subqueries, so with
`spark.sql.optimizer.scalarSubqueryUseSingleJoin=false` single pass now accepts
`select *, (with cte as (select * from y) select count(*) from z where z1 = x1
and exists (select * from cte) group by z1) from x`, as fixed point does after
inlining; the base branch rejected it with
`MUST_AGGREGATE_CORRELATED_SCALAR_SUBQUERY`. No new case covers this: the only
nested-reference case fails on its FULL OUTER JOIN however references are
counted. If this counting changed, for example in the fix for references inside
unused CTEs, single pass would reject the query again and no test would fail.
See **Shared repair plan 1** in the review body.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ValidateSubqueryExpression.scala:
##########
@@ -239,9 +250,37 @@ object ValidateSubqueryExpression
}
}
+ /**
+ * Returns the [[CTERelationDef]]s of `plan`'s [[WithCTE]]s that a
[[CTERelationRef]] in `plan`
+ * references, keyed by CTE id. The single-pass resolver validates a
subquery before its CTEs
+ * are inlined, so validation follows these references instead. References
inside subquery
+ * expressions count only if `includeReferencesInSubqueries` is set.
+ */
+ private def referencedCteDefinitions(
+ plan: LogicalPlan,
+ includeReferencesInSubqueries: Boolean = true): Map[Long,
CTERelationDef] = {
+ if (!plan.containsPattern(CTE)) {
+ Map.empty
+ } else {
+ val collectReferencedId: PartialFunction[LogicalPlan, Long] = {
+ case ref: CTERelationRef => ref.cteId
+ }
+ val referencedIds = if (includeReferencesInSubqueries) {
+ plan.collectWithSubqueries(collectReferencedId).toSet
+ } else {
+ plan.collect(collectReferencedId).toSet
Review Comment:
**Blocking (P1):** This set counts every `CTERelationRef` in the plan,
including references inside the body of a CTE that nothing uses. The default
fixed-point analyzer validates the output of `InlineCTE(alwaysInline = true,
keepDanglingRelations = true, isAnalysis = true)`, which keeps such dangling
definitions with their original bodies (`cleanCTEMap` discounts the references
they make). So a definition referenced only from an unused definition now lands
in the map, is skipped at its `WithCTE`, and is validated at that reference,
with the reference's context.
For example, with the views from scalar-subquery-group-by.sql:
```sql
select *, (with a as (select * from y where y1 = x1),
b as (select * from z left join a on z1 = y1)
select count(*) from z) from x
```
The subquery's query uses neither CTE. On the base branch this analyzes: `a`
is checked in place and the reference inside `b` is a leaf, just as the
existing cte-nested.sql case `SELECT (WITH unreferenced AS (SELECT id) SELECT
1) FROM range(1)` analyzes and returns 1. With this change the reference on the
right side of `b`'s LEFT JOIN is followed into `a`, and the default analyzer
fails with `ACCESSING_OUTER_QUERY_COLUMN_IS_NOT_ALLOWED`. The same applies to
EXISTS, IN and lateral subqueries and to the other correlation-restricted
positions, although the description says the fixed-point analyzer's behavior is
unchanged. If the subquery's query also reads `a` (`... select count(*) from
a`), fixed point still accepts it, because inlining removes `a`, but single
pass now rejects it, which dual run reports as
`SINGLE_PASS_FAILED_FIXED_POINT_SUCCEEDED`.
See **Shared repair plan 1** in the review body.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/subquery.scala:
##########
@@ -369,7 +376,10 @@ object SubExprUtils extends PredicateHelper {
_: Range |
_: SubqueryAlias =>
ExpressionSet(plan.children.flatMap(child =>
- getCorrelatedEquivalentInnerExpressions(child)))
+ equivalentExpressions(child)))
+
+ case cteRef: CTERelationRef if cteDefinitions.contains(cteRef.cteId) =>
Review Comment:
**Non-blocking (P2):** This returns equivalences expressed in the
definition's attribute IDs for every reference it reaches. In single pass only
the first `CTERelationRef` keeps the definition's output IDs
(`ExpressionIdAssigner` gives later references fresh ones), and fixed point
inlines a re-instanced reference as an alias `Project` over a deduplicated
copy, whose equivalences carry fresh IDs. So an equality found through a later
reference is credited to the first reference's columns.
With `spark.sql.optimizer.scalarSubqueryUseSingleJoin=false`:
```sql
select *, (with cte as (select * from y where y1 = 2)
select count(*) from z left join cte a on z2 = a.y2 join cte b on
true
where z1 = x1 group by a.y1) from x
```
Single pass accepts this, because `b`'s `y1 = 2` is credited to `a.y1`,
while fixed point rejects it with `NON_CORRELATED_COLUMNS_IN_GROUP_BY` (dual
run: `FIXED_POINT_FAILED_SINGLE_PASS_SUCCEEDED`). For `x1 = 1` the subquery
returns two groups (`a.y1 = 2` and `NULL`), so with the single-pass resolver
enabled the left-outer rewrite duplicates that outer row. The same happens when
the first reference is the first branch of a `UNION ALL`. This affects only the
single-pass resolver.
**Recommended change:** In the group-by equivalence traversal, follow a CTE
reference only when it keeps its definition's output attributes, or when it is
the reference through which cleanQueryInScalarSubquery reached the aggregate
being checked; pass that path from the shape check to nonEquivalentGroupbyCols.
Add scalar-subquery group-by golden cases, with the single join disabled, for
four queries: a CTE's first reference on the null-supplying side of a LEFT JOIN
and, separately, in the first branch of a UNION ALL, while a later reference
supplies an equality on the column the query groups by; a query that groups by
a later reference's own column; and a query whose aggregate is reached through
a re-instanced reference to its CTE.
**Why this works:** Fixed point gives every re-instanced reference a private
copy, whose equivalences can only qualify an aggregate inside that copy.
Following only the references that keep the definition's attributes, plus the
reference on the path to the aggregate, gives single pass the same set of
usable equivalences.
**Scope:** Which CTE references contribute group-by equivalences in
single-pass validation.
**Risks:** Queries whose aggregate sits behind a re-instanced reference
depend on the shape path being passed through; if the path is lost, single pass
alone would reject them.
**Constraints:** The optimizer's one-row check keeps the default empty map
and is unaffected.
**Success:** Single pass rejects a grouped correlated scalar subquery whose
grouping column is pinned only through another reference to the same CTE, as
fixed point does. A CTE referenced once in a grouped scalar subquery
contributes the same equivalences as before this repair.
**Verification:**
- **Regression:** With the single join disabled, scalar subqueries that
group by a column of a CTE's first reference, where that reference sits on a
path that propagates no equivalences, such as the null-supplying side of a LEFT
JOIN or the first branch of a UNION ALL, while a later reference supplies an
equality on that column, fail in dual run with
NON_CORRELATED_COLUMNS_IN_GROUP_BY and both analyzers agreeing.
- **Behavior:** Dual run accepts, with both analyzers agreeing, a scalar
subquery whose aggregate is reached through a re-instanced reference to its CTE.
- **Compatibility:** With the single join disabled, a scalar subquery that
groups by a later reference's own column fails in dual run with
NON_CORRELATED_COLUMNS_IN_GROUP_BY and both analyzers agreeing.
- **Compatibility:** The legal and illegal CTE cases this pull request adds
keep their recorded outcomes.
##########
sql/core/src/test/resources/sql-tests/inputs/subquery/scalar-subquery/scalar-subquery-group-by.sql:
##########
@@ -44,3 +44,21 @@ set
spark.sql.legacy.scalarSubqueryAllowGroupByNonEqualityCorrelatedPredicate =
set spark.sql.optimizer.scalarSubqueryUseSingleJoin = false;
select * from x where (select count(*) from y where y1 > x1 group by y1) = 1;
reset
spark.sql.legacy.scalarSubqueryAllowGroupByNonEqualityCorrelatedPredicate;
+
+-- Scalar subqueries with a WITH clause are validated with their CTEs inlined.
+set spark.sql.legacy.scalarSubqueryAllowGroupByNonEqualityCorrelatedPredicate
= false;
+-- Legal queries
+select *, (with cte as (select y1, count(*) as cnt from y where y1 = x1 group
by y1) select cnt from cte) from x;
+select *, (with cte as (select * from y where y1 = x1) select count(*) from
cte group by y1) from x;
+select *, (with cte1 as (select * from y where y1 = x1), cte2 as (select y1,
count(*) as cnt from cte1 group by y1) select cnt from cte2) from x;
+select *, (with cte as (select * from y where y1 = x1) select count(*) from
cte left join z on y2 = z2 group by y1) from x;
+-- Illegal queries
+select *, (with cte as (select y1, count(*) as cnt from y where y1 > x1 group
by y1) select cnt from cte) from x;
Review Comment:
**Nit (P3):** The description says these illegal cases fail with the same
errors as in fixed point, but they run only through dual run, and
`HybridAnalyzer` rethrows the fixed-point exception whenever both analyzers
fail. This case and the grouped left-join case below were already rejected by
the base branch's single pass, with
`MUST_AGGREGATE_CORRELATED_SCALAR_SUBQUERY`, so their goldens are identical
with and without this change. Nothing observes the error single pass reports
for the new negative cases, so a later change that makes it report a different
error for them, as users would see once the resolver is enabled, would pass
every test. Suites such as `AsOfJoinSQLSuite` already assert single-pass errors
with `checkError` under `spark.sql.analyzer.singlePassResolver.enabled`.
**Verification:**
- **Regression:** With the single-pass resolver enabled on its own, every
new negative CTE case that both analyzers reject fails with the same error
condition as in fixed point; in particular the grouped CTE case with a
non-equality correlation fails with NON_CORRELATED_COLUMNS_IN_GROUP_BY, and the
grouped left-join CTE case fails with
ACCESSING_OUTER_QUERY_COLUMN_IS_NOT_ALLOWED.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ValidateSubqueryExpression.scala:
##########
@@ -402,6 +448,9 @@ object ValidateSubqueryExpression
// are on the path from the operator hosting the correlated expressions
// up to the operator producing the correlated values.
plan match {
+ case cteRef: CTERelationRef if referencedCtes.contains(cteRef.cteId) =>
Review Comment:
**Nit (P3):** This follows the reference into its body every time, with no
record of what was already validated, and the `WithCTE` case skips referenced
definitions, so a body is checked once per reference path rather than once per
definition. In single pass, `select * from x where exists (with c0 as (select *
from y), c1 as (select * from c0 union all select * from c0), ..., cK as
(select * from c(K-1) union all select * from c(K-1)) select * from cK where y1
= x1)` validates `c0`'s body 2^K times (about 10^6 at K = 20), where the base
branch visited each definition once. The new arm in
`getCorrelatedEquivalentInnerExpressions` repeats the pattern through inner
joins and INTERSECT. The verdict depends only on the body, `aggregated` and
`canContainOuter`, so the repeated visits add work without changing any result.
This affects only the single-pass resolver; the default analyzer already
validates an inlined plan of comparable size.
**Verification:**
- **Behavior:** Single-pass analysis of a subquery whose WITH clause chains
twenty CTEs, each referencing the previous one twice, finishes promptly and
validates each definition a bounded number of times.
- **Behavior:** With the single join disabled, single-pass analysis of a
correlated grouped scalar subquery whose WITH clause chains twenty CTEs, each
inner-joining two references to the previous one, computes each definition's
equivalence set once and finishes promptly.
- **Compatibility:** Subquery and CTE golden files keep their results in
dual run.
##########
sql/core/src/test/resources/sql-tests/inputs/sql-udf.sql:
##########
@@ -748,6 +748,13 @@ SELECT foo3_2e1(
ARRAY(STRUCT('2022-01-01 10:11:12', '1'), STRUCT('2022-01-01 10:11:15',
'2')),
'2022-01-01');
+-- Correlated scalar function whose body aggregates in a CTE
+CREATE FUNCTION foo3_2e2(x INT) RETURNS INT RETURN
+WITH t AS (SELECT c1, SUM(c2) AS s FROM t2 WHERE c1 = x GROUP BY c1)
+SELECT s FROM t;
+
+SELECT c1, foo3_2e2(c1) FROM t1;
Review Comment:
**Nit (P3):** The description says this case will cover the single-pass path
once SQL functions are supported, but it can't fail for the changed validation,
now or then. Today `ResolverGuard` rejects non-builtin functions, so single
pass never runs it, and fixed point inlines the CTE before validating. This
file also never sets `spark.sql.optimizer.scalarSubqueryUseSingleJoin`, and
with its default (`true`) the scalar shape and group-by checks this PR changes
are skipped. The remaining correlation check accepts the body both in place and
at the reference, so even a future single pass gives the same result with or
without this change.
See **Shared repair plan 2** in the review body.
--
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]
