spark-test commented on code in PR #58229: URL: https://github.com/apache/spark/pull/58229#discussion_r4481930120
########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/ConvertViewToMaterializedCTE.scala: ########## @@ -0,0 +1,297 @@ +/* + * 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.optimizer + +import scala.collection.mutable + +import org.apache.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute} +import org.apache.spark.sql.catalyst.plans.logical._ +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.internal.SQLConf + +/** + * Rewrites multiple references to the same view into a single `CTERelationDef` with multiple + * `CTERelationRef`s, so that the view's underlying plan is computed once (through exchange + * reuse at the physical layer) instead of once per reference. + * + * The rule runs first in `FinishAnalysis`, before `EliminateResolvedHint`, + * `EliminateSubqueryAliases` and `EliminatePipeOperators` strip analysis-only nodes from + * the plan. Those rules are applied to the main plan before its subquery plans, so a + * definition body taken from a first-visited subquery occurrence may still hold such + * nodes; running the conversion first lets the cleanup rules reach the body at the top + * level, wherever it was taken from. `EliminateView` runs after the cleanup rules: no + * `View` nodes remain and every reference site holds an independent copy of the view's + * plan. + * + * A join hint written at the top of a view body binds to the join(s) the view takes part + * in (see `SubqueryHintPropagationSuite`): `EliminateResolvedHint` extracts hints from + * Join children by walking down unary nodes, and processes the body inline today so the + * hint reaches the consuming join. After this rule the body lives inside a `CTERelationDef` + * the walk cannot reach, so the rule strips those top-of-body hints itself and re-attaches + * them above every reference site. Hints deeper in the body bind to the body's own joins + * and are left in place. + * + * A converted definition keeps `forceSkipInline = true` only when at least two references + * survive the rewrite; otherwise `InlineCTE` would immediately flatten it back into + * duplicated subtrees (the definition body is deterministic in every case we convert), + * making the rule a no-op. Definitions with fewer than two surviving references keep the + * flag off and are inlined: their occurrences do not all survive the rewrite (a later + * occurrence's body is dropped, taking references to nested converted views with it), and + * some references may sit inside unreferenced user-written CTEs - materializing either + * way would add a shuffle that exchange reuse cannot deduplicate. + * + * Only deterministic, batch views are eligible: a multi-reference CTE guarantees that its + * definition is evaluated exactly once (even for non-deterministic definitions), while + * multiple references to a non-deterministic view are evaluated independently today. + * Converting such views would change query results. + * + * A view body may contain correlated subqueries whose outer references resolve to relations + * inside the same body (e.g. `t WHERE x IN (SELECT y FROM s WHERE s.k = t.k)`). The view is + * analyzed standalone when it is created, so an outer reference that does not resolve inside + * the body fails view analysis and can never escape to the outer query. The converted + * definition contains the whole body, so internal correlations resolve within it and these + * bodies are safe to convert. `InlineCTE`'s rejection of boundary-crossing outer references + * is only a generic safety net for non-view `forceSkipInline` producers. + * + * Each reference site gains a shuffle boundary added by `ReplaceCTERefWithRepartition` and + * deduplicated by exchange reuse, so the conversion trades recomputation for a + * shuffle plus reuse; it is therefore gated behind + * [[SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE]] and off by default. + */ +object ConvertViewToMaterializedCTE extends Rule[LogicalPlan] { + + override def apply(plan: LogicalPlan): LogicalPlan = { + if (!SQLConf.get.getConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE)) return plan + // FinishAnalysis re-runs the rule batch on Subquery roots and requires the result + // to stay a Subquery; the top-level pass already covers views inside subqueries. + if (plan.isInstanceOf[Subquery]) return plan + val occurrences = plan.collectWithSubqueries { case v: View => v } + if (occurrences.length < 2) return plan + + // Admit an identifier only when every occurrence of it forms one qualifying group. + // Grouping is by identifier first because the rewrite below matches occurrences by + // identifier: an identifier carried by occurrences with divergent bodies (e.g. one + // occurrence resolved against a view definition that was replaced after another + // occurrence captured it), or whose occurrences fail qualification for any other + // reason, must not convert at all - otherwise the rewrite would rebind those + // occurrences against a definition their plan does not match, silently changing + // query results. + // + // Inner CTE definition ids are normalized before the bodies are compared: a view + // body is re-analyzed per occurrence, and the re-analysis re-substitutes the body's + // inner CTEs, minting a fresh `CTERelationDef` id each time (e.g. for a SQL view + // whose body has a WITH clause over a persistent table). The re-minted ids are the + // only difference between such occurrences' bodies, so the normalization lets them + // group together; bodies that differ in anything else still stay apart. + val qualifiedIdentifiers = occurrences + .groupBy(_.desc.identifier) + .collect { + case (identifier, occs) + if qualifies(occs) && occs.map(occ => + normalizeCteIds(occ.child).canonicalized).distinct.length == 1 => identifier + } + .toSet + if (qualifiedIdentifiers.isEmpty) return plan + + // Bottom-up rewrite: nested views are visited (and their definitions appended) before + // the views containing them, so a referenced definition always precedes its referrer + // in `cteDefs`. The first occurrence of a group creates the definition, replaced by a + // bare reference; later occurrences are wrapped in a `Project` re-minting the + // occurrence's ids from the definition output, so consumers above need no rewriting. + val cteDefs = mutable.ArrayBuffer.empty[CTERelationDef] + val defByGroup = mutable.HashMap.empty[TableIdentifier, (CTERelationDef, Seq[HintInfo])] + + val rewritten = plan.transformUpWithSubqueries { + case v: View if qualifiedIdentifiers.contains(v.desc.identifier) => + // The body's top-of-node hints must not stay in the definition, where neither + // this rule nor `EliminateResolvedHint` can hoist them into the consuming joins; + // build the definition from the hint-free body and re-apply the hints above each + // reference site instead. All occurrences share one canonicalized body, so the + // hints extracted from the first one describe every reference site uniformly. + defByGroup.get(v.desc.identifier) match { + case Some((cteDef, bodyHints)) => + // Later occurrence: re-bind the reference output to this occurrence's + // attributes positionally. The group qualification has already asserted that + // name, type and nullability align element-wise. + val ref = CTERelationRef( + cteDef.id, + _resolved = true, + output = cteDef.output, + isStreaming = false, + maxRows = cteDef.maxRows) + withBodyHints(bodyHints, + Project(rebindingProjectList(v.output, cteDef.output), ref)) + + case None => + // First occurrence: consumers above already reference this occurrence's + // expression ids, which are exactly the definition output, so the bare + // reference is output-compatible. `forceSkipInline` is decided after the + // rewrite, by the number of surviving references. + val (hintFreeBody, bodyHints) = + EliminateResolvedHint.extractHintsFromPlan(v.child) + val cteDef = CTERelationDef(hintFreeBody) + defByGroup.put(v.desc.identifier, (cteDef, bodyHints)) + cteDefs += cteDef + withBodyHints(bodyHints, CTERelationRef( + cteDef.id, + _resolved = true, + output = v.child.output, + isStreaming = false, + maxRows = cteDef.maxRows)) + } + } + + if (cteDefs.isEmpty) { + plan + } else { + // Not every occurrence survives the rewrite as a reference: the body of a later + // occurrence is dropped (taking references to nested converted views with it), and + // some references sit inside unreferenced user-written CTEs. Materializing a + // definition with fewer than two live references would add a shuffle that exchange + // reuse cannot deduplicate, so re-count after the rewrite and leave such definitions + // inlinable - InlineCTE flattens them back into their consumers. + val newDefIds = cteDefs.map(_.id).toSet + val withDefs = attachDefs(rewritten, cteDefs.toSeq) + val liveRefCounts = countLiveRefs(withDefs) + withDefs.transformUpWithSubqueries { + case d: CTERelationDef if newDefIds.contains(d.id) => + d.copy(forceSkipInline = liveRefCounts.getOrElse(d.id, 0) >= 2) + } + } + } + + // Counts, per definition id, the references reachable from the query root: a reference + // inside a definition counts only when the definition itself is live, mirroring how + // `InlineCTE` discounts references held by unreferenced definitions. Recursive CTEs are + // handled by the visited-set worklist. + private def countLiveRefs(plan: LogicalPlan): Map[Long, Int] = { + // The region of a reference is its innermost enclosing definition's id, or + // ROOT_REGION at the query root. + val ROOT_REGION = -1L + val outgoing = mutable.HashMap.empty[Long, mutable.ArrayBuffer[Long]] + + def walk(p: LogicalPlan, region: Long): Unit = { + p match { + case d: CTERelationDef => walk(d.child, d.id) + case r: CTERelationRef => + outgoing.getOrElseUpdate(region, mutable.ArrayBuffer.empty) += r.cteId + case _ => p.children.foreach(walk(_, region)) + } + p.subqueries.foreach(walk(_, region)) + } + walk(plan, ROOT_REGION) + + val liveRefCounts = mutable.HashMap.empty[Long, Int].withDefaultValue(0) + val visitedDefs = mutable.HashSet.empty[Long] + val pending = mutable.Queue.empty[Long] ++= + outgoing.getOrElse(ROOT_REGION, mutable.ArrayBuffer.empty) + while (pending.nonEmpty) { + val target = pending.dequeue() + liveRefCounts(target) += 1 + if (visitedDefs.add(target)) { + pending ++= outgoing.getOrElse(target, mutable.ArrayBuffer.empty) + } + } + liveRefCounts.toMap + } + + // Group by view identity: the rule dedupes references of the SAME view, not distinct + // views with coincidentally equal bodies, and all occurrences of one view resolve + // through the same (db-qualified) identifier. The canonicalized body comparison is a + // precondition guard in the identifier admission above: if occurrences of one view + // ever diverge structurally, we skip conversion instead of building a wrong shared + // definition. + // + // Rewrites every inner CTE definition id and reference to an appearance ordinal, so + // that bodies differing only in re-minted inner CTE ids canonicalize equal. Same tree + // shapes are visited in the same order, so equal bodies map to equal ordinals, while + // genuinely different structures (e.g. two sibling identical CTEs) still produce + // distinct ordinals and stay distinguishable. + private def normalizeCteIds(plan: LogicalPlan): LogicalPlan = { + val ids = mutable.HashMap.empty[Long, Long] + plan.transformUpWithSubqueries { + case d: CTERelationDef => d.copy(id = ids.getOrElseUpdate(d.id, ids.size)) + case r: CTERelationRef => r.copy(cteId = ids.getOrElseUpdate(r.cteId, ids.size)) + } + } + + private def qualifies(occs: Seq[View]): Boolean = { Review Comment: **Blocking (P1):** Admission only inspects each occurrence's body, but materializing the view also moves everything the query evaluates above a reference behind the per-site round-robin shuffle. `input_file_name()`, `input_file_block_start()` and `input_file_block_length()` read the file block from `InputFileBlockHolder`, which only the scanning task sets, so above a converted reference they return `''` and `-1`. For example, with the flag on and `v` defined as `SELECT a FROM parquet_tbl`, `SELECT input_file_name() AS f, a FROM v UNION ALL SELECT input_file_name(), a FROM v` passes `PreReadCheck` (one file source per branch) and returns `''` for every row. A filter such as `WHERE input_file_name() LIKE '%2024%'` above a reference is non-deterministic, so it is not pushed into the definition, and it drops every row. Lineage and audit queries that tag rows with their source file get wrong results without any error. **Recommended change:** Decline conversion for a view when input_file_name(), input_file_block_start() or input_file_block_length() is evaluated by any operator above one of its occurrences, including in subquery plans. Add end-to-end tests that use these expressions over a repeatedly referenced view. **Why this works:** Before rewriting, find every operator, in the main plan and in subqueries, that evaluates one of the three expressions, collect the view identifiers in the subtree it reads, and remove those identifiers from the admitted set. Their references stay inlined, so the expressions keep running in the task that scans the file, as on the base branch. **Scope:** Extend admission to check operators above each occurrence for file-block expressions, and add tests. **Risks:** Queries that evaluate these expressions above a repeated view no longer convert that view, so they keep the base branch's evaluation once per reference. **Constraints:** Only views with an occurrence under such an operator are declined; other views in the same query still convert. **Success:** input_file_name(), input_file_block_start() and input_file_block_length() evaluated above any reference to a repeatedly referenced view return the same values with the flag on as with it off. **Verification:** - **Regression:** With the flag on, a UNION ALL of two branches that each select input_file_name() and a column from the same parquet-backed view returns the real file paths, and a filter on input_file_name() above a reference keeps the same rows as with the flag off. - **Behavior:** A query that references the same view twice without these expressions still converts it into one shared definition. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/ConvertViewToMaterializedCTE.scala: ########## @@ -0,0 +1,297 @@ +/* + * 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.optimizer + +import scala.collection.mutable + +import org.apache.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute} +import org.apache.spark.sql.catalyst.plans.logical._ +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.internal.SQLConf + +/** + * Rewrites multiple references to the same view into a single `CTERelationDef` with multiple + * `CTERelationRef`s, so that the view's underlying plan is computed once (through exchange + * reuse at the physical layer) instead of once per reference. + * + * The rule runs first in `FinishAnalysis`, before `EliminateResolvedHint`, + * `EliminateSubqueryAliases` and `EliminatePipeOperators` strip analysis-only nodes from + * the plan. Those rules are applied to the main plan before its subquery plans, so a + * definition body taken from a first-visited subquery occurrence may still hold such + * nodes; running the conversion first lets the cleanup rules reach the body at the top + * level, wherever it was taken from. `EliminateView` runs after the cleanup rules: no + * `View` nodes remain and every reference site holds an independent copy of the view's + * plan. + * + * A join hint written at the top of a view body binds to the join(s) the view takes part + * in (see `SubqueryHintPropagationSuite`): `EliminateResolvedHint` extracts hints from + * Join children by walking down unary nodes, and processes the body inline today so the + * hint reaches the consuming join. After this rule the body lives inside a `CTERelationDef` + * the walk cannot reach, so the rule strips those top-of-body hints itself and re-attaches + * them above every reference site. Hints deeper in the body bind to the body's own joins + * and are left in place. + * + * A converted definition keeps `forceSkipInline = true` only when at least two references + * survive the rewrite; otherwise `InlineCTE` would immediately flatten it back into + * duplicated subtrees (the definition body is deterministic in every case we convert), + * making the rule a no-op. Definitions with fewer than two surviving references keep the + * flag off and are inlined: their occurrences do not all survive the rewrite (a later + * occurrence's body is dropped, taking references to nested converted views with it), and + * some references may sit inside unreferenced user-written CTEs - materializing either + * way would add a shuffle that exchange reuse cannot deduplicate. + * + * Only deterministic, batch views are eligible: a multi-reference CTE guarantees that its + * definition is evaluated exactly once (even for non-deterministic definitions), while + * multiple references to a non-deterministic view are evaluated independently today. + * Converting such views would change query results. + * + * A view body may contain correlated subqueries whose outer references resolve to relations + * inside the same body (e.g. `t WHERE x IN (SELECT y FROM s WHERE s.k = t.k)`). The view is + * analyzed standalone when it is created, so an outer reference that does not resolve inside + * the body fails view analysis and can never escape to the outer query. The converted + * definition contains the whole body, so internal correlations resolve within it and these + * bodies are safe to convert. `InlineCTE`'s rejection of boundary-crossing outer references + * is only a generic safety net for non-view `forceSkipInline` producers. + * + * Each reference site gains a shuffle boundary added by `ReplaceCTERefWithRepartition` and + * deduplicated by exchange reuse, so the conversion trades recomputation for a + * shuffle plus reuse; it is therefore gated behind + * [[SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE]] and off by default. + */ +object ConvertViewToMaterializedCTE extends Rule[LogicalPlan] { + + override def apply(plan: LogicalPlan): LogicalPlan = { + if (!SQLConf.get.getConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE)) return plan + // FinishAnalysis re-runs the rule batch on Subquery roots and requires the result + // to stay a Subquery; the top-level pass already covers views inside subqueries. + if (plan.isInstanceOf[Subquery]) return plan + val occurrences = plan.collectWithSubqueries { case v: View => v } + if (occurrences.length < 2) return plan + + // Admit an identifier only when every occurrence of it forms one qualifying group. + // Grouping is by identifier first because the rewrite below matches occurrences by + // identifier: an identifier carried by occurrences with divergent bodies (e.g. one + // occurrence resolved against a view definition that was replaced after another + // occurrence captured it), or whose occurrences fail qualification for any other + // reason, must not convert at all - otherwise the rewrite would rebind those + // occurrences against a definition their plan does not match, silently changing + // query results. + // + // Inner CTE definition ids are normalized before the bodies are compared: a view + // body is re-analyzed per occurrence, and the re-analysis re-substitutes the body's + // inner CTEs, minting a fresh `CTERelationDef` id each time (e.g. for a SQL view + // whose body has a WITH clause over a persistent table). The re-minted ids are the + // only difference between such occurrences' bodies, so the normalization lets them + // group together; bodies that differ in anything else still stay apart. + val qualifiedIdentifiers = occurrences + .groupBy(_.desc.identifier) + .collect { + case (identifier, occs) + if qualifies(occs) && occs.map(occ => + normalizeCteIds(occ.child).canonicalized).distinct.length == 1 => identifier + } + .toSet + if (qualifiedIdentifiers.isEmpty) return plan + + // Bottom-up rewrite: nested views are visited (and their definitions appended) before + // the views containing them, so a referenced definition always precedes its referrer + // in `cteDefs`. The first occurrence of a group creates the definition, replaced by a + // bare reference; later occurrences are wrapped in a `Project` re-minting the + // occurrence's ids from the definition output, so consumers above need no rewriting. + val cteDefs = mutable.ArrayBuffer.empty[CTERelationDef] + val defByGroup = mutable.HashMap.empty[TableIdentifier, (CTERelationDef, Seq[HintInfo])] + + val rewritten = plan.transformUpWithSubqueries { + case v: View if qualifiedIdentifiers.contains(v.desc.identifier) => + // The body's top-of-node hints must not stay in the definition, where neither + // this rule nor `EliminateResolvedHint` can hoist them into the consuming joins; + // build the definition from the hint-free body and re-apply the hints above each + // reference site instead. All occurrences share one canonicalized body, so the + // hints extracted from the first one describe every reference site uniformly. + defByGroup.get(v.desc.identifier) match { + case Some((cteDef, bodyHints)) => + // Later occurrence: re-bind the reference output to this occurrence's + // attributes positionally. The group qualification has already asserted that + // name, type and nullability align element-wise. + val ref = CTERelationRef( + cteDef.id, + _resolved = true, + output = cteDef.output, + isStreaming = false, + maxRows = cteDef.maxRows) + withBodyHints(bodyHints, + Project(rebindingProjectList(v.output, cteDef.output), ref)) Review Comment: **Blocking (P1):** With the flag on, `SELECT a, (SELECT sum(d) FROM vr WHERE a = c) sum_d FROM l l1 GROUP BY 1, 2` over a persistent view `vr` (the SPARK-34269 test in `SubquerySuite`) stops working. `GROUP BY 2` resolves to the same scalar subquery as the select list, so both copies hold an occurrence of `vr`. The rewrite visits the grouping copy first and gives it a bare `CTERelationRef`, while the select-list copy gets this `Project` over the reference. Because `ScalarSubquery` canonicalization includes its plan, the two copies are no longer semantically equal, and `PullOutGroupingExpressions` pulls out only the grouping copy. The select-list copy stays behind as a correlated scalar subquery that is not part of the grouping, which plan-change validation rejects (`SCALAR_SUBQUERY_IS_IN_GROUP_BY_OR_AGGREGATE_FUNCTION`); without validation, `RewriteCorrelatedScalarSubquery` leaves its result neither grouped nor aggregated. Any query in which the analyzer duplicates an expression c ontaining a converted view, and a later rule matches the copies, is exposed the same way. **Recommended change:** Represent every occurrence of a converted view the same way. Replace each occurrence with a CTERelationRef to the shared definition whose output is that occurrence's own output, wrapped in that occurrence's re-attached hints, and drop the rebinding Project for later occurrences. Add plan and end-to-end tests that group by a correlated scalar subquery over a converted view. **Why this works:** With one reference shape per occurrence, copies of an expression that held equal view occurrences canonicalize equal again, because a CTE reference's top-level output attributes are normalized during canonicalization. PullOutGroupingExpressions therefore replaces the select-list copy, and the correlated subquery is grouped as on the base branch. The downstream CTE rules already map each reference's output to the definition's by position, so every reference site keeps its own attribute ids. **Scope:** Use one reference shape for every view occurrence, and cover grouping by a correlated subquery over a converted view. **Compatibility:** Each converted reference site keeps exposing the attribute ids, names, qualifiers and metadata that its consumers already reference. **Risks:** Every reference except the first now receives a de-duplicated copy of the definition plan when references are replaced by repartitions. The copies stay canonically identical, so exchange reuse still shares the shuffle; the reused-exchange test should keep passing. **Constraints:** Every occurrence keeps its names, qualifiers, metadata and nullability; the existing positional schema check keeps guaranteeing type and nullability agreement. **Success:** Grouping by a correlated scalar subquery over a converted view returns the same rows with the flag on as with it off. Every reference to a converted view is a CTE reference whose output is that occurrence's own output, with no rebinding Project. **Verification:** - **Regression:** With the flag on, a query that groups by and selects the same correlated scalar subquery over a qualifying persistent view, both by GROUP BY ordinal and by the repeated expression, returns the same rows as with the flag off and passes plan-change validation. - **Behavior:** Converted self-joins, references inside scalar subqueries and nested views still produce one shared, reused shuffle and the same rows as with the flag off. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/ConvertViewToMaterializedCTE.scala: ########## @@ -0,0 +1,297 @@ +/* + * 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.optimizer + +import scala.collection.mutable + +import org.apache.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute} +import org.apache.spark.sql.catalyst.plans.logical._ +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.internal.SQLConf + +/** + * Rewrites multiple references to the same view into a single `CTERelationDef` with multiple + * `CTERelationRef`s, so that the view's underlying plan is computed once (through exchange + * reuse at the physical layer) instead of once per reference. + * + * The rule runs first in `FinishAnalysis`, before `EliminateResolvedHint`, + * `EliminateSubqueryAliases` and `EliminatePipeOperators` strip analysis-only nodes from + * the plan. Those rules are applied to the main plan before its subquery plans, so a + * definition body taken from a first-visited subquery occurrence may still hold such + * nodes; running the conversion first lets the cleanup rules reach the body at the top + * level, wherever it was taken from. `EliminateView` runs after the cleanup rules: no + * `View` nodes remain and every reference site holds an independent copy of the view's + * plan. + * + * A join hint written at the top of a view body binds to the join(s) the view takes part + * in (see `SubqueryHintPropagationSuite`): `EliminateResolvedHint` extracts hints from + * Join children by walking down unary nodes, and processes the body inline today so the + * hint reaches the consuming join. After this rule the body lives inside a `CTERelationDef` + * the walk cannot reach, so the rule strips those top-of-body hints itself and re-attaches + * them above every reference site. Hints deeper in the body bind to the body's own joins + * and are left in place. + * + * A converted definition keeps `forceSkipInline = true` only when at least two references + * survive the rewrite; otherwise `InlineCTE` would immediately flatten it back into + * duplicated subtrees (the definition body is deterministic in every case we convert), + * making the rule a no-op. Definitions with fewer than two surviving references keep the + * flag off and are inlined: their occurrences do not all survive the rewrite (a later + * occurrence's body is dropped, taking references to nested converted views with it), and + * some references may sit inside unreferenced user-written CTEs - materializing either + * way would add a shuffle that exchange reuse cannot deduplicate. + * + * Only deterministic, batch views are eligible: a multi-reference CTE guarantees that its + * definition is evaluated exactly once (even for non-deterministic definitions), while + * multiple references to a non-deterministic view are evaluated independently today. + * Converting such views would change query results. + * + * A view body may contain correlated subqueries whose outer references resolve to relations + * inside the same body (e.g. `t WHERE x IN (SELECT y FROM s WHERE s.k = t.k)`). The view is + * analyzed standalone when it is created, so an outer reference that does not resolve inside + * the body fails view analysis and can never escape to the outer query. The converted + * definition contains the whole body, so internal correlations resolve within it and these + * bodies are safe to convert. `InlineCTE`'s rejection of boundary-crossing outer references + * is only a generic safety net for non-view `forceSkipInline` producers. + * + * Each reference site gains a shuffle boundary added by `ReplaceCTERefWithRepartition` and + * deduplicated by exchange reuse, so the conversion trades recomputation for a + * shuffle plus reuse; it is therefore gated behind + * [[SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE]] and off by default. + */ +object ConvertViewToMaterializedCTE extends Rule[LogicalPlan] { + + override def apply(plan: LogicalPlan): LogicalPlan = { + if (!SQLConf.get.getConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE)) return plan Review Comment: **Non-blocking (P2):** The rule only checks its own flag, but its only benefit comes from exchange reuse: `ReuseExchangeAndSubquery` and AQE stage reuse are what collapse the per-site round-robin shuffles, and both are skipped when `spark.sql.exchange.reuse` is false. With exchange reuse disabled and this flag on, each reference still evaluates the whole view body and now also shuffles it, so every converted query does strictly more work, and the config doc's "computed once (through exchange reuse)" no longer holds. `MergeSubplans` returns early when subquery reuse is off for the same reason; returning the plan unchanged here when `conf.exchangeReuseEnabled` is false, and saying so in the config doc, would match it. **Verification:** - **Regression:** With exchange reuse disabled and the flag on, a self-joined view produces the same plan, with no added repartition, as with the flag off. ########## sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ConvertViewToMaterializedCTESuite.scala: ########## @@ -0,0 +1,670 @@ +/* + * 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.optimizer + +import org.apache.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType} +import org.apache.spark.sql.catalyst.dsl.expressions._ +import org.apache.spark.sql.catalyst.dsl.plans._ +import org.apache.spark.sql.catalyst.expressions.{AttributeReference, ExprId, In, ListQuery, Literal, NamedExpression, OuterReference, ScalarSubquery} +import org.apache.spark.sql.catalyst.plans.Inner +import org.apache.spark.sql.catalyst.plans.PlanTest +import org.apache.spark.sql.catalyst.plans.logical._ +import org.apache.spark.sql.catalyst.rules.RuleExecutor +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.{IntegerType, StructType} + +class ConvertViewToMaterializedCTESuite extends PlanTest { + + object Optimize extends RuleExecutor[LogicalPlan] { + val batches = + Batch("Convert View to Materialized CTE", FixedPoint(1), ConvertViewToMaterializedCTE) :: Nil + } + + object OptimizeWithInlineCTE extends RuleExecutor[LogicalPlan] { + val batches = + Batch("Convert View to Materialized CTE", FixedPoint(1), ConvertViewToMaterializedCTE) :: + Batch("Inline CTE", FixedPoint(1), InlineCTE()) :: Nil + } + + private def attr(name: String, id: Long): AttributeReference = + AttributeReference(name, IntegerType, nullable = true)(exprId = ExprId(id)) + + private def viewDesc(name: String, schema: StructType): CatalogTable = + CatalogTable( + identifier = TableIdentifier(name), + tableType = CatalogTableType.VIEW, + storage = CatalogStorageFormat.empty, + schema = schema) + + private def tempView(name: String, child: LogicalPlan): View = + View(viewDesc(name, child.schema), isTempView = true, child) + + /** + * One view `name` referenced twice, as the analyzer produces it: two `View` occurrences + * whose bodies are built from the same base attributes, except that the second + * occurrence's attributes carry renewed expression ids. The renewed attributes are + * returned as well for tests that reference them (e.g. in a join condition). Calling + * this instead of writing `tempView("v", ...)` twice makes it unambiguous that the pair + * stands for the same view used twice, not two views that happen to share a name. + */ + private def sameViewTwice( + name: String, + base: AttributeReference, + make: AttributeReference => LogicalPlan): (View, View, AttributeReference) = { + val renewed = base.withExprId(NamedExpression.newExprId) + (tempView(name, make(base)), tempView(name, make(renewed)), renewed) + } + + private def sameViewTwice( + name: String, + base: Seq[AttributeReference], + make: Seq[AttributeReference] => LogicalPlan): (View, View, Seq[AttributeReference]) = { + val renewed = base.map(_.withExprId(NamedExpression.newExprId)) + (tempView(name, make(base)), tempView(name, make(renewed)), renewed) + } + + // Same as above, for bodies that mint their own occurrence-specific ids per call + // (e.g. through `NamedExpression.newExprId`), so there is no base attribute to renew. + private def sameViewTwice(name: String, make: () => LogicalPlan): (View, View) = { + (tempView(name, make()), tempView(name, make())) + } + + // A simple deterministic view body: LocalRelation [a, b] filtered on a. + private def simpleBody(a: AttributeReference, b: AttributeReference): LogicalPlan = + LocalRelation(Seq(a, b)).where(a > 10) + + // A view body that contains a surviving (multi-ref) inner CTE. The references carry + // distinct expression ids, mirroring analyzer output. The inner CTE survives inlining + // either because it is non-deterministic or because it sets forceSkipInline. + private def nestedCteBody(defId: Long, deterministic: Boolean): LogicalPlan = { + val innerProject = + if (deterministic) OneRowRelation().select(Literal(1).as("r")) + else OneRowRelation().select(rand(0).as("r")) + val innerDef = CTERelationDef( + innerProject, + id = defId, + forceSkipInline = deterministic) + def mkRef(): CTERelationRef = CTERelationRef( + defId, + _resolved = true, + output = innerDef.output.map(_.withExprId(NamedExpression.newExprId)), + isStreaming = false) + WithCTE(Except(mkRef(), mkRef(), isAll = true), Seq(innerDef)) + } + + test("converts a self-joined view into one CTE definition and two references") { + withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") { + val base = Seq(attr("a", 100), attr("b", 101)) + val (v1, v2, renewed) = sameViewTwice( + "v", base, (as: Seq[AttributeReference]) => simpleBody(as(0), as(1))) + val query = Join(v1, v2, Inner, Some(base(0) === renewed(0)), JoinHint(None, None)) + + val optimized = Optimize.execute(query) + + // The root is wrapped in a WithCTE carrying exactly one definition. + val WithCTE(mainPlan, cteDefs) = optimized + assert(cteDefs.length == 1) + val cteDef = cteDefs.head + assert(cteDef.forceSkipInline, + "converted CTE must set forceSkipInline so InlineCTE keeps it materialized") + assert(cteDef.child.canonicalized == simpleBody(base(0), base(1)).canonicalized) + + // Exactly two references in the main plan. + val refs = mainPlan.collect { case r: CTERelationRef => r } + assert(refs.length == 2) + + // The first reference adopts the definition output directly. + assert(refs.exists(_.output.map(_.exprId) == cteDef.output.map(_.exprId))) + + // The second reference is re-bound through an aliasing Project that re-mints the + // original expression ids of that occurrence, so consumers need no rewriting. + val projects = mainPlan.collect { + case p @ Project(_, _: CTERelationRef) => p + } + assert(projects.length == 1) + assert(projects.head.output.map(_.name) == Seq("a", "b")) + assert(projects.head.output.map(_.exprId) == renewed.map(_.exprId)) + + // The join condition still references the original attributes of both occurrences. + val join = mainPlan.collect { case j: Join => j }.head + assert(join.condition.get == (base(0) === renewed(0))) + + // The query output schema is unchanged. + assert(optimized.output == query.output) + } + } + + test("moves view-body join hints above the references") { + withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") { + // A join hint at the top of the view body binds to the join(s) the view takes part + // in (SPARK-40999): after this rule replaces the View with a CTERelationRef, which + // is a leaf, the following EliminateResolvedHint can no longer hoist the hint + // through the tree. The definition is built from the hint-free body, and the + // extracted hints are re-attached above each reference so that hint hoisting keeps + // reaching the outer joins as before conversion. + val broadcastHint = HintInfo(Some(BROADCAST)) + val base = Seq(attr("a", 100), attr("b", 101)) + val (v1, v2, renewed) = sameViewTwice("v", base, (as: Seq[AttributeReference]) => + ResolvedHint(simpleBody(as.head, as(1)), broadcastHint)) + val query = Join(v1, v2, Inner, Some(base(0) === renewed(0)), JoinHint(None, None)) + + val optimized = Optimize.execute(query) + + val WithCTE(mainPlan, cteDefs) = optimized + assert(cteDefs.length == 1) + // The definition must be built from the hint-free body. + assert(cteDefs.head.child.collect { case h: ResolvedHint => h }.isEmpty, + s"definition body still carries the hint: ${cteDefs.head.child}") + + // Each reference carries the body's hint above it: the first occurrence wraps the + // bare reference, the later occurrence wraps its rebinding Project. + val hinted = mainPlan.collect { + case h @ ResolvedHint(_, hints) => (h, hints) + } + assert(hinted.length == 2, s"expected two hinted references in: $mainPlan") + assert(hinted.forall { case (_, hints) => hints == broadcastHint }) + assert(hinted.count { case (h, _) => h.child.isInstanceOf[CTERelationRef] } == 1, + s"first reference must be hinted directly: $mainPlan") + assert(hinted.count { case (h, _) => h.child.isInstanceOf[Project] } == 1, + s"rebinding reference must be hinted above its Project: $mainPlan") + + // The rebinding occurrence's identity is preserved under its hint. + val bindingProject = hinted.collect { case (ResolvedHint(p: Project, _), _) => p }.head + assert(bindingProject.output.map(_.exprId) == renewed.map(_.exprId)) + + // The query output schema is unchanged. + assert(optimized.output == query.output) + } + } + + test("leaves a single-reference view unchanged") { + withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") { + val a1 = attr("a", 100) + val b1 = attr("b", 101) + val query = Filter(a1 > 5, tempView("v", simpleBody(a1, b1))) + comparePlans(Optimize.execute(query), query) + } + } + + test("does not convert non-deterministic views") { + withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") { + def randBody(r: AttributeReference): LogicalPlan = + Project(Seq(rand(0).as("r")), OneRowRelation()) + val (v1, v2, _) = sameViewTwice("v", attr("r", 100), r => randBody(r)) + val query = Join(v1, v2, Inner, None, JoinHint(None, None)) + comparePlans(Optimize.execute(query), query) + } + } + + test("does not convert streaming views") { + withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") { + val (v1, v2, _) = sameViewTwice( + "v", attr("a", 100), + (a: AttributeReference) => LocalRelation(Seq(a), Nil, isStreaming = true)) + val query = Join(v1, v2, Inner, None, JoinHint(None, None)) + comparePlans(Optimize.execute(query), query) + } + } + + test("does not convert views with mixed effective SQL configs") { + withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") { + val a1 = attr("a", 100) + val b1 = attr("b", 101) + val a2 = attr("a", 200) + val b2 = attr("b", 201) + val confKey = s"${CatalogTable.VIEW_SQL_CONFIG_PREFIX}spark.sql.foo" + val descWithConf = viewDesc("v", simpleBody(a1, b1).schema).copy( + properties = Map(confKey -> "bar")) + val v1 = View(descWithConf, isTempView = true, simpleBody(a1, b1)) + val v2 = tempView("v", simpleBody(a2, b2)) + val query = Join(v1, v2, Inner, Some(a1 === a2), JoinHint(None, None)) + comparePlans(Optimize.execute(query), query) + } + } + + test("bails out when occurrence schemas mismatch") { + withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") { + // The two bodies are canonically equal, but the output column names differ + // ("x" vs "y"), so positional re-binding is unsafe and the group must be skipped. + val a1 = attr("a", 100) + val a2 = attr("a", 200) + val v1 = tempView("v", Project(Seq(a1.as("x")), LocalRelation(Seq(a1)))) + val v2 = tempView("v", Project(Seq(a2.as("y")), LocalRelation(Seq(a2)))) + val query = Join(v1, v2, Inner, None, JoinHint(None, None)) + comparePlans(Optimize.execute(query), query) + } + } + + test("refuses conversion when occurrences of the same view diverge into multiple groups") { Review Comment: **Nit (P3):** This and the other divergence tests pair a one-column body with a two-column body, so `schemasAlign` rejects them inside `qualifies` before the canonicalized bodies are compared. The `normalizeCteIds` tests cover only its collapsing side, bodies that differ only in re-minted inner CTE ids. Nothing checks the side that must keep bodies apart: a `normalizeCteIds` that mapped every inner CTE id to one ordinal would keep both suites green, yet admit two same-schema occurrences whose inner CTEs are wired differently, such as an `EXCEPT` over two inner CTEs with its operands swapped, and rebind one to the other's definition. A plan-level test with two such occurrences of one view, asserting that the view is not converted, would cover it. **Verification:** - **Regression:** A normalization that maps every inner CTE id to the same ordinal makes the new test fail. ########## sql/core/src/test/scala/org/apache/spark/sql/ConvertViewToMaterializedCTEQuerySuite.scala: ########## @@ -0,0 +1,450 @@ +/* + * 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 + +import org.apache.spark.sql.catalyst.plans.logical.{BROADCAST, Join, RepartitionByExpression, ResolvedHint} +import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper +import org.apache.spark.sql.execution.exchange.ReusedExchangeExec +import org.apache.spark.sql.execution.joins.BroadcastHashJoinExec +import org.apache.spark.sql.functions.rand +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.test.SharedSparkSession + +/** + * Integration tests for the `ConvertViewToMaterializedCTE` optimizer rule: repeated + * references to the same view are rewritten into one CTE definition with multiple + * references when `spark.sql.optimizer.convertViewToMaterializedCTE.enabled` is enabled. After + * the final `Replace CTE with Repartition` batch, a converted view shows up as one + * repartition node per reference site; exchange reuse deduplicates them at execution + * time. + */ +class ConvertViewToMaterializedCTEQuerySuite extends QueryTest with SharedSparkSession + with AdaptiveSparkPlanHelper { + import testImplicits._ + + private val selfJoinQuery = + "SELECT t1.id, t2.k FROM v t1 JOIN v t2 ON t1.id = t2.id WHERE t1.id < 10" + + private def withSelfJoinedView(f: => Unit): Unit = { + withTempView("v") { + spark.range(0, 100).select($"id", ($"id" % 10).as("k")).createOrReplaceTempView("v") + f + } + } + + private def countRepartitions(query: String): Int = + countRepartitions(spark.sql(query)) + + private def countRepartitions(df: DataFrame): Int = + df.queryExecution.optimizedPlan.collect { + case _: RepartitionByExpression => true + }.length + + test("self-joined view returns identical results with conversion enabled") { + withSelfJoinedView { + val expected = spark.sql(selfJoinQuery).collect() + withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") { + checkAnswer(spark.sql(selfJoinQuery), expected) + } + } + } + + test("conversion adds one repartition per reference site") { + withSelfJoinedView { + assert(countRepartitions(selfJoinQuery) == 0) + withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") { + // One shuffle boundary per reference site; identical shuffles are then reused. + assert(countRepartitions(selfJoinQuery) == 2) + } + } + } + + test("the converted view body is computed once") { + // Exchange reuse deduplicates the per-reference shuffles added by the conversion, + // so the view's body is evaluated once. Pin it on the executed plan: with AQE on + // the reused exchange hides behind AdaptiveSparkPlanExec and query stage nodes, + // whose subtrees are not reachable through plain `children` traversal; the + // `collect` from AdaptiveSparkPlanHelper descends through them. + withSelfJoinedView { + Seq(true, false).foreach { aqeEnabled => + withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true", + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqeEnabled.toString) { + val df = spark.sql(selfJoinQuery) + df.collect() + assert(collect(df.queryExecution.executedPlan) { + case _: ReusedExchangeExec => 1 + }.length == 1) + } + } + } + } + + test("non-deterministic views are not converted") { + withTempView("rand_view") { + spark.range(0, 10).select($"id", rand(0).as("r")) + .createOrReplaceTempView("rand_view") + val query = "SELECT t1.r FROM rand_view t1 JOIN rand_view t2 ON t1.id = t2.id" + withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") { + assert(countRepartitions(query) == 0) + } + } + } + + test("a divergent view occurrence is not rebound to a converted definition") { + // Two references to v are resolved while v has one body; a third reference sees a + // replaced body. The identifier maps to two distinct canonicalized bodies, so the + // rule must not convert at all: rewriting the divergent occurrence against the + // qualifying pair's definition would return the old body's rows for it. + withTempView("v") { + sql("CREATE OR REPLACE TEMP VIEW v AS SELECT id, id AS k FROM range(4)") + val held = spark.table("v") + val pair = held.join(held.as("b"), "id").select($"id") + sql("CREATE OR REPLACE TEMP VIEW v AS SELECT id, id + 100 AS k FROM range(4)") + // A fresh Dataset per setting so each collect analyzes a new plan (the rule runs + // during analysis; reusing one Dataset would reuse its cached analyzed plan). + def mkQuery(): DataFrame = pair.join(spark.table("v"), "id").select($"id", $"k") + assert(countRepartitions(mkQuery()) == 0) + val expected = mkQuery().collect() + withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") { + // Pin the no-conversion contract on the plan as well: rows matching the + // baseline would not catch a conversion that is semantically harmless here, + // but no repartition boundary may appear even then. + assert(countRepartitions(mkQuery()) == 0) + checkAnswer(mkQuery(), expected) + } + } + } + + test("stale view pair inside a scalar subquery does not crash") { + // The stale-view scenario where the pair sits inside a scalar subquery: the + // top-level pass declines the identifier (divergent bodies), so the rule must + // not convert inside the Subquery-rooted pass either - attaching definitions + // there would wrap the Subquery root in WithCTE and break FinishAnalysis's + // destructuring of the subquery result. + withTempView("v", "w") { + sql("CREATE OR REPLACE TEMP VIEW v AS SELECT id, id AS k FROM range(4)") + val held = spark.table("v") + val pair = held.join(held.as("b"), "id").select($"id") + pair.createOrReplaceTempView("w") + sql("CREATE OR REPLACE TEMP VIEW v AS SELECT id, id + 100 AS k FROM range(4)") + val query = "SELECT (SELECT count(*) FROM w) AS cw, (SELECT count(*) FROM v) AS cv" + val expected = spark.sql(query).collect() + withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") { + assert(countRepartitions(query) == 0) Review Comment: **Nit (P3):** `countRepartitions` uses `optimizedPlan.collect`, which does not descend into subquery plans. Here every occurrence of `v` and `w`, and every reference site, sits inside a scalar subquery, so this assertion passes whatever the rule does, and `checkAnswer` cannot tell either, because both bodies of `v` return four rows. The `ResolvedHint` assertion in "converted definition picked from a subquery occurrence drops view hints" has the same blind spot, and "insert into a table selecting from a repeatedly referenced view" checks only the rows written, which are the same whether the rule converts or not. Collecting with subqueries in the first two, and asserting that the INSERT's query plan contains the shared definition's repartitions, would make these tests fail when the behavior they are named for regresses. This extends an [existing discussion](https://github.com/apache/spark/pull/58229#discussion_r4217883538). **Verification:** - **Regression:** Each of the three tests fails when its behavior regresses: a divergent subquery occurrence that is converted or rebound, a ResolvedHint left in a subquery-resident definition, or a missing conversion in the INSERT query. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/ConvertViewToMaterializedCTE.scala: ########## @@ -0,0 +1,297 @@ +/* + * 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.optimizer + +import scala.collection.mutable + +import org.apache.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute} +import org.apache.spark.sql.catalyst.plans.logical._ +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.internal.SQLConf + +/** + * Rewrites multiple references to the same view into a single `CTERelationDef` with multiple + * `CTERelationRef`s, so that the view's underlying plan is computed once (through exchange + * reuse at the physical layer) instead of once per reference. + * + * The rule runs first in `FinishAnalysis`, before `EliminateResolvedHint`, + * `EliminateSubqueryAliases` and `EliminatePipeOperators` strip analysis-only nodes from + * the plan. Those rules are applied to the main plan before its subquery plans, so a + * definition body taken from a first-visited subquery occurrence may still hold such + * nodes; running the conversion first lets the cleanup rules reach the body at the top + * level, wherever it was taken from. `EliminateView` runs after the cleanup rules: no + * `View` nodes remain and every reference site holds an independent copy of the view's + * plan. + * + * A join hint written at the top of a view body binds to the join(s) the view takes part + * in (see `SubqueryHintPropagationSuite`): `EliminateResolvedHint` extracts hints from + * Join children by walking down unary nodes, and processes the body inline today so the + * hint reaches the consuming join. After this rule the body lives inside a `CTERelationDef` + * the walk cannot reach, so the rule strips those top-of-body hints itself and re-attaches + * them above every reference site. Hints deeper in the body bind to the body's own joins + * and are left in place. + * + * A converted definition keeps `forceSkipInline = true` only when at least two references + * survive the rewrite; otherwise `InlineCTE` would immediately flatten it back into + * duplicated subtrees (the definition body is deterministic in every case we convert), + * making the rule a no-op. Definitions with fewer than two surviving references keep the + * flag off and are inlined: their occurrences do not all survive the rewrite (a later + * occurrence's body is dropped, taking references to nested converted views with it), and + * some references may sit inside unreferenced user-written CTEs - materializing either + * way would add a shuffle that exchange reuse cannot deduplicate. + * + * Only deterministic, batch views are eligible: a multi-reference CTE guarantees that its + * definition is evaluated exactly once (even for non-deterministic definitions), while + * multiple references to a non-deterministic view are evaluated independently today. + * Converting such views would change query results. + * + * A view body may contain correlated subqueries whose outer references resolve to relations + * inside the same body (e.g. `t WHERE x IN (SELECT y FROM s WHERE s.k = t.k)`). The view is + * analyzed standalone when it is created, so an outer reference that does not resolve inside + * the body fails view analysis and can never escape to the outer query. The converted + * definition contains the whole body, so internal correlations resolve within it and these + * bodies are safe to convert. `InlineCTE`'s rejection of boundary-crossing outer references + * is only a generic safety net for non-view `forceSkipInline` producers. + * + * Each reference site gains a shuffle boundary added by `ReplaceCTERefWithRepartition` and + * deduplicated by exchange reuse, so the conversion trades recomputation for a + * shuffle plus reuse; it is therefore gated behind + * [[SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE]] and off by default. + */ +object ConvertViewToMaterializedCTE extends Rule[LogicalPlan] { + + override def apply(plan: LogicalPlan): LogicalPlan = { + if (!SQLConf.get.getConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE)) return plan + // FinishAnalysis re-runs the rule batch on Subquery roots and requires the result + // to stay a Subquery; the top-level pass already covers views inside subqueries. + if (plan.isInstanceOf[Subquery]) return plan + val occurrences = plan.collectWithSubqueries { case v: View => v } + if (occurrences.length < 2) return plan + + // Admit an identifier only when every occurrence of it forms one qualifying group. + // Grouping is by identifier first because the rewrite below matches occurrences by + // identifier: an identifier carried by occurrences with divergent bodies (e.g. one + // occurrence resolved against a view definition that was replaced after another + // occurrence captured it), or whose occurrences fail qualification for any other + // reason, must not convert at all - otherwise the rewrite would rebind those + // occurrences against a definition their plan does not match, silently changing + // query results. + // + // Inner CTE definition ids are normalized before the bodies are compared: a view + // body is re-analyzed per occurrence, and the re-analysis re-substitutes the body's + // inner CTEs, minting a fresh `CTERelationDef` id each time (e.g. for a SQL view + // whose body has a WITH clause over a persistent table). The re-minted ids are the + // only difference between such occurrences' bodies, so the normalization lets them + // group together; bodies that differ in anything else still stay apart. + val qualifiedIdentifiers = occurrences Review Comment: **Blocking (P1):** This admits every view whose occurrences qualify, without considering whether the base branch already avoids repeating the view's work, and with the flag on each surviving reference then gets a locally sorted round-robin shuffle of the whole definition. Where the base branch does no repeated work, that shuffle only adds cost: - Pushed limits: for `SELECT * FROM v UNION ALL SELECT * FROM v LIMIT 10`, or `show()` over that union, `LimitPushDown` puts a `LocalLimit` on each branch, so the base branch's copies stop after a few rows. With the flag on, each branch is a `CTERelationRef` leaf, so the limit never reaches the view body, and the shuffle's map stage evaluates and writes the entire view before either branch returns a row. - Disjoint slices: `v t1 JOIN v t2 ON t1.k = t2.k AND t1.year = 2023 AND t2.year = 2024` over a partitioned table scans one year per copy today. The shared definition scans both years and shuffles them, and each side reads both. When dynamic partition pruning picks the slices instead, the definition gets no partition filter at all. - Shared copies: in a sort-merge self-join whose copies optimize identically, exchange reuse already runs the body once; conversion adds the round-robin shuffle and discards the body's partitioning. **Recommended change:** Confine materialization to references that still repeat work, using signals that are visible when the rule runs. (1) When deciding forceSkipInline, do not count a reference that a pushed limit reaches: a UNION ALL branch, or an outer join's preserved side, under a LIMIT, through only Project, Filter and SubqueryAlias nodes. (2) Decline a view whose occurrences are each restricted, by predicates in filters or inner-join conditions that reference only that occurrence, to non-overlapping literal values of the same view column, such as year = 2023 and year = 2024. (3) Decline a view whose occurrences are the two sides of an equi-join on the same view columns and have symmetric consumers: the same view columns are referenced above each side and no predicate references only one side, so a shuffled join exchanges two identical copies that exchange reuse already shares. Add a TODO at admission that slices chosen by dynamic partition pruning, and other copies that o nly become identical after operator optimization, are not visible when the rule runs. Add plan tests for a symmetric self-join, a self-join filtered to different years of a partitioned table, a UNION ALL of the view with itself under LIMIT, and the PR's motivating self-join, which must still convert. **Why this works:** Each check reads only the analyzed plan above the occurrences: the limit position, the conjuncts of filters and inner-join conditions that reference a single occurrence, the join keys, and the view columns referenced above each occurrence. Asymmetric consumers, such as the motivating query's t1.tier = 'priority' in the join condition, still convert, because that asymmetry is what makes the base branch's copies differ and repeat work. Declined views and uncounted references stay inlined, so they keep per-reference pushdown, pruning, limits and exchange reuse as on the base branch. **Scope:** Make the materialization decision skip references and views that already avoid repeated work, using limit positions and the predicates, join keys and columns above each occurrence, and add plan tests for those shapes. **Risks:** The checks approximate later optimizer decisions. A symmetric self-join that is planned as a broadcast join scans the view twice on the base branch, so declining it gives up the benefit there. Copies that become identical or disjoint only through later rules, such as inferred filters, keep converting, as do the shapes named in the exclusions, which still regress until the decision can be made after per-reference optimization. Existing conversion tests that use a symmetric self-join, such as SELECT t1.id FROM v t1 JOIN v t2 ON t1.id = t2.id, stop converting and need a one-sided predicate to keep exercising conversion. Everything stays behind the existing flag. **Constraints:** Queries whose references overlap and differ, which the change targets, keep converting, including when the distinguishing predicate sits in a join condition or in a filter above the join. The checks only decline views or leave references out of the materialization count, so they compose with the one-row-subplan check. **Exclusions:** - Slices chosen by dynamic partition pruning, and other copies that only become identical after operator optimization, such as the same aggregation of the view in two places: neither is visible when the rule runs, and the TODO at admission names the later decision point they need. - Recovering dynamic partition pruning for overlapping references that still need materialization; that trade-off is part of the decision about materialization costs. **Success:** A symmetric self-join of a view, a self-join filtered to different years of a partitioned table, and a UNION ALL of a view with itself under LIMIT produce the same plans with the flag on as with it off. The PR's motivating self-join, filtered on one side in its join condition, still converts. **Verification:** - **Regression:** With the flag on, a symmetric self-join of a view under a sort-merge join, a self-join filtered to different years of a partitioned table, and a UNION ALL of the view with itself under LIMIT show no added round-robin repartition and scan the same partitions as with the flag off. - **Behavior:** The PR's motivating self-join of an aggregated view, filtered on one side in its join condition, still converts and evaluates the view once. ########## sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ConvertViewToMaterializedCTESuite.scala: ########## @@ -0,0 +1,670 @@ +/* + * 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.optimizer + +import org.apache.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType} +import org.apache.spark.sql.catalyst.dsl.expressions._ +import org.apache.spark.sql.catalyst.dsl.plans._ +import org.apache.spark.sql.catalyst.expressions.{AttributeReference, ExprId, In, ListQuery, Literal, NamedExpression, OuterReference, ScalarSubquery} +import org.apache.spark.sql.catalyst.plans.Inner +import org.apache.spark.sql.catalyst.plans.PlanTest +import org.apache.spark.sql.catalyst.plans.logical._ +import org.apache.spark.sql.catalyst.rules.RuleExecutor +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.{IntegerType, StructType} + +class ConvertViewToMaterializedCTESuite extends PlanTest { + + object Optimize extends RuleExecutor[LogicalPlan] { + val batches = + Batch("Convert View to Materialized CTE", FixedPoint(1), ConvertViewToMaterializedCTE) :: Nil + } + + object OptimizeWithInlineCTE extends RuleExecutor[LogicalPlan] { + val batches = + Batch("Convert View to Materialized CTE", FixedPoint(1), ConvertViewToMaterializedCTE) :: + Batch("Inline CTE", FixedPoint(1), InlineCTE()) :: Nil + } + + private def attr(name: String, id: Long): AttributeReference = + AttributeReference(name, IntegerType, nullable = true)(exprId = ExprId(id)) + + private def viewDesc(name: String, schema: StructType): CatalogTable = + CatalogTable( + identifier = TableIdentifier(name), + tableType = CatalogTableType.VIEW, + storage = CatalogStorageFormat.empty, + schema = schema) + + private def tempView(name: String, child: LogicalPlan): View = + View(viewDesc(name, child.schema), isTempView = true, child) + + /** + * One view `name` referenced twice, as the analyzer produces it: two `View` occurrences + * whose bodies are built from the same base attributes, except that the second + * occurrence's attributes carry renewed expression ids. The renewed attributes are + * returned as well for tests that reference them (e.g. in a join condition). Calling + * this instead of writing `tempView("v", ...)` twice makes it unambiguous that the pair + * stands for the same view used twice, not two views that happen to share a name. + */ + private def sameViewTwice( + name: String, + base: AttributeReference, + make: AttributeReference => LogicalPlan): (View, View, AttributeReference) = { + val renewed = base.withExprId(NamedExpression.newExprId) + (tempView(name, make(base)), tempView(name, make(renewed)), renewed) + } + + private def sameViewTwice( + name: String, + base: Seq[AttributeReference], + make: Seq[AttributeReference] => LogicalPlan): (View, View, Seq[AttributeReference]) = { + val renewed = base.map(_.withExprId(NamedExpression.newExprId)) + (tempView(name, make(base)), tempView(name, make(renewed)), renewed) + } + + // Same as above, for bodies that mint their own occurrence-specific ids per call + // (e.g. through `NamedExpression.newExprId`), so there is no base attribute to renew. + private def sameViewTwice(name: String, make: () => LogicalPlan): (View, View) = { + (tempView(name, make()), tempView(name, make())) + } + + // A simple deterministic view body: LocalRelation [a, b] filtered on a. + private def simpleBody(a: AttributeReference, b: AttributeReference): LogicalPlan = + LocalRelation(Seq(a, b)).where(a > 10) + + // A view body that contains a surviving (multi-ref) inner CTE. The references carry + // distinct expression ids, mirroring analyzer output. The inner CTE survives inlining + // either because it is non-deterministic or because it sets forceSkipInline. + private def nestedCteBody(defId: Long, deterministic: Boolean): LogicalPlan = { + val innerProject = + if (deterministic) OneRowRelation().select(Literal(1).as("r")) + else OneRowRelation().select(rand(0).as("r")) + val innerDef = CTERelationDef( + innerProject, + id = defId, + forceSkipInline = deterministic) + def mkRef(): CTERelationRef = CTERelationRef( + defId, + _resolved = true, + output = innerDef.output.map(_.withExprId(NamedExpression.newExprId)), + isStreaming = false) + WithCTE(Except(mkRef(), mkRef(), isAll = true), Seq(innerDef)) + } + + test("converts a self-joined view into one CTE definition and two references") { + withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") { + val base = Seq(attr("a", 100), attr("b", 101)) + val (v1, v2, renewed) = sameViewTwice( + "v", base, (as: Seq[AttributeReference]) => simpleBody(as(0), as(1))) + val query = Join(v1, v2, Inner, Some(base(0) === renewed(0)), JoinHint(None, None)) + + val optimized = Optimize.execute(query) + + // The root is wrapped in a WithCTE carrying exactly one definition. + val WithCTE(mainPlan, cteDefs) = optimized + assert(cteDefs.length == 1) + val cteDef = cteDefs.head + assert(cteDef.forceSkipInline, + "converted CTE must set forceSkipInline so InlineCTE keeps it materialized") + assert(cteDef.child.canonicalized == simpleBody(base(0), base(1)).canonicalized) + + // Exactly two references in the main plan. + val refs = mainPlan.collect { case r: CTERelationRef => r } + assert(refs.length == 2) + + // The first reference adopts the definition output directly. + assert(refs.exists(_.output.map(_.exprId) == cteDef.output.map(_.exprId))) + + // The second reference is re-bound through an aliasing Project that re-mints the + // original expression ids of that occurrence, so consumers need no rewriting. + val projects = mainPlan.collect { + case p @ Project(_, _: CTERelationRef) => p + } + assert(projects.length == 1) + assert(projects.head.output.map(_.name) == Seq("a", "b")) + assert(projects.head.output.map(_.exprId) == renewed.map(_.exprId)) + + // The join condition still references the original attributes of both occurrences. + val join = mainPlan.collect { case j: Join => j }.head + assert(join.condition.get == (base(0) === renewed(0))) + + // The query output schema is unchanged. + assert(optimized.output == query.output) + } + } + + test("moves view-body join hints above the references") { + withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") { + // A join hint at the top of the view body binds to the join(s) the view takes part + // in (SPARK-40999): after this rule replaces the View with a CTERelationRef, which + // is a leaf, the following EliminateResolvedHint can no longer hoist the hint + // through the tree. The definition is built from the hint-free body, and the + // extracted hints are re-attached above each reference so that hint hoisting keeps + // reaching the outer joins as before conversion. + val broadcastHint = HintInfo(Some(BROADCAST)) + val base = Seq(attr("a", 100), attr("b", 101)) + val (v1, v2, renewed) = sameViewTwice("v", base, (as: Seq[AttributeReference]) => + ResolvedHint(simpleBody(as.head, as(1)), broadcastHint)) + val query = Join(v1, v2, Inner, Some(base(0) === renewed(0)), JoinHint(None, None)) + + val optimized = Optimize.execute(query) + + val WithCTE(mainPlan, cteDefs) = optimized + assert(cteDefs.length == 1) + // The definition must be built from the hint-free body. + assert(cteDefs.head.child.collect { case h: ResolvedHint => h }.isEmpty, + s"definition body still carries the hint: ${cteDefs.head.child}") + + // Each reference carries the body's hint above it: the first occurrence wraps the + // bare reference, the later occurrence wraps its rebinding Project. + val hinted = mainPlan.collect { + case h @ ResolvedHint(_, hints) => (h, hints) + } + assert(hinted.length == 2, s"expected two hinted references in: $mainPlan") + assert(hinted.forall { case (_, hints) => hints == broadcastHint }) + assert(hinted.count { case (h, _) => h.child.isInstanceOf[CTERelationRef] } == 1, + s"first reference must be hinted directly: $mainPlan") + assert(hinted.count { case (h, _) => h.child.isInstanceOf[Project] } == 1, + s"rebinding reference must be hinted above its Project: $mainPlan") + + // The rebinding occurrence's identity is preserved under its hint. + val bindingProject = hinted.collect { case (ResolvedHint(p: Project, _), _) => p }.head + assert(bindingProject.output.map(_.exprId) == renewed.map(_.exprId)) + + // The query output schema is unchanged. + assert(optimized.output == query.output) + } + } + + test("leaves a single-reference view unchanged") { + withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") { + val a1 = attr("a", 100) + val b1 = attr("b", 101) + val query = Filter(a1 > 5, tempView("v", simpleBody(a1, b1))) + comparePlans(Optimize.execute(query), query) + } + } + + test("does not convert non-deterministic views") { + withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") { + def randBody(r: AttributeReference): LogicalPlan = + Project(Seq(rand(0).as("r")), OneRowRelation()) + val (v1, v2, _) = sameViewTwice("v", attr("r", 100), r => randBody(r)) + val query = Join(v1, v2, Inner, None, JoinHint(None, None)) + comparePlans(Optimize.execute(query), query) + } + } + + test("does not convert streaming views") { + withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") { + val (v1, v2, _) = sameViewTwice( + "v", attr("a", 100), + (a: AttributeReference) => LocalRelation(Seq(a), Nil, isStreaming = true)) + val query = Join(v1, v2, Inner, None, JoinHint(None, None)) + comparePlans(Optimize.execute(query), query) + } + } + + test("does not convert views with mixed effective SQL configs") { + withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") { + val a1 = attr("a", 100) + val b1 = attr("b", 101) + val a2 = attr("a", 200) + val b2 = attr("b", 201) + val confKey = s"${CatalogTable.VIEW_SQL_CONFIG_PREFIX}spark.sql.foo" + val descWithConf = viewDesc("v", simpleBody(a1, b1).schema).copy( + properties = Map(confKey -> "bar")) + val v1 = View(descWithConf, isTempView = true, simpleBody(a1, b1)) + val v2 = tempView("v", simpleBody(a2, b2)) + val query = Join(v1, v2, Inner, Some(a1 === a2), JoinHint(None, None)) + comparePlans(Optimize.execute(query), query) + } + } + + test("bails out when occurrence schemas mismatch") { + withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") { + // The two bodies are canonically equal, but the output column names differ + // ("x" vs "y"), so positional re-binding is unsafe and the group must be skipped. + val a1 = attr("a", 100) + val a2 = attr("a", 200) + val v1 = tempView("v", Project(Seq(a1.as("x")), LocalRelation(Seq(a1)))) + val v2 = tempView("v", Project(Seq(a2.as("y")), LocalRelation(Seq(a2)))) + val query = Join(v1, v2, Inner, None, JoinHint(None, None)) + comparePlans(Optimize.execute(query), query) + } + } + + test("refuses conversion when occurrences of the same view diverge into multiple groups") { + withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") { + // Degenerate scenario the analyzer cannot produce today: two pairs of occurrences Review Comment: **Nit (P3):** This shape is reachable: capturing `spark.table("v")` before `CREATE OR REPLACE TEMP VIEW v` produces divergent occurrences of one identifier, as the adjacent "qualifying pair plus a divergent singleton" test and the query suite's stale-view tests show, and repeating the capture produces two divergent pairs. The same label at line 304 has the same problem for the deterministic-plus-non-deterministic case. Describing both as reachable through Datasets captured before the view is replaced would keep these guards from looking like dead defensive code. **Verification:** - **Inspection:** The multi-group and mixed-determinism divergence tests describe their scenarios as reachable through stale captured Datasets. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/ConvertViewToMaterializedCTE.scala: ########## @@ -0,0 +1,297 @@ +/* + * 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.optimizer + +import scala.collection.mutable + +import org.apache.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute} +import org.apache.spark.sql.catalyst.plans.logical._ +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.internal.SQLConf + +/** + * Rewrites multiple references to the same view into a single `CTERelationDef` with multiple + * `CTERelationRef`s, so that the view's underlying plan is computed once (through exchange + * reuse at the physical layer) instead of once per reference. + * + * The rule runs first in `FinishAnalysis`, before `EliminateResolvedHint`, + * `EliminateSubqueryAliases` and `EliminatePipeOperators` strip analysis-only nodes from + * the plan. Those rules are applied to the main plan before its subquery plans, so a + * definition body taken from a first-visited subquery occurrence may still hold such + * nodes; running the conversion first lets the cleanup rules reach the body at the top + * level, wherever it was taken from. `EliminateView` runs after the cleanup rules: no + * `View` nodes remain and every reference site holds an independent copy of the view's + * plan. + * + * A join hint written at the top of a view body binds to the join(s) the view takes part + * in (see `SubqueryHintPropagationSuite`): `EliminateResolvedHint` extracts hints from + * Join children by walking down unary nodes, and processes the body inline today so the + * hint reaches the consuming join. After this rule the body lives inside a `CTERelationDef` + * the walk cannot reach, so the rule strips those top-of-body hints itself and re-attaches + * them above every reference site. Hints deeper in the body bind to the body's own joins + * and are left in place. + * + * A converted definition keeps `forceSkipInline = true` only when at least two references + * survive the rewrite; otherwise `InlineCTE` would immediately flatten it back into + * duplicated subtrees (the definition body is deterministic in every case we convert), + * making the rule a no-op. Definitions with fewer than two surviving references keep the + * flag off and are inlined: their occurrences do not all survive the rewrite (a later + * occurrence's body is dropped, taking references to nested converted views with it), and + * some references may sit inside unreferenced user-written CTEs - materializing either + * way would add a shuffle that exchange reuse cannot deduplicate. + * + * Only deterministic, batch views are eligible: a multi-reference CTE guarantees that its + * definition is evaluated exactly once (even for non-deterministic definitions), while + * multiple references to a non-deterministic view are evaluated independently today. + * Converting such views would change query results. + * + * A view body may contain correlated subqueries whose outer references resolve to relations + * inside the same body (e.g. `t WHERE x IN (SELECT y FROM s WHERE s.k = t.k)`). The view is + * analyzed standalone when it is created, so an outer reference that does not resolve inside + * the body fails view analysis and can never escape to the outer query. The converted + * definition contains the whole body, so internal correlations resolve within it and these + * bodies are safe to convert. `InlineCTE`'s rejection of boundary-crossing outer references + * is only a generic safety net for non-view `forceSkipInline` producers. + * + * Each reference site gains a shuffle boundary added by `ReplaceCTERefWithRepartition` and + * deduplicated by exchange reuse, so the conversion trades recomputation for a + * shuffle plus reuse; it is therefore gated behind + * [[SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE]] and off by default. + */ +object ConvertViewToMaterializedCTE extends Rule[LogicalPlan] { + + override def apply(plan: LogicalPlan): LogicalPlan = { + if (!SQLConf.get.getConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE)) return plan + // FinishAnalysis re-runs the rule batch on Subquery roots and requires the result + // to stay a Subquery; the top-level pass already covers views inside subqueries. + if (plan.isInstanceOf[Subquery]) return plan + val occurrences = plan.collectWithSubqueries { case v: View => v } + if (occurrences.length < 2) return plan + + // Admit an identifier only when every occurrence of it forms one qualifying group. + // Grouping is by identifier first because the rewrite below matches occurrences by + // identifier: an identifier carried by occurrences with divergent bodies (e.g. one + // occurrence resolved against a view definition that was replaced after another + // occurrence captured it), or whose occurrences fail qualification for any other + // reason, must not convert at all - otherwise the rewrite would rebind those + // occurrences against a definition their plan does not match, silently changing + // query results. + // + // Inner CTE definition ids are normalized before the bodies are compared: a view + // body is re-analyzed per occurrence, and the re-analysis re-substitutes the body's + // inner CTEs, minting a fresh `CTERelationDef` id each time (e.g. for a SQL view + // whose body has a WITH clause over a persistent table). The re-minted ids are the + // only difference between such occurrences' bodies, so the normalization lets them + // group together; bodies that differ in anything else still stay apart. + val qualifiedIdentifiers = occurrences + .groupBy(_.desc.identifier) + .collect { + case (identifier, occs) + if qualifies(occs) && occs.map(occ => + normalizeCteIds(occ.child).canonicalized).distinct.length == 1 => identifier + } + .toSet + if (qualifiedIdentifiers.isEmpty) return plan + + // Bottom-up rewrite: nested views are visited (and their definitions appended) before + // the views containing them, so a referenced definition always precedes its referrer + // in `cteDefs`. The first occurrence of a group creates the definition, replaced by a + // bare reference; later occurrences are wrapped in a `Project` re-minting the + // occurrence's ids from the definition output, so consumers above need no rewriting. + val cteDefs = mutable.ArrayBuffer.empty[CTERelationDef] + val defByGroup = mutable.HashMap.empty[TableIdentifier, (CTERelationDef, Seq[HintInfo])] + + val rewritten = plan.transformUpWithSubqueries { + case v: View if qualifiedIdentifiers.contains(v.desc.identifier) => + // The body's top-of-node hints must not stay in the definition, where neither + // this rule nor `EliminateResolvedHint` can hoist them into the consuming joins; + // build the definition from the hint-free body and re-apply the hints above each + // reference site instead. All occurrences share one canonicalized body, so the + // hints extracted from the first one describe every reference site uniformly. + defByGroup.get(v.desc.identifier) match { + case Some((cteDef, bodyHints)) => + // Later occurrence: re-bind the reference output to this occurrence's + // attributes positionally. The group qualification has already asserted that + // name, type and nullability align element-wise. + val ref = CTERelationRef( + cteDef.id, + _resolved = true, + output = cteDef.output, + isStreaming = false, + maxRows = cteDef.maxRows) + withBodyHints(bodyHints, + Project(rebindingProjectList(v.output, cteDef.output), ref)) + + case None => + // First occurrence: consumers above already reference this occurrence's + // expression ids, which are exactly the definition output, so the bare + // reference is output-compatible. `forceSkipInline` is decided after the + // rewrite, by the number of surviving references. + val (hintFreeBody, bodyHints) = + EliminateResolvedHint.extractHintsFromPlan(v.child) + val cteDef = CTERelationDef(hintFreeBody) + defByGroup.put(v.desc.identifier, (cteDef, bodyHints)) + cteDefs += cteDef + withBodyHints(bodyHints, CTERelationRef( + cteDef.id, + _resolved = true, + output = v.child.output, + isStreaming = false, + maxRows = cteDef.maxRows)) + } + } + + if (cteDefs.isEmpty) { + plan + } else { + // Not every occurrence survives the rewrite as a reference: the body of a later + // occurrence is dropped (taking references to nested converted views with it), and + // some references sit inside unreferenced user-written CTEs. Materializing a + // definition with fewer than two live references would add a shuffle that exchange + // reuse cannot deduplicate, so re-count after the rewrite and leave such definitions + // inlinable - InlineCTE flattens them back into their consumers. + val newDefIds = cteDefs.map(_.id).toSet + val withDefs = attachDefs(rewritten, cteDefs.toSeq) + val liveRefCounts = countLiveRefs(withDefs) + withDefs.transformUpWithSubqueries { + case d: CTERelationDef if newDefIds.contains(d.id) => + d.copy(forceSkipInline = liveRefCounts.getOrElse(d.id, 0) >= 2) + } + } + } + + // Counts, per definition id, the references reachable from the query root: a reference + // inside a definition counts only when the definition itself is live, mirroring how + // `InlineCTE` discounts references held by unreferenced definitions. Recursive CTEs are + // handled by the visited-set worklist. + private def countLiveRefs(plan: LogicalPlan): Map[Long, Int] = { + // The region of a reference is its innermost enclosing definition's id, or + // ROOT_REGION at the query root. + val ROOT_REGION = -1L + val outgoing = mutable.HashMap.empty[Long, mutable.ArrayBuffer[Long]] + + def walk(p: LogicalPlan, region: Long): Unit = { + p match { + case d: CTERelationDef => walk(d.child, d.id) + case r: CTERelationRef => + outgoing.getOrElseUpdate(region, mutable.ArrayBuffer.empty) += r.cteId + case _ => p.children.foreach(walk(_, region)) + } + p.subqueries.foreach(walk(_, region)) + } + walk(plan, ROOT_REGION) + + val liveRefCounts = mutable.HashMap.empty[Long, Int].withDefaultValue(0) + val visitedDefs = mutable.HashSet.empty[Long] + val pending = mutable.Queue.empty[Long] ++= + outgoing.getOrElse(ROOT_REGION, mutable.ArrayBuffer.empty) + while (pending.nonEmpty) { + val target = pending.dequeue() + liveRefCounts(target) += 1 + if (visitedDefs.add(target)) { + pending ++= outgoing.getOrElse(target, mutable.ArrayBuffer.empty) + } + } + liveRefCounts.toMap + } + + // Group by view identity: the rule dedupes references of the SAME view, not distinct + // views with coincidentally equal bodies, and all occurrences of one view resolve + // through the same (db-qualified) identifier. The canonicalized body comparison is a + // precondition guard in the identifier admission above: if occurrences of one view + // ever diverge structurally, we skip conversion instead of building a wrong shared + // definition. + // + // Rewrites every inner CTE definition id and reference to an appearance ordinal, so + // that bodies differing only in re-minted inner CTE ids canonicalize equal. Same tree + // shapes are visited in the same order, so equal bodies map to equal ordinals, while + // genuinely different structures (e.g. two sibling identical CTEs) still produce + // distinct ordinals and stay distinguishable. + private def normalizeCteIds(plan: LogicalPlan): LogicalPlan = { + val ids = mutable.HashMap.empty[Long, Long] + plan.transformUpWithSubqueries { + case d: CTERelationDef => d.copy(id = ids.getOrElseUpdate(d.id, ids.size)) + case r: CTERelationRef => r.copy(cteId = ids.getOrElseUpdate(r.cteId, ids.size)) + } + } + + private def qualifies(occs: Seq[View]): Boolean = { + val first = occs.head + occs.length >= 2 && occs.forall { v => + v.resolved && + v.child.deterministic && + !v.child.isStreaming && + !hasMainRegionSort(v.child) && + v.desc.viewSQLConfigs == first.desc.viewSQLConfigs && + schemasAlign(first, v) + } + } + + // The per-reference shuffle boundary is added above the whole definition, so it buries + // every ordering the body computes: not only a top-level ORDER BY, but also sorts + // reachable deeper through operator children - under LIMIT/OFFSET, in inner views, + // inline subqueries, the WITH clause's re-substituted definitions, and branches of set + // operations - whose order consumers observe today. Sorts confined to subquery + // expressions cannot affect the body's exposed rows and are skipped: their plans are Review Comment: **Nit (P3):** Sorts confined to subquery expressions can affect which rows the body exposes: in this PR's own pinned case, `WHERE id IN (SELECT id FROM t ORDER BY id DESC LIMIT 5)`, the subquery's `Sort` decides them. Skipping such sorts is safe because they do not define the body's output order, which is what the shuffle would destroy. Two other comments in this file have the same kind of problem. Lines 99-100 say re-minted inner CTE ids are the only difference between occurrences, but a recursive inner CTE also re-mints `UnionLoop.id`, `UnionLoopRef.loopId` and `outputAttrIds` on every analysis, which `normalizeCteIds` leaves alone, so views with `WITH RECURSIVE` never convert. And the first paragraph above `normalizeCteIds` (line 214) describes the identifier grouping in `apply`, not that function. **Verification:** - **Inspection:** The comments above hasMainRegionSort, normalizeCteIds and the identifier grouping describe their own code, including the fact that bodies with WITH RECURSIVE are not normalized. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/ConvertViewToMaterializedCTE.scala: ########## @@ -0,0 +1,297 @@ +/* + * 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.optimizer + +import scala.collection.mutable + +import org.apache.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute} +import org.apache.spark.sql.catalyst.plans.logical._ +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.internal.SQLConf + +/** + * Rewrites multiple references to the same view into a single `CTERelationDef` with multiple + * `CTERelationRef`s, so that the view's underlying plan is computed once (through exchange + * reuse at the physical layer) instead of once per reference. + * + * The rule runs first in `FinishAnalysis`, before `EliminateResolvedHint`, + * `EliminateSubqueryAliases` and `EliminatePipeOperators` strip analysis-only nodes from + * the plan. Those rules are applied to the main plan before its subquery plans, so a + * definition body taken from a first-visited subquery occurrence may still hold such + * nodes; running the conversion first lets the cleanup rules reach the body at the top + * level, wherever it was taken from. `EliminateView` runs after the cleanup rules: no + * `View` nodes remain and every reference site holds an independent copy of the view's + * plan. + * + * A join hint written at the top of a view body binds to the join(s) the view takes part + * in (see `SubqueryHintPropagationSuite`): `EliminateResolvedHint` extracts hints from + * Join children by walking down unary nodes, and processes the body inline today so the + * hint reaches the consuming join. After this rule the body lives inside a `CTERelationDef` + * the walk cannot reach, so the rule strips those top-of-body hints itself and re-attaches + * them above every reference site. Hints deeper in the body bind to the body's own joins + * and are left in place. + * + * A converted definition keeps `forceSkipInline = true` only when at least two references + * survive the rewrite; otherwise `InlineCTE` would immediately flatten it back into + * duplicated subtrees (the definition body is deterministic in every case we convert), + * making the rule a no-op. Definitions with fewer than two surviving references keep the + * flag off and are inlined: their occurrences do not all survive the rewrite (a later + * occurrence's body is dropped, taking references to nested converted views with it), and + * some references may sit inside unreferenced user-written CTEs - materializing either + * way would add a shuffle that exchange reuse cannot deduplicate. + * + * Only deterministic, batch views are eligible: a multi-reference CTE guarantees that its + * definition is evaluated exactly once (even for non-deterministic definitions), while + * multiple references to a non-deterministic view are evaluated independently today. + * Converting such views would change query results. + * + * A view body may contain correlated subqueries whose outer references resolve to relations + * inside the same body (e.g. `t WHERE x IN (SELECT y FROM s WHERE s.k = t.k)`). The view is + * analyzed standalone when it is created, so an outer reference that does not resolve inside + * the body fails view analysis and can never escape to the outer query. The converted + * definition contains the whole body, so internal correlations resolve within it and these + * bodies are safe to convert. `InlineCTE`'s rejection of boundary-crossing outer references + * is only a generic safety net for non-view `forceSkipInline` producers. + * + * Each reference site gains a shuffle boundary added by `ReplaceCTERefWithRepartition` and + * deduplicated by exchange reuse, so the conversion trades recomputation for a + * shuffle plus reuse; it is therefore gated behind + * [[SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE]] and off by default. + */ +object ConvertViewToMaterializedCTE extends Rule[LogicalPlan] { + + override def apply(plan: LogicalPlan): LogicalPlan = { + if (!SQLConf.get.getConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE)) return plan + // FinishAnalysis re-runs the rule batch on Subquery roots and requires the result + // to stay a Subquery; the top-level pass already covers views inside subqueries. + if (plan.isInstanceOf[Subquery]) return plan + val occurrences = plan.collectWithSubqueries { case v: View => v } + if (occurrences.length < 2) return plan + + // Admit an identifier only when every occurrence of it forms one qualifying group. + // Grouping is by identifier first because the rewrite below matches occurrences by + // identifier: an identifier carried by occurrences with divergent bodies (e.g. one + // occurrence resolved against a view definition that was replaced after another + // occurrence captured it), or whose occurrences fail qualification for any other + // reason, must not convert at all - otherwise the rewrite would rebind those + // occurrences against a definition their plan does not match, silently changing + // query results. + // + // Inner CTE definition ids are normalized before the bodies are compared: a view + // body is re-analyzed per occurrence, and the re-analysis re-substitutes the body's + // inner CTEs, minting a fresh `CTERelationDef` id each time (e.g. for a SQL view + // whose body has a WITH clause over a persistent table). The re-minted ids are the + // only difference between such occurrences' bodies, so the normalization lets them + // group together; bodies that differ in anything else still stay apart. + val qualifiedIdentifiers = occurrences + .groupBy(_.desc.identifier) + .collect { + case (identifier, occs) + if qualifies(occs) && occs.map(occ => + normalizeCteIds(occ.child).canonicalized).distinct.length == 1 => identifier + } + .toSet + if (qualifiedIdentifiers.isEmpty) return plan + + // Bottom-up rewrite: nested views are visited (and their definitions appended) before + // the views containing them, so a referenced definition always precedes its referrer + // in `cteDefs`. The first occurrence of a group creates the definition, replaced by a + // bare reference; later occurrences are wrapped in a `Project` re-minting the + // occurrence's ids from the definition output, so consumers above need no rewriting. + val cteDefs = mutable.ArrayBuffer.empty[CTERelationDef] + val defByGroup = mutable.HashMap.empty[TableIdentifier, (CTERelationDef, Seq[HintInfo])] + + val rewritten = plan.transformUpWithSubqueries { + case v: View if qualifiedIdentifiers.contains(v.desc.identifier) => + // The body's top-of-node hints must not stay in the definition, where neither + // this rule nor `EliminateResolvedHint` can hoist them into the consuming joins; + // build the definition from the hint-free body and re-apply the hints above each + // reference site instead. All occurrences share one canonicalized body, so the Review Comment: **Nit (P3):** This premise does not hold: `ResolvedHint.doCanonicalize` returns `child.canonicalized`, so occurrences whose bodies differ only in a hint canonicalize equal and are admitted together. That happens, for example, when a Dataset captured `spark.table("v")` while `v` carried `/*+ BROADCAST(t) */`, and `v` was then replaced without the hint. Every reference site then gets the first-visited occurrence's hints, so a broadcast hint is added to or dropped from joins whose own view body said otherwise. Results do not change, but join strategies do. Extracting each occurrence's own top-of-body hints and re-attaching them above that occurrence's reference, while still building the definition from the first occurrence's hint-free body, keeps every site's hints; this comment should then be corrected. This extends an [existing discussion](https://github.com/apache/spark/pull/58229#discussion_r4221085479). **Verification:** - **Regression:** With the flag on, a query that joins occurrences of a view captured with a BROADCAST hint and occurrences of the same view replaced without it keeps the hint only at the captured occurrences' joins, as with the flag off. - **Behavior:** When every occurrence carries the same hint, both reference sites still get the hint and keep their broadcast joins. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/ConvertViewToMaterializedCTE.scala: ########## @@ -0,0 +1,297 @@ +/* + * 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.optimizer + +import scala.collection.mutable + +import org.apache.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute} +import org.apache.spark.sql.catalyst.plans.logical._ +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.internal.SQLConf + +/** + * Rewrites multiple references to the same view into a single `CTERelationDef` with multiple + * `CTERelationRef`s, so that the view's underlying plan is computed once (through exchange + * reuse at the physical layer) instead of once per reference. + * + * The rule runs first in `FinishAnalysis`, before `EliminateResolvedHint`, + * `EliminateSubqueryAliases` and `EliminatePipeOperators` strip analysis-only nodes from + * the plan. Those rules are applied to the main plan before its subquery plans, so a + * definition body taken from a first-visited subquery occurrence may still hold such + * nodes; running the conversion first lets the cleanup rules reach the body at the top + * level, wherever it was taken from. `EliminateView` runs after the cleanup rules: no + * `View` nodes remain and every reference site holds an independent copy of the view's + * plan. + * + * A join hint written at the top of a view body binds to the join(s) the view takes part + * in (see `SubqueryHintPropagationSuite`): `EliminateResolvedHint` extracts hints from + * Join children by walking down unary nodes, and processes the body inline today so the + * hint reaches the consuming join. After this rule the body lives inside a `CTERelationDef` + * the walk cannot reach, so the rule strips those top-of-body hints itself and re-attaches + * them above every reference site. Hints deeper in the body bind to the body's own joins + * and are left in place. + * + * A converted definition keeps `forceSkipInline = true` only when at least two references + * survive the rewrite; otherwise `InlineCTE` would immediately flatten it back into + * duplicated subtrees (the definition body is deterministic in every case we convert), + * making the rule a no-op. Definitions with fewer than two surviving references keep the + * flag off and are inlined: their occurrences do not all survive the rewrite (a later + * occurrence's body is dropped, taking references to nested converted views with it), and + * some references may sit inside unreferenced user-written CTEs - materializing either + * way would add a shuffle that exchange reuse cannot deduplicate. + * + * Only deterministic, batch views are eligible: a multi-reference CTE guarantees that its + * definition is evaluated exactly once (even for non-deterministic definitions), while + * multiple references to a non-deterministic view are evaluated independently today. + * Converting such views would change query results. + * + * A view body may contain correlated subqueries whose outer references resolve to relations + * inside the same body (e.g. `t WHERE x IN (SELECT y FROM s WHERE s.k = t.k)`). The view is + * analyzed standalone when it is created, so an outer reference that does not resolve inside + * the body fails view analysis and can never escape to the outer query. The converted + * definition contains the whole body, so internal correlations resolve within it and these + * bodies are safe to convert. `InlineCTE`'s rejection of boundary-crossing outer references + * is only a generic safety net for non-view `forceSkipInline` producers. + * + * Each reference site gains a shuffle boundary added by `ReplaceCTERefWithRepartition` and + * deduplicated by exchange reuse, so the conversion trades recomputation for a + * shuffle plus reuse; it is therefore gated behind + * [[SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE]] and off by default. + */ +object ConvertViewToMaterializedCTE extends Rule[LogicalPlan] { + + override def apply(plan: LogicalPlan): LogicalPlan = { + if (!SQLConf.get.getConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE)) return plan + // FinishAnalysis re-runs the rule batch on Subquery roots and requires the result + // to stay a Subquery; the top-level pass already covers views inside subqueries. + if (plan.isInstanceOf[Subquery]) return plan + val occurrences = plan.collectWithSubqueries { case v: View => v } + if (occurrences.length < 2) return plan + + // Admit an identifier only when every occurrence of it forms one qualifying group. + // Grouping is by identifier first because the rewrite below matches occurrences by + // identifier: an identifier carried by occurrences with divergent bodies (e.g. one + // occurrence resolved against a view definition that was replaced after another + // occurrence captured it), or whose occurrences fail qualification for any other + // reason, must not convert at all - otherwise the rewrite would rebind those + // occurrences against a definition their plan does not match, silently changing + // query results. + // + // Inner CTE definition ids are normalized before the bodies are compared: a view + // body is re-analyzed per occurrence, and the re-analysis re-substitutes the body's + // inner CTEs, minting a fresh `CTERelationDef` id each time (e.g. for a SQL view + // whose body has a WITH clause over a persistent table). The re-minted ids are the + // only difference between such occurrences' bodies, so the normalization lets them + // group together; bodies that differ in anything else still stay apart. + val qualifiedIdentifiers = occurrences + .groupBy(_.desc.identifier) + .collect { + case (identifier, occs) + if qualifies(occs) && occs.map(occ => + normalizeCteIds(occ.child).canonicalized).distinct.length == 1 => identifier + } + .toSet + if (qualifiedIdentifiers.isEmpty) return plan + + // Bottom-up rewrite: nested views are visited (and their definitions appended) before + // the views containing them, so a referenced definition always precedes its referrer + // in `cteDefs`. The first occurrence of a group creates the definition, replaced by a + // bare reference; later occurrences are wrapped in a `Project` re-minting the + // occurrence's ids from the definition output, so consumers above need no rewriting. + val cteDefs = mutable.ArrayBuffer.empty[CTERelationDef] + val defByGroup = mutable.HashMap.empty[TableIdentifier, (CTERelationDef, Seq[HintInfo])] + + val rewritten = plan.transformUpWithSubqueries { + case v: View if qualifiedIdentifiers.contains(v.desc.identifier) => + // The body's top-of-node hints must not stay in the definition, where neither + // this rule nor `EliminateResolvedHint` can hoist them into the consuming joins; + // build the definition from the hint-free body and re-apply the hints above each + // reference site instead. All occurrences share one canonicalized body, so the + // hints extracted from the first one describe every reference site uniformly. + defByGroup.get(v.desc.identifier) match { + case Some((cteDef, bodyHints)) => + // Later occurrence: re-bind the reference output to this occurrence's + // attributes positionally. The group qualification has already asserted that + // name, type and nullability align element-wise. + val ref = CTERelationRef( + cteDef.id, + _resolved = true, + output = cteDef.output, + isStreaming = false, + maxRows = cteDef.maxRows) + withBodyHints(bodyHints, + Project(rebindingProjectList(v.output, cteDef.output), ref)) + + case None => + // First occurrence: consumers above already reference this occurrence's + // expression ids, which are exactly the definition output, so the bare + // reference is output-compatible. `forceSkipInline` is decided after the + // rewrite, by the number of surviving references. + val (hintFreeBody, bodyHints) = + EliminateResolvedHint.extractHintsFromPlan(v.child) + val cteDef = CTERelationDef(hintFreeBody) + defByGroup.put(v.desc.identifier, (cteDef, bodyHints)) + cteDefs += cteDef + withBodyHints(bodyHints, CTERelationRef( + cteDef.id, + _resolved = true, + output = v.child.output, + isStreaming = false, + maxRows = cteDef.maxRows)) + } + } + + if (cteDefs.isEmpty) { + plan + } else { + // Not every occurrence survives the rewrite as a reference: the body of a later + // occurrence is dropped (taking references to nested converted views with it), and + // some references sit inside unreferenced user-written CTEs. Materializing a + // definition with fewer than two live references would add a shuffle that exchange + // reuse cannot deduplicate, so re-count after the rewrite and leave such definitions + // inlinable - InlineCTE flattens them back into their consumers. + val newDefIds = cteDefs.map(_.id).toSet + val withDefs = attachDefs(rewritten, cteDefs.toSeq) + val liveRefCounts = countLiveRefs(withDefs) + withDefs.transformUpWithSubqueries { + case d: CTERelationDef if newDefIds.contains(d.id) => + d.copy(forceSkipInline = liveRefCounts.getOrElse(d.id, 0) >= 2) Review Comment: **Blocking (P1):** With two live references this sets `forceSkipInline`, so the definition survives `InlineCTE` and is still in the plan when `SparkOptimizer` runs `MergeSubplans`, which returns any plan containing a CTE unchanged. For queries whose references to the view are all one-row subplans, such as this PR's `SELECT (SELECT count(*) FROM v) AS c, (SELECT max(id) FROM v) AS m`, the base branch already merges them into one scan and one aggregate with no repartition, so the view body runs once without any shuffle. With the flag on, the body still runs once but now pays a round-robin shuffle of its output and one aggregation per subquery, and every other mergeable subplan in the same query also loses merging. **Recommended change:** When every live reference to a converted view is a one-row subplan, meaning a non-correlated scalar subquery or a non-grouping aggregate over the view, leave the definition inlinable instead of forcing materialization. InlineCTE then restores the base branch's plan and MergeSubplans can merge the subplans. Add a plan test that pins the merged plan for scalar subqueries over a view. **Why this works:** While counting live references, also record whether each one sits inside a one-row subplan. A definition gets forceSkipInline only when it has at least two live references and at least one of them is not such a subplan. Inlinable definitions are flattened by InlineCTE, so no CTE remains when MergeSubplans runs. **Scope:** Refine the materialization decision for views referenced only from one-row subplans, and add tests. **Risks:** Views referenced only from one-row subplans are evaluated as on the base branch; when MergeSubplans cannot merge those subplans, they are again evaluated once per reference. **Constraints:** Queries in which at least one reference is not a one-row subplan still materialize the view, so MergeSubplans stays skipped for them; that cost belongs to the separate trade-off about undocumented materialization costs. **Exclusions:** - Restoring MergeSubplans for mixed queries that still materialize a view. **Success:** SELECT (SELECT count(*) FROM v) AS c, (SELECT max(id) FROM v) AS m produces the same merged single-scan plan with the flag on as with it off. **Verification:** - **Regression:** With the flag on, two non-correlated scalar subqueries that aggregate the same view are merged into one subplan with no round-robin repartition and return the same results as with the flag off. - **Behavior:** A view referenced twice in the main plan, such as a self-join, still converts into one shared definition. ########## sql/core/src/test/scala/org/apache/spark/sql/ConvertViewToMaterializedCTEQuerySuite.scala: ########## @@ -0,0 +1,450 @@ +/* + * 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 + +import org.apache.spark.sql.catalyst.plans.logical.{BROADCAST, Join, RepartitionByExpression, ResolvedHint} +import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper +import org.apache.spark.sql.execution.exchange.ReusedExchangeExec +import org.apache.spark.sql.execution.joins.BroadcastHashJoinExec +import org.apache.spark.sql.functions.rand +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.test.SharedSparkSession + +/** + * Integration tests for the `ConvertViewToMaterializedCTE` optimizer rule: repeated + * references to the same view are rewritten into one CTE definition with multiple + * references when `spark.sql.optimizer.convertViewToMaterializedCTE.enabled` is enabled. After + * the final `Replace CTE with Repartition` batch, a converted view shows up as one + * repartition node per reference site; exchange reuse deduplicates them at execution + * time. + */ +class ConvertViewToMaterializedCTEQuerySuite extends QueryTest with SharedSparkSession + with AdaptiveSparkPlanHelper { + import testImplicits._ + + private val selfJoinQuery = + "SELECT t1.id, t2.k FROM v t1 JOIN v t2 ON t1.id = t2.id WHERE t1.id < 10" + + private def withSelfJoinedView(f: => Unit): Unit = { + withTempView("v") { + spark.range(0, 100).select($"id", ($"id" % 10).as("k")).createOrReplaceTempView("v") + f + } + } + + private def countRepartitions(query: String): Int = + countRepartitions(spark.sql(query)) + + private def countRepartitions(df: DataFrame): Int = + df.queryExecution.optimizedPlan.collect { + case _: RepartitionByExpression => true + }.length + + test("self-joined view returns identical results with conversion enabled") { + withSelfJoinedView { + val expected = spark.sql(selfJoinQuery).collect() + withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") { + checkAnswer(spark.sql(selfJoinQuery), expected) + } + } + } + + test("conversion adds one repartition per reference site") { + withSelfJoinedView { + assert(countRepartitions(selfJoinQuery) == 0) + withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") { + // One shuffle boundary per reference site; identical shuffles are then reused. + assert(countRepartitions(selfJoinQuery) == 2) + } + } + } + + test("the converted view body is computed once") { + // Exchange reuse deduplicates the per-reference shuffles added by the conversion, + // so the view's body is evaluated once. Pin it on the executed plan: with AQE on + // the reused exchange hides behind AdaptiveSparkPlanExec and query stage nodes, + // whose subtrees are not reachable through plain `children` traversal; the + // `collect` from AdaptiveSparkPlanHelper descends through them. + withSelfJoinedView { + Seq(true, false).foreach { aqeEnabled => + withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true", + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqeEnabled.toString) { + val df = spark.sql(selfJoinQuery) + df.collect() + assert(collect(df.queryExecution.executedPlan) { + case _: ReusedExchangeExec => 1 + }.length == 1) + } + } + } + } + + test("non-deterministic views are not converted") { + withTempView("rand_view") { + spark.range(0, 10).select($"id", rand(0).as("r")) + .createOrReplaceTempView("rand_view") + val query = "SELECT t1.r FROM rand_view t1 JOIN rand_view t2 ON t1.id = t2.id" + withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") { + assert(countRepartitions(query) == 0) + } + } + } + + test("a divergent view occurrence is not rebound to a converted definition") { + // Two references to v are resolved while v has one body; a third reference sees a + // replaced body. The identifier maps to two distinct canonicalized bodies, so the + // rule must not convert at all: rewriting the divergent occurrence against the + // qualifying pair's definition would return the old body's rows for it. + withTempView("v") { + sql("CREATE OR REPLACE TEMP VIEW v AS SELECT id, id AS k FROM range(4)") + val held = spark.table("v") + val pair = held.join(held.as("b"), "id").select($"id") + sql("CREATE OR REPLACE TEMP VIEW v AS SELECT id, id + 100 AS k FROM range(4)") + // A fresh Dataset per setting so each collect analyzes a new plan (the rule runs + // during analysis; reusing one Dataset would reuse its cached analyzed plan). Review Comment: **Nit (P3):** The rule does not run during analysis: it is registered only in the optimizer's `FinishAnalysis` batch, so converted CTEs never appear in the analyzed plan. A fresh Dataset is needed because `QueryExecution` caches the optimized plan. Separately, the comment at line 284 says the subquery-Sort exemption is "pinned by the next test", but the next test is the TABLESAMPLE rejection; naming "a Sort confined to a subquery expression does not block conversion" there would keep the reference from going stale again. **Verification:** - **Inspection:** No comment in the suite says the rule runs during analysis, and the subquery-Sort cross-reference names the test that asserts conversion. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/ConvertViewToMaterializedCTE.scala: ########## @@ -0,0 +1,297 @@ +/* + * 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.optimizer + +import scala.collection.mutable + +import org.apache.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute} +import org.apache.spark.sql.catalyst.plans.logical._ +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.internal.SQLConf + +/** + * Rewrites multiple references to the same view into a single `CTERelationDef` with multiple + * `CTERelationRef`s, so that the view's underlying plan is computed once (through exchange + * reuse at the physical layer) instead of once per reference. + * + * The rule runs first in `FinishAnalysis`, before `EliminateResolvedHint`, + * `EliminateSubqueryAliases` and `EliminatePipeOperators` strip analysis-only nodes from + * the plan. Those rules are applied to the main plan before its subquery plans, so a + * definition body taken from a first-visited subquery occurrence may still hold such + * nodes; running the conversion first lets the cleanup rules reach the body at the top + * level, wherever it was taken from. `EliminateView` runs after the cleanup rules: no + * `View` nodes remain and every reference site holds an independent copy of the view's + * plan. + * + * A join hint written at the top of a view body binds to the join(s) the view takes part + * in (see `SubqueryHintPropagationSuite`): `EliminateResolvedHint` extracts hints from + * Join children by walking down unary nodes, and processes the body inline today so the + * hint reaches the consuming join. After this rule the body lives inside a `CTERelationDef` + * the walk cannot reach, so the rule strips those top-of-body hints itself and re-attaches + * them above every reference site. Hints deeper in the body bind to the body's own joins + * and are left in place. + * + * A converted definition keeps `forceSkipInline = true` only when at least two references + * survive the rewrite; otherwise `InlineCTE` would immediately flatten it back into + * duplicated subtrees (the definition body is deterministic in every case we convert), + * making the rule a no-op. Definitions with fewer than two surviving references keep the + * flag off and are inlined: their occurrences do not all survive the rewrite (a later + * occurrence's body is dropped, taking references to nested converted views with it), and + * some references may sit inside unreferenced user-written CTEs - materializing either + * way would add a shuffle that exchange reuse cannot deduplicate. + * + * Only deterministic, batch views are eligible: a multi-reference CTE guarantees that its + * definition is evaluated exactly once (even for non-deterministic definitions), while + * multiple references to a non-deterministic view are evaluated independently today. + * Converting such views would change query results. + * + * A view body may contain correlated subqueries whose outer references resolve to relations + * inside the same body (e.g. `t WHERE x IN (SELECT y FROM s WHERE s.k = t.k)`). The view is + * analyzed standalone when it is created, so an outer reference that does not resolve inside + * the body fails view analysis and can never escape to the outer query. The converted + * definition contains the whole body, so internal correlations resolve within it and these + * bodies are safe to convert. `InlineCTE`'s rejection of boundary-crossing outer references + * is only a generic safety net for non-view `forceSkipInline` producers. + * + * Each reference site gains a shuffle boundary added by `ReplaceCTERefWithRepartition` and + * deduplicated by exchange reuse, so the conversion trades recomputation for a + * shuffle plus reuse; it is therefore gated behind + * [[SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE]] and off by default. + */ +object ConvertViewToMaterializedCTE extends Rule[LogicalPlan] { + + override def apply(plan: LogicalPlan): LogicalPlan = { + if (!SQLConf.get.getConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE)) return plan + // FinishAnalysis re-runs the rule batch on Subquery roots and requires the result + // to stay a Subquery; the top-level pass already covers views inside subqueries. + if (plan.isInstanceOf[Subquery]) return plan + val occurrences = plan.collectWithSubqueries { case v: View => v } + if (occurrences.length < 2) return plan + + // Admit an identifier only when every occurrence of it forms one qualifying group. + // Grouping is by identifier first because the rewrite below matches occurrences by + // identifier: an identifier carried by occurrences with divergent bodies (e.g. one + // occurrence resolved against a view definition that was replaced after another + // occurrence captured it), or whose occurrences fail qualification for any other + // reason, must not convert at all - otherwise the rewrite would rebind those + // occurrences against a definition their plan does not match, silently changing + // query results. + // + // Inner CTE definition ids are normalized before the bodies are compared: a view + // body is re-analyzed per occurrence, and the re-analysis re-substitutes the body's + // inner CTEs, minting a fresh `CTERelationDef` id each time (e.g. for a SQL view + // whose body has a WITH clause over a persistent table). The re-minted ids are the + // only difference between such occurrences' bodies, so the normalization lets them + // group together; bodies that differ in anything else still stay apart. + val qualifiedIdentifiers = occurrences + .groupBy(_.desc.identifier) + .collect { + case (identifier, occs) + if qualifies(occs) && occs.map(occ => + normalizeCteIds(occ.child).canonicalized).distinct.length == 1 => identifier + } + .toSet + if (qualifiedIdentifiers.isEmpty) return plan + + // Bottom-up rewrite: nested views are visited (and their definitions appended) before + // the views containing them, so a referenced definition always precedes its referrer + // in `cteDefs`. The first occurrence of a group creates the definition, replaced by a + // bare reference; later occurrences are wrapped in a `Project` re-minting the + // occurrence's ids from the definition output, so consumers above need no rewriting. + val cteDefs = mutable.ArrayBuffer.empty[CTERelationDef] + val defByGroup = mutable.HashMap.empty[TableIdentifier, (CTERelationDef, Seq[HintInfo])] + + val rewritten = plan.transformUpWithSubqueries { + case v: View if qualifiedIdentifiers.contains(v.desc.identifier) => + // The body's top-of-node hints must not stay in the definition, where neither + // this rule nor `EliminateResolvedHint` can hoist them into the consuming joins; + // build the definition from the hint-free body and re-apply the hints above each + // reference site instead. All occurrences share one canonicalized body, so the + // hints extracted from the first one describe every reference site uniformly. + defByGroup.get(v.desc.identifier) match { + case Some((cteDef, bodyHints)) => + // Later occurrence: re-bind the reference output to this occurrence's + // attributes positionally. The group qualification has already asserted that + // name, type and nullability align element-wise. + val ref = CTERelationRef( + cteDef.id, + _resolved = true, + output = cteDef.output, + isStreaming = false, + maxRows = cteDef.maxRows) + withBodyHints(bodyHints, + Project(rebindingProjectList(v.output, cteDef.output), ref)) + + case None => + // First occurrence: consumers above already reference this occurrence's + // expression ids, which are exactly the definition output, so the bare + // reference is output-compatible. `forceSkipInline` is decided after the + // rewrite, by the number of surviving references. + val (hintFreeBody, bodyHints) = + EliminateResolvedHint.extractHintsFromPlan(v.child) + val cteDef = CTERelationDef(hintFreeBody) + defByGroup.put(v.desc.identifier, (cteDef, bodyHints)) + cteDefs += cteDef + withBodyHints(bodyHints, CTERelationRef( + cteDef.id, + _resolved = true, + output = v.child.output, + isStreaming = false, + maxRows = cteDef.maxRows)) + } + } + + if (cteDefs.isEmpty) { + plan + } else { + // Not every occurrence survives the rewrite as a reference: the body of a later + // occurrence is dropped (taking references to nested converted views with it), and + // some references sit inside unreferenced user-written CTEs. Materializing a + // definition with fewer than two live references would add a shuffle that exchange + // reuse cannot deduplicate, so re-count after the rewrite and leave such definitions + // inlinable - InlineCTE flattens them back into their consumers. + val newDefIds = cteDefs.map(_.id).toSet + val withDefs = attachDefs(rewritten, cteDefs.toSeq) + val liveRefCounts = countLiveRefs(withDefs) + withDefs.transformUpWithSubqueries { + case d: CTERelationDef if newDefIds.contains(d.id) => + d.copy(forceSkipInline = liveRefCounts.getOrElse(d.id, 0) >= 2) + } + } + } + + // Counts, per definition id, the references reachable from the query root: a reference + // inside a definition counts only when the definition itself is live, mirroring how + // `InlineCTE` discounts references held by unreferenced definitions. Recursive CTEs are + // handled by the visited-set worklist. + private def countLiveRefs(plan: LogicalPlan): Map[Long, Int] = { + // The region of a reference is its innermost enclosing definition's id, or + // ROOT_REGION at the query root. + val ROOT_REGION = -1L + val outgoing = mutable.HashMap.empty[Long, mutable.ArrayBuffer[Long]] + + def walk(p: LogicalPlan, region: Long): Unit = { + p match { + case d: CTERelationDef => walk(d.child, d.id) + case r: CTERelationRef => + outgoing.getOrElseUpdate(region, mutable.ArrayBuffer.empty) += r.cteId + case _ => p.children.foreach(walk(_, region)) + } + p.subqueries.foreach(walk(_, region)) + } + walk(plan, ROOT_REGION) + + val liveRefCounts = mutable.HashMap.empty[Long, Int].withDefaultValue(0) + val visitedDefs = mutable.HashSet.empty[Long] + val pending = mutable.Queue.empty[Long] ++= + outgoing.getOrElse(ROOT_REGION, mutable.ArrayBuffer.empty) + while (pending.nonEmpty) { + val target = pending.dequeue() + liveRefCounts(target) += 1 + if (visitedDefs.add(target)) { + pending ++= outgoing.getOrElse(target, mutable.ArrayBuffer.empty) + } + } + liveRefCounts.toMap + } + + // Group by view identity: the rule dedupes references of the SAME view, not distinct + // views with coincidentally equal bodies, and all occurrences of one view resolve + // through the same (db-qualified) identifier. The canonicalized body comparison is a + // precondition guard in the identifier admission above: if occurrences of one view + // ever diverge structurally, we skip conversion instead of building a wrong shared + // definition. + // + // Rewrites every inner CTE definition id and reference to an appearance ordinal, so + // that bodies differing only in re-minted inner CTE ids canonicalize equal. Same tree + // shapes are visited in the same order, so equal bodies map to equal ordinals, while + // genuinely different structures (e.g. two sibling identical CTEs) still produce + // distinct ordinals and stay distinguishable. + private def normalizeCteIds(plan: LogicalPlan): LogicalPlan = { + val ids = mutable.HashMap.empty[Long, Long] + plan.transformUpWithSubqueries { + case d: CTERelationDef => d.copy(id = ids.getOrElseUpdate(d.id, ids.size)) + case r: CTERelationRef => r.copy(cteId = ids.getOrElseUpdate(r.cteId, ids.size)) + } + } + + private def qualifies(occs: Seq[View]): Boolean = { + val first = occs.head + occs.length >= 2 && occs.forall { v => + v.resolved && + v.child.deterministic && Review Comment: **Nit (P3):** `Sample` has no expressions and does not override `deterministic`, so a body containing `TABLESAMPLE (50 PERCENT)` without `REPEATABLE` passes this check, and every re-analysis of the view produces the same `Sample(seed = None)`, so the occurrences also share one canonical body. `SampleExec` picks a fresh seed per physical node, so on the base branch references without a shared exchange above them sample independently. For example, `SELECT count(DISTINCT id) FROM (SELECT id FROM v UNION ALL SELECT id FROM v)` draws two independent samples today, and with the flag on both branches read one shared sample, changing the result without any error. Treating a body that contains a `Sample` without a seed as ineligible, and naming unseeded `TABLESAMPLE` in the class and config docs, keeps the rule that non-deterministic views are not converted. **Verification:** - **Regression:** With the flag on, a view over TABLESAMPLE (50 PERCENT) referenced twice in a UNION ALL produces no added repartition and draws an independent sample per reference. - **Behavior:** A view over TABLESAMPLE (50 PERCENT) REPEATABLE (1) referenced twice still converts. -- 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]
