peter-toth commented on code in PR #58229: URL: https://github.com/apache/spark/pull/58229#discussion_r4093602457
########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/ConvertViewToMaterializedCTE.scala: ########## @@ -0,0 +1,219 @@ +/* + * 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 in `FinishAnalysis`, immediately before `EliminateView`: after `EliminateView` + * no `View` nodes remain and every reference site holds an independent copy of the view's plan. + * + * A converted definition always sets `forceSkipInline = true`; 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. + * + * 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] + + val rewritten = plan.transformUpWithSubqueries { + case v: View if qualifiedIdentifiers.contains(v.desc.identifier) => + defByGroup.get(v.desc.identifier) match { + case Some(cteDef) => + // 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) + 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. + val cteDef = CTERelationDef(v.child, forceSkipInline = true) + defByGroup.put(v.desc.identifier, cteDef) + cteDefs += cteDef + CTERelationRef( + cteDef.id, + _resolved = true, + output = v.child.output, + isStreaming = false, + maxRows = cteDef.maxRows) + } + } + + if (cteDefs.isEmpty) { + plan + } else { + attachDefs(rewritten, cteDefs.toSeq) + } + } + + // 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 && + !hasTopLevelSort(v.child) && + v.desc.viewSQLConfigs == first.desc.viewSQLConfigs && + schemasAlign(first, v) + } + } + + // The per-reference shuffle boundary is added above the definition, so a + // top-level ORDER BY in the view body would be destroyed by it. + private def hasTopLevelSort(plan: LogicalPlan): Boolean = plan match { + case _: Sort => true + case Project(_, child) => hasTopLevelSort(child) + case Filter(_, child) => hasTopLevelSort(child) + case SubqueryAlias(_, child) => hasTopLevelSort(child) + case GlobalLimit(_, child) => hasTopLevelSort(child) + case LocalLimit(_, child) => hasTopLevelSort(child) Review Comment: **Finding 6.** `hasTopLevelSort` looks through `Project`, `Filter`, `SubqueryAlias` and the two limits, but not `Offset`. `ORDER BY id OFFSET 1` analyzes to `Offset(1, Sort(...))`, and `ORDER BY id LIMIT 50 OFFSET 1` to `GlobalLimit(LocalLimit(Offset(Sort(...))))`. Both reach `case _ => false`, so the view converts. I ran your `view with top-level ORDER BY is not converted` test with those two bodies on this head: | view body | repartitions | rows, conversion off | rows, conversion on | |---|---|---|---| | `SELECT id FROM range(100) ORDER BY id OFFSET 1` | 2 | `1, 2, 3` | `19, 61, 14` | | `SELECT id FROM range(100) ORDER BY id LIMIT 50 OFFSET 1` | 2 | `1, 2, 3` | `26, 20, 3` | That is the change the guard is there to prevent, and the config doc says these views are never converted. ```suggestion case LocalLimit(_, child) => hasTopLevelSort(child) case Offset(_, child) => hasTopLevelSort(child) ``` With it applied, both bodies give 0 repartitions and `1, 2, 3`, and `ConvertViewToMaterializedCTEQuerySuite` stays green. Adding the two bodies to the ORDER BY test would pin it. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/ConvertViewToMaterializedCTE.scala: ########## @@ -0,0 +1,219 @@ +/* + * 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 in `FinishAnalysis`, immediately before `EliminateView`: after `EliminateView` + * no `View` nodes remain and every reference site holds an independent copy of the view's plan. + * + * A converted definition always sets `forceSkipInline = true`; 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. + * + * 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] + + val rewritten = plan.transformUpWithSubqueries { + case v: View if qualifiedIdentifiers.contains(v.desc.identifier) => + defByGroup.get(v.desc.identifier) match { + case Some(cteDef) => + // 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) + 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. + val cteDef = CTERelationDef(v.child, forceSkipInline = true) Review Comment: **Finding 7.** The rule decides from the number of `View` occurrences, but not every occurrence survives the rewrite as a reference. Then a definition ends up with one live reference, and `forceSkipInline` still gives it a shuffle that nothing reuses. I measured two shapes on this head: - **Nested views.** In `nested view is converted together with the view it references`, `v1` occurs once inside each `v2`, so both views qualify. The second `v2` occurrence is replaced by `Project(..., ref)` at line 114, which drops its body together with its `v1` occurrence. That leaves `v1`'s definition with one reference, inside `v2`'s definition. The executed plan (AQE on and off, broadcast off) has 2 round-robin shuffles and 1 reused exchange. One is `v2`'s shared shuffle. The other is `v1`'s shuffle underneath it, which runs once and saves nothing. The test's 4 repartitions is that shuffle counted once per `v2` site. - **A view that a dead CTE also mentions.** `WITH c AS (SELECT id FROM v) SELECT id FROM v WHERE id < 10` reads `v` once. But the unreferenced `c` still holds a second occurrence when this rule runs, so the rule converts it. The executed plan gains a round-robin shuffle over a single reference. Both costs are opt-in. But nested views are common in the ETL setups the description targets. To cover both shapes, count the references to each new definition after the rewrite, and set `forceSkipInline` only where there are at least two. `InlineCTE` already does this kind of counting: it discounts references that come from unreferenced definitions (`InlineCTE.scala:231`). With `forceSkipInline = false`, `InlineCTE` would then flatten the single-reference ones back. -- 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]
