pratham76 commented on code in PR #58656:
URL: https://github.com/apache/spark/pull/58656#discussion_r4056565047
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -450,10 +471,100 @@ object RewritePredicateSubquery extends
Rule[LogicalPlan] with PredicateHelper {
ExistenceJoin(exists), newConditions, joinHint)
introducedAttrs += exists
exists
+ // A sub-query that `canRewrite` declined is left as it is, children
included.
+ case sq @ (_: Exists | Not(_: InSubquery) | _: InSubquery) => sq
+ case other => other.mapChildren(rewrite)
}
}
+ val newExprs = exprs.map(rewrite)
(newExprs.reduceOption(And), newPlan, introducedAttrs.toSeq)
}
+
+ /**
+ * Returns true if `e` references any of the attributes produced by `plan`.
+ */
+ private def referencesPlan(e: Expression, plan: LogicalPlan): Boolean = {
+ e.references.intersect(plan.outputSet).nonEmpty
+ }
+
+ /**
+ * Rewrites the existential sub-queries that are nested in the correlated
predicates which
+ * [[PullupCorrelatedPredicates]] hoisted out of a predicate sub-query, and
which therefore end
+ * up in the condition of the semi/anti join that replaces that sub-query.
+ *
+ * A hoisted predicate can carry a nested existential sub-query out of the
sub-query plan,
+ * because a correlated predicate is hoisted as a whole when it is a
disjunction. For example
+ *
+ * SELECT * FROM t1 WHERE EXISTS (
+ * SELECT 1 FROM t2 WHERE t1.a = t2.c1 OR t2.c1 IN (SELECT col1 FROM t3))
+ *
+ * hoists `a = c1 OR c1 IN (SELECT col1 FROM t3)` into the join condition.
Such a nested
+ * sub-query must be rewritten against the plan that produces the attributes
it references:
+ * the one above references `c1`, which is produced by the sub-query plan
and not by the outer
+ * plan, so its existence join has to be built on top of the sub-query plan.
Building it on top
+ * of the outer plan instead yields a join whose condition references an
attribute that neither
+ * of its children can produce (SPARK-59351).
+ *
+ * A nested sub-query that references both plans cannot be rewritten into an
existence join on
+ * either side, so it is left in the join condition, where both plans are in
scope. This is only
+ * correct for an uncorrelated sub-query: the join condition of a correlated
one is dropped when
+ * it is planned as an in-subquery filter, which would silently change the
result, so it is
+ * reported as unsupported instead. Note that such a sub-query can be
correlated only to the
+ * sub-query plan, as being correlated to the outer plan as well would
require two levels of
+ * correlation, which the Analyzer rejects.
+ *
+ * Returns the rewritten condition along with the updated outer and
sub-query plans.
+ */
+ private def rewriteExistentialExprInJoinCondition(
+ conditions: Seq[Expression],
+ outerPlan: LogicalPlan,
+ subPlan: LogicalPlan): (Option[Expression], LogicalPlan, LogicalPlan) = {
+ val (subCond, newSubPlan) =
+ rewriteExistentialExprInSubqueryPlan(conditions, outerPlan, subPlan)
+ // The sub-queries that do not reference the sub-query plan are rewritten
against the outer
+ // plan, as they only reference attributes of the outer plan, if any.
+ val (newCond, newOuterPlan, _) = rewriteExistentialExprWithAttrs(
+ subCond.toSeq, outerPlan, e => !referencesPlan(e, subPlan))
+ (newCond, newOuterPlan, newSubPlan)
+ }
+
+ /**
+ * Rewrites the existential sub-queries in `conditions` that can only be
evaluated by the
+ * sub-query plan, that is those referencing the sub-query plan but not the
outer plan, into
+ * existence joins on top of the sub-query plan. See
+ * [[rewriteExistentialExprInJoinCondition]] for details.
+ *
+ * Returns the rewritten condition along with the updated sub-query plan.
+ */
+ private def rewriteExistentialExprInSubqueryPlan(
+ conditions: Seq[Expression],
+ outerPlan: LogicalPlan,
+ subPlan: LogicalPlan): (Option[Expression], LogicalPlan) = {
+ // A correlated sub-query referencing both plans has to stay in the join
condition, where its
+ // own join condition is lost, so reject it rather than silently return a
wrong result.
+ val unsupported = conditions.flatMap(_.collect {
+ case sq @ (_: Exists | _: InSubquery)
+ if isCorrelatedSubquery(sq) &&
+ referencesPlan(sq, subPlan) && referencesPlan(sq, outerPlan) => sq
+ })
+ if (unsupported.nonEmpty) {
+ throw
QueryCompilationErrors.unsupportedCorrelatedSubqueryInJoinConditionError(unsupported)
+ }
+ val (newCond, newSubPlan, _) = rewriteExistentialExprWithAttrs(conditions,
subPlan,
Review Comment:
Confirmed, and thank you — this was a wrong-result regression on the new
route. Reproduced with
`t1(a)=(1,2,3)`, `t2(c1)=(1,8,9)`, `t3(col1)=(3,NULL)`, where the NULL
matches nothing in `t2`:
```sql
SELECT * FROM t1 WHERE EXISTS (
SELECT 1 FROM t2 WHERE a = c1 OR ((c1 IN (SELECT col1 FROM t3)) <=>
false));
```
| | result |
|---|---|
| PostgreSQL / DuckDB | `1` |
| master | `INTERNAL_ERROR` (attribute not found) |
| this PR, before this fixup | **`1, 2, 3`** |
and the mirror case `(c1 IN (SELECT col1 FROM t3)) IS NULL` returned `1`
where it should return
`1, 2, 3`. Exactly as you describe: `NULL <=> false` is FALSE while `exists
<=> false` is TRUE,
because the `exists` attribute is FALSE where IN is unknown.
One finding worth recording before the fix: **this is not specific to the
new route.** The same
substitution on the outer route and in a projection is already wrong on
released 4.0.1, with no
correlation involved:
```sql
-- 4.0.1 returns 1, 2 ; PostgreSQL/DuckDB return 1
SELECT * FROM t1 WHERE EXISTS (
SELECT 1 FROM t2 WHERE a = c1 OR ((a IN (SELECT col1 FROM t3)) <=> false));
-- 4.0.1 returns true, true, false ; PostgreSQL/DuckDB return false, false,
false
SELECT a, ((a IN (SELECT col1 FROM t3)) <=> false) FROM t1;
```
So the owner-level representation you describe would fix a pre-existing bug
as well as this one.
### What this fixup does
The distinction is only observable when the IN value escapes a predicate
position. Within `AND`,
`OR` and `NOT` under a `Filter` or a join condition — which is what a
hoisted correlated predicate
normally is — unknown and FALSE are indistinguishable, and the existing
two-valued rewrite is
exact. So the fixup rejects precisely the observable case on the new route:
* the IN value reaches something other than a predicate **and** its row
comparison can be unknown
(`values.zip(listQuery.plan.output).exists(_.nullable || _.nullable)`) ->
rejected with
`0A000 ... NESTED_IN_SUBQUERY_WITH_UNKNOWN_RESULT`, independent of
`decorrelatePredicateSubqueriesInJoinPredicate`;
* otherwise -> rewritten as before.
The ordinary shape therefore keeps working with nullable columns, which
matters because real tables
are nullable and that is the case the JIRA is about:
```sql
-- returns 1, as PostgreSQL and DuckDB do, with t3 = (3, NULL)
SELECT * FROM t1 WHERE EXISTS (
SELECT 1 FROM t2 WHERE a = c1 OR c1 IN (SELECT col1 FROM t3));
```
Leaving the subquery in the join condition is not an alternative here: under
the default
configuration `RewritePredicateSubquery` re-offers the join it just produced
to its own handling of
predicate subqueries in join conditions, which rebuilds the same two-valued
existence join. That is
why the rejection is in the rewrite itself, and why the test asserts it
under both settings of that
conf. NOT IN is rejected in that position too, since its null-aware join
condition is equally
two-valued; its behaviour as a predicate is untouched.
### On the full representation change
I did not implement the TRUE/FALSE/NULL reconstruction in this PR.
Reconstructing it needs two
existence joins per nested IN — one on the comparison and one on `cond OR
isnull(cond)`, which is
what gives correct row-comparison semantics for multi-column IN, since
`isnull` of the conjunction
is true exactly when no equality is FALSE and at least one is NULL — and the
second join needs a
copy of the subquery plan with fresh expression ids. Applying that at the
owner, as you ask, also
changes the outer and projection routes, i.e. long-standing plans and
results, well beyond this
JIRA.
Given that, I would rather do it as a follow-up with its own JIRA and
plan-stability review than
fold it in here, while this PR stops the new route from producing wrong
rows. Happy to do it in
this PR instead if you prefer the whole scope in one change — and equally
happy to file the
follow-up now with the 4.0.1 repros above, since the pre-existing half is a
correctness bug in its
own right. Let me know which you want.
--
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]
