AveryQi115 commented on code in PR #58269:
URL: https://github.com/apache/spark/pull/58269#discussion_r3867969389
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -89,29 +89,45 @@ case class InlineCTE(
}
private def validateNoOuterReferencesAcrossCTEBoundary(cteDef:
CTERelationDef): Unit = {
- val outerRefs = cteDef.child.flatMap(
- _.expressions.flatMap(_.collect { case o: OuterReference => o }))
- if (outerRefs.nonEmpty) {
+ // Only an outer reference that actually escapes the CTE definition is
invalid. An outer
+ // reference that points to an attribute produced by an operator inside
the definition
+ // body (e.g. a correlated subquery whose correlated column lives in the
body) does not
+ // escape, so the definition is self-contained and safe to materialize.
Walk the whole
+ // definition (main tree plus nested subquery plans) once, collect every
attribute bound
+ // anywhere in the definition, and reject only references that resolve to
none of them.
+ val allNodes = cteDef.child.collectWithSubqueries { case n: LogicalPlan =>
n }
+ val boundExprIds = allNodes.iterator
Review Comment:
Using exprId to match the outer references and outer scope references might
not be very reliable, especially for dataframe cases, which can contain
duplicate exprIds without running single pass analyzer.
I use
```
val outerRefs = cteDef.child.flatMap(
_.expressions.flatMap(_.collect { case o: OuterReference => o }))
```
because this is the previous behavior check before `forceSkipInline` is
added. If you want this validation to skip self-contained correlations, we can
change the validation to be:
1. check if there are direct outer references exists without traversing to
the subquery.plan or subquery child. If a direct outer reference exists in
cteDef body, then it must refer to attributes outside the cteDef.
```
val outerRefs =
cteDef.child.exists(_.expressions.exists(SubExprUtils.containsOuter))
```
2. check if there are subquery expressions has `OuterScopeReference` in
cteDef body. This means that the outer references within the subquery plan is
not from the direct parent scope (which is the cteDef itself) of the subquery.
```
val outerScopeSubqueries = cteDef.child.flatMap(
_.expressions.flatMap(_.collect {
case s: SubqueryExpression if s.outerScopeAttrs.nonEmpty => s
}))
```
I'm also okay with the current approach. But we'd better add some dataframe
testcases.
--
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]
