ulysses-you commented on code in PR #58661: URL: https://github.com/apache/spark/pull/58661#discussion_r3975884109
########## 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: This scan (and the ref-following at L65-67) treats every `OuterReference` in the materialized def, or in any def reachable through a `CTERelationRef`, as a boundary violation. But `OuterReference` is scope-relative: a CTE def nested inside the MATERIALIZED body wraps correlations to the MATERIALIZED CTE's own relations in `OuterReference`, so a fully self-contained CTE is rejected: WITH v AS MATERIALIZED ( SELECT a.c1, (WITH u AS (SELECT max(c2) m FROM t2 WHERE t2.c1 = a.c1) SELECT m FROM u) AS mm FROM t a) SELECT * FROM v ORDER BY c1 Observed: plain `AS (` returns rows; `AS MATERIALIZED (` fails with `UNSUPPORTED_FEATURE.MATERIALIZED_CTE_WITH_OUTER_REFERENCE` naming column `c1` - v's own column - on both the legacy and single-pass resolver. The `LATERAL (WITH ...)` shape fails too, via this scan directly (`WithCTE.children = cteDefs :+ plan`, so `cteDef.child.foreach` descends into plan-level nested defs). The peer scan `InlineCTE.validateNoOuterReferencesAcrossCTEBoundary` (InlineCTE.scala:98-124) is safe only because it runs on internally produced `forceSkipInline` defs (SPARK-58006) whose contents are producer-controlled; user-written MATERIALIZED bodies don't carry that precondition. Suggestion: reject a reference only when its target is not produced within the materialized def's subtree (collect the attribute set over `cteDef.child` with `foreachWithSubqueries`), and evaluate followed defs against the materialized boundary rather than their own. All existing error tests target enclosing-main-query columns, so they remain rejected. Please also add the two shapes above as negatives: every current error test correlates to the enclosing query, so the suite cannot distinguish "crosses the materialization boundary" from "is an `OuterReference`". ########## 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: PG also gates inlining on `!cte->cterecursive`: a recursive CTE is never inlined, even under `NOT MATERIALIZED`. Spark currently honors the option for recursive defs - it works and results are correct (verified end-to-end, `ResolveWithCTE` preserves the flag via `copy`) - but if the determinism guard in InlineCTE.scala:86 is adopted, this is the remaining parity gap to decide explicitly: ignore `NOT MATERIALIZED` on `RECURSIVE` like PG, or keep honoring it and document the divergence. ########## 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: `NOT MATERIALIZED` currently overrides the determinism guard: a non-deterministic multi-reference CTE is inlined and each reference evaluates independently. PostgreSQL does not do this - in `SS_process_ctes` (subselect.c), `!contain_volatile_functions` is a conjunct of the whole inline condition, including `CTEMaterializeNever`: if ((cte->ctematerialized == CTEMaterializeNever || (cte->ctematerialized == CTEMaterializeDefault && cte->cterefcount == 1)) && !cte->cterecursive && cmdType == CMD_SELECT && !contain_dml(cte->ctequery) && (cte->cterefcount <= 1 || !contain_outer_selfref(cte->ctequery)) && !contain_volatile_functions(cte->ctequery)) inline_cte(root, cte); Spark's own precedence points the same way: `CollapseProject` checks `!a.child.deterministic` before `alwaysInline` (Optimizer.scala:1496), the `None` branch below is the same guard, and `forceSkipInline` sits above everything. Observable divergence: `WITH v AS NOT MATERIALIZED (SELECT rand() r) SELECT * FROM v a JOIN v b ON a.r = b.r` draws per reference (join nearly always empty), while the default evaluates once and matches rows. Suggestion: `case Some(false) => refCount == 1 || cteDef.deterministic || <the outerRef condition>` - keep the guard in force. Outcomes then align with PG class by class. Note the corollary: in Spark the option becomes observably equivalent to omitting it (the default already inlines deterministic multi-ref CTEs - the class PG's keyword exists for), so its value is SQL-text portability and explicit intent; worth saying so in the description/docs. This flips `InlineCTESuite.scala:124` and `CTEInlineSuite.scala:1002`; suggest adding a pin that the optimized plan is identical with and without the keyword. ########## 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: This branch is end-to-end unreachable: a correlation crossing the CTE-def boundary from inside a nested subquery fails resolution first (`UNRESOLVED_COLUMN` for plain / NOT MATERIALIZED / MATERIALIZED alike, on both resolvers), so only the hand-built `AnalysisErrorSuite` plans exercise it. Fine as defense-in-depth (parity with InlineCTE.scala:108-123), but per usual practice for unreachable defensive branches, please add a comment stating that causality. Conversely, please don't extend the scan into subquery plans: a two-level inner correlation (an `OuterScopeReference` held by a subquery inside another subquery, targeting a column internal to the def) is legitimate, and the current non-descending walk handles it correctly. ########## common/utils/src/main/resources/error/error-conditions.json: ########## @@ -9090,6 +9090,11 @@ "Literal for '<value>' of <type>." ] }, + "MATERIALIZED_CTE_WITH_OUTER_REFERENCE" : { Review Comment: Nit (optional, depends on the fix shape for the boundary issue): "MATERIALIZED CTE referencing the outer query column <colName>" currently fires with <colName> naming the CTE's own column in the false-rejection cases. If the message survives the fix unchanged, consider "... column <colName> defined outside the CTE". ########## 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: This paragraph misses a few verified caveats: - `MATERIALIZED` is not an optimization fence: per-reference predicates are OR-merged and pushed into the shared definition (this PR's own explain golden pushes `Or(key > 20, key < 15)` into the scan). PG's materialized CTE is "evaluated as written". - The option is silently ignored in multi-insert, non-`CTEInChildren` commands, and under the legacy inline/precedence configs (verified: analyzed plan keeps no `WithCTE` there). This is only stated in the PR description. - A correlated `MATERIALIZED` CTE is rejected even when it is never referenced. - A single-reference `MATERIALIZED` CTE pays the RoundRobin shuffle boundary with no reuse benefit. ########## 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: Nit: the scaladoc says referenced CTEs are checked "unless they are MATERIALIZED themselves", but the code follows and checks referenced defs regardless of their own flag (harmless - a MATERIALIZED ref is checked on its own anyway). Please align the wording when addressing the boundary issue. ########## sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisErrorSuite.scala: ########## @@ -1330,6 +1331,35 @@ class AnalysisErrorSuite extends AnalysisTest with DataTypeErrorsBase { "expr" -> "\"_w0\"", "exprType" -> "\"MAP<STRING, STRING>\"")) + test("MATERIALIZED CTE cannot reference the outer query") { + val relation = LocalRelation($"a".int) + def cteRef(cteDef: CTERelationDef): CTERelationRef = { + CTERelationRef(cteDef.id, cteDef.resolved, cteDef.output, cteDef.isStreaming) + } + def assertMaterializedCTEError( + cteChild: LogicalPlan, + colName: String, + referencedCTEDefs: Seq[CTERelationDef] = Nil): Unit = { + val cteDef = CTERelationDef(cteChild, materialized = Some(true)) + val plan = WithCTE(cteRef(cteDef).select(colName), referencedCTEDefs :+ cteDef) + checkError( + exception = intercept[AnalysisException](getAnalyzer.checkAnalysis(plan)), + condition = "UNSUPPORTED_FEATURE.MATERIALIZED_CTE_WITH_OUTER_REFERENCE", + parameters = Map("colName" -> "`a`")) Review Comment: Nit: the helper takes a per-case `colName` (used for the referencing projection) but the expected parameter is hardcoded to "`a`" - correct for all three cases today, brittle if a case with a different correlated column is added. -- 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]
