pan3793 commented on code in PR #58661: URL: https://github.com/apache/spark/pull/58661#discussion_r3976568335
########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/MaterializedCTECheck.scala: ########## @@ -0,0 +1,71 @@ +/* + * 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.{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, neither directly nor + * through a subquery with an outer-scope reference, as it is evaluated once on its own. The CTEs + * it references are checked as well, since they are inlined into it unless they are MATERIALIZED + * themselves. 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 _ => + } + val checked = mutable.HashSet.empty[Long] + cteDefs.values.filter(_.materialized.contains(true)).foreach { cteDef => + checkCTERelationDef(cteDef, cteDefs, checked) + } + } + } + + private def checkCTERelationDef( + cteDef: CTERelationDef, + cteDefs: collection.Map[Long, CTERelationDef], + checked: mutable.Set[Long]): Unit = { + if (checked.add(cteDef.id)) { + cteDef.child.foreach(_.expressions.foreach(_.foreach { + case o: OuterReference => Review Comment: Fixed in 7e742bb73a4. The check now collects the attributes produced inside the materialized boundary (the definition and the definitions it references transitively, subqueries included) and rejects an outer reference only when its target is outside that set. Both shapes above are added as passing cases in `CTEInlineSuite` and `cte.sql`, plus a unit case in `AnalysisErrorSuite`. All existing error cases still fail, as they correlate to the enclosing query. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala: ########## @@ -77,15 +81,18 @@ case class InlineCTE( private def shouldInline(cteDef: CTERelationDef, refCount: Int): Boolean = { // A CTE definition that requests to skip inlining is never inlined, even in `alwaysInline` // mode, so that a producer can guarantee the CTE is materialized rather than duplicated. - !cteDef.forceSkipInline && (alwaysInline || { - // We do not need to check enclosed `CTERelationRef`s for `deterministic` or - // `OuterReference`, because: - // 1) It is fine to inline a CTE if it references another CTE that is non-deterministic; - // 2) Any `CTERelationRef` that contains `OuterReference` would have been inlined first. - refCount == 1 || - cteDef.deterministic || - cteDef.child.exists(_.expressions.exists(_.isInstanceOf[OuterReference])) - }) + !cteDef.forceSkipInline && (alwaysInline || (cteDef.materialized match { + // The user-specified MATERIALIZED / NOT MATERIALIZED option overrides the default decision. + case Some(materialized) => !materialized Review Comment: Kept as is, documented in 7e742bb73a4. The default paths guard determinism, agreed, but Spark already evaluates a non-deterministic CTE per reference when asked to inline: `spark.sql.legacy.inlineCTEInCommands` inlines every CTE in a command, and `CTESubstitution` does the same by default for multi-insert and for commands that are not `CTEInChildren`, without a determinism guard. `NOT MATERIALIZED` is the same opt-in, made per CTE in the query text. PG's volatile guard exists for side effects (`nextval()`, data-modifying CTEs), which Spark SQL does not have; Spark's `deterministic` is about cross-reference consistency, which is what the user waives here. A uniform rule (`MATERIALIZED` evaluates once, `NOT MATERIALIZED` inlines) avoids a third state where the option is silently ignored. The divergence is stated in the docs and in the PR description's planning notes. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/MaterializedCTECheck.scala: ########## @@ -0,0 +1,71 @@ +/* + * 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.{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, neither directly nor + * through a subquery with an outer-scope reference, as it is evaluated once on its own. The CTEs + * it references are checked as well, since they are inlined into it unless they are MATERIALIZED + * themselves. 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 _ => + } + val checked = mutable.HashSet.empty[Long] + cteDefs.values.filter(_.materialized.contains(true)).foreach { cteDef => + checkCTERelationDef(cteDef, cteDefs, checked) + } + } + } + + private def checkCTERelationDef( + cteDef: CTERelationDef, + cteDefs: collection.Map[Long, CTERelationDef], + checked: mutable.Set[Long]): Unit = { + if (checked.add(cteDef.id)) { + cteDef.child.foreach(_.expressions.foreach(_.foreach { + case o: OuterReference => + throw QueryCompilationErrors.materializedCTEWithOuterReferenceError(o) + case s: SubqueryExpression if s.outerScopeAttrs.nonEmpty => Review Comment: Kept as defense in depth with a comment on the causality in 7e742bb73a4. The scan still does not descend into subquery plans. ########## docs/sql-ref-syntax-qry-select-cte.md: ########## @@ -40,6 +40,15 @@ expression_name [ ( column_name [ , ... ] ) ] [ AS ] ( query ) Specifies a name for the common table expression. +* **MATERIALIZED**, **NOT MATERIALIZED** + + Optionally specifies how the common table expression is evaluated. `MATERIALIZED` forces it Review Comment: The sql-ref pages describe observable behavior only; even the hints page says what Spark does with a hint without naming rules, plan nodes, or configs. The fence and shuffle-cost points are planning details with no result difference, and the ignored paths can only be named in implementation terms, so those are in the PR description's planning notes. The unreferenced case is already covered by the sentence that a MATERIALIZED CTE cannot reference an outer query column. Added the one observable point in 7e742bb73a4: under NOT MATERIALIZED each reference evaluates non-deterministic expressions independently. ########## sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4: ########## @@ -658,7 +658,7 @@ ctes ; namedQuery - : name=errorCapturingIdentifier (columnAliases=identifierList)? (MAX RECURSION LEVEL integerValue)? AS? LEFT_PAREN query RIGHT_PAREN + : name=errorCapturingIdentifier (columnAliases=identifierList)? (MAX RECURSION LEVEL integerValue)? AS? (NOT? MATERIALIZED)? LEFT_PAREN query RIGHT_PAREN Review Comment: Kept uniform: `NOT MATERIALIZED` inlines a recursive definition at each reference, which is also what the default does for a deterministic recursive body. Stated as a divergence from PG in the PR description's planning notes. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/MaterializedCTECheck.scala: ########## @@ -0,0 +1,71 @@ +/* + * 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.{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, neither directly nor + * through a subquery with an outer-scope reference, as it is evaluated once on its own. The CTEs + * it references are checked as well, since they are inlined into it unless they are MATERIALIZED Review Comment: Reworded in 7e742bb73a4 along with the boundary fix. -- 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]
