peter-toth commented on code in PR #58661: URL: https://github.com/apache/spark/pull/58661#discussion_r3977890098
########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/MaterializedCTECheck.scala: ########## @@ -0,0 +1,100 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.catalyst.analysis + +import scala.collection.mutable + +import org.apache.spark.sql.catalyst.expressions.{AttributeSet, OuterReference, OuterScopeReference, SubqueryExpression} +import org.apache.spark.sql.catalyst.plans.logical.{CTERelationDef, CTERelationRef, LogicalPlan} +import org.apache.spark.sql.catalyst.trees.TreePattern.CTE +import org.apache.spark.sql.errors.QueryCompilationErrors + +/** + * Checks that a MATERIALIZED CTE does not reference the query enclosing it, as it is evaluated + * once on its own. The CTEs it references, transitively, are checked against the same boundary, + * since they are inlined into it. An outer reference crosses the boundary only if it targets an + * attribute that is not produced within it: a CTE nested in a subquery of the definition may + * legitimately be correlated to the definition's own relations. The check covers the given plan + * and all its subqueries. + */ +object MaterializedCTECheck extends (LogicalPlan => Unit) { + override def apply(plan: LogicalPlan): Unit = { + if (plan.containsPattern(CTE)) { + // All CTE definitions, including those of nested subqueries, so that references from a + // MATERIALIZED CTE can be followed across subquery boundaries. + val cteDefs = mutable.LinkedHashMap.empty[Long, CTERelationDef] + plan.foreachWithSubqueries { + case cteDef: CTERelationDef => cteDefs(cteDef.id) = cteDef + case _ => + } + cteDefs.values.filter(_.materialized.contains(true)).foreach { cteDef => + checkMaterializedCTE(cteDef, cteDefs) + } + } + } + + private def checkMaterializedCTE( + cteDef: CTERelationDef, + cteDefs: collection.Map[Long, CTERelationDef]): Unit = { + val inlinedDefs = collectInlinedDefs(cteDef, cteDefs) + // The attributes produced within the materialized boundary, including those of nested + // subqueries, as a correlation to any of them does not cross the boundary. Unresolved + // operators are skipped: they may not have an output, and are reported by the analysis + // checks that follow. + val internalAttrs = AttributeSet(inlinedDefs.flatMap( Review Comment: **Finding 1.** `internalAttrs` is built over every def in `inlinedDefs`, so it also holds the *output* attributes of the CTEs the materialized definition references. Those are not private to the boundary: `CTESubstitution.resolveWithCTERelations` builds every `CTERelationRef` with `d.output`, and `DeduplicateRelations` renames only the conflicting occurrence, so one reference keeps the definition's own ids. A correlation from the materialized definition to a CTE reference in the enclosing query therefore looks internal. ```sql WITH s AS (SELECT c1, c2 FROM t) SELECT * FROM s o WHERE EXISTS ( WITH v AS MATERIALIZED (SELECT i.c1 FROM s i WHERE i.c1 = o.c1) SELECT * FROM v ) ``` On this head with the fixed-point analyzer, analysis passes and `v` keeps `Filter (c1#27 = outer(c1#9))` with `c1#9` coming from the outer `s` reference. The query then fails in the optimizer with ``` org.apache.spark.SparkUnsupportedOperationException: Decorrelate inner query through WithCTE is not supported. ``` The single-pass resolver raises `UNSUPPORTED_FEATURE.MATERIALIZED_CTE_WITH_OUTER_REFERENCE` for the same text, because it assigns fresh ids per CTE reference. So the two analyzers do not agree here, contrary to the description, and under `spark.sql.analyzer.singlePassResolver.dualRunWithLegacy` (default `Utils.isTesting`, so on in every test run) the query fails with `HYBRID_ANALYZER_EXCEPTION.SINGLE_PASS_FAILED_FIXED_POINT_SUCCEEDED`. Narrowing the set to `cteDef.child`, as suggested on the earlier thread, does not work either. I measured it: it then rejects ```sql WITH v1 AS (SELECT a.c1 AS k, (WITH u AS (SELECT max(c2) m FROM t2 WHERE t2.c1 = a.c1) SELECT m FROM u) AS m FROM t a), v2 AS MATERIALIZED (SELECT * FROM v1) SELECT * FROM v2 ``` which is the shape the widening was for. An attribute-id set cannot separate the two cases, because the same id is both produced inside the boundary and visible outside it. What does work is closing the boundary before the scan, so a correlation kept inside a subquery stays inside it. Then no attribute set is needed at all: every `OuterReference` still sitting at the operator level crosses the boundary. ```scala val boundary = InlineCTE(alwaysInline = true, isAnalysis = true) .apply(WithCTE(cteDef.child, inlinedDefs.filterNot(_ eq cteDef))) boundary.foreach(_.expressions.foreach(_.foreach { case o: OuterReference => throw QueryCompilationErrors.materializedCTEWithOuterReferenceError(o) case s: SubqueryExpression if s.outerScopeAttrs.nonEmpty => s.outerScopeAttrs.flatMap(_.collect { case r: OuterScopeReference => r }) .headOption .foreach(r => throw QueryCompilationErrors.materializedCTEWithOuterReferenceError(r)) case _ => })) ``` With that in place both queries above behave: the first is rejected on both analyzers, the second still runs, and `CTEInlineSuite` is green in both AQE modes. Reaching for an optimizer rule from an analysis check is a little unusual, though `CheckAnalysis` right below already does it. Whichever fix you take, please add the shared-CTE shape as a negative test. Every current negative correlates to a plain table in the enclosing query, so none of them exercises the case where the outer attribute id is also produced inside the boundary. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala: ########## @@ -316,6 +316,9 @@ trait CheckAnalysis extends LookupCatalog with QueryErrorsBase with PlanToString } def checkAnalysis(plan: LogicalPlan): Unit = { + // Check MATERIALIZED CTE relations before inlining, as `InlineCTE` below inlines them to + // restore the original plan shape. + MaterializedCTECheck(plan) Review Comment: **Finding 3.** This runs before `checkAnalysis0` reports unresolved operators and attributes, so it can preempt the more fundamental error. ```sql SELECT * FROM t o WHERE EXISTS ( WITH v AS MATERIALIZED (SELECT no_such_col FROM t2 i WHERE i.c1 = o.c1) SELECT * FROM v ) ``` The fixed-point analyzer reports `UNSUPPORTED_FEATURE.MATERIALIZED_CTE_WITH_OUTER_REFERENCE` naming `c1`. The single-pass resolver reports `UNRESOLVED_COLUMN.WITH_SUGGESTION` for `no_such_col`. Same query, two errors, and the useful one is the second -- `no_such_col` is the actual mistake. The pre-inline position is the constraint, so the call has to stay above `InlineCTE`. Skipping it when the plan is not resolved would let the resolution checks report first; I have not tried that, so it needs checking against the analysis-only commands that reach `checkAnalysis` before their body is resolved. -- 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]
