peter-toth commented on code in PR #58229: URL: https://github.com/apache/spark/pull/58229#discussion_r4480461831
########## 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: **Finding 12.** A `TABLESAMPLE` without `REPEATABLE` analyzes to `Sample(seed = None)`, and the seed is drawn per exec node at execution time (`Sample.resolveSeed`, called from `SampleExec`). `Sample` has no expressions, so `v.child.deterministic` is `true`. Both occurrences also canonicalize equal, so the view converts, and every reference then reads one shared sample instead of drawing its own. I ran it on `44c52ac` over a 20000-row table `t`, with AQE and broadcast off: ```sql CREATE VIEW vs AS SELECT id FROM t TABLESAMPLE (50 PERCENT); SELECT count(DISTINCT id), count(*) FROM (SELECT id FROM vs UNION ALL SELECT id FROM vs); ``` | | run 1 | run 2 | |---|---|---| | config off | `[14970, 19864]` | `[15047, 20052]` | | config on | `[9998, 19996]` | `[10060, 20120]` | The same view with `REPEATABLE (7)` gives `[10005, 20010]` both ways. The existing TABLESAMPLE test passes only because its inner view has an `ORDER BY`, which `hasMainRegionSort` rejects first. Treating a seedless `Sample` as non-deterministic closes this: ```suggestion v.child.deterministic && v.child.collectWithSubqueries { case s: Sample if s.seed.isEmpty => s }.isEmpty && ``` With it, the seedless view stays unconverted (about 15000 distinct ids either way), the `REPEATABLE` one still converts, and both suites pass. A test with a plain `TABLESAMPLE (50 PERCENT)` view that asserts `countRepartitions(query) == 0` would pin it. ########## 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 Review Comment: **Finding 13.** The guards look only at the view body. The shuffle that `ReplaceCTERefWithRepartition` puts above each reference also affects what the consuming query evaluates on top of it. `input_file_name()`, `input_file_block_start()` and `input_file_block_length()` read the file of the current input split, and after the round-robin shuffle there is none. I ran this on `44c52ac`, with `v` defined as `SELECT id, k FROM t` over a parquet table: ```sql SELECT input_file_name() <> '', id FROM v WHERE id < 2 UNION ALL SELECT input_file_name() <> '', id FROM v WHERE id < 2 ``` With the config off every row is `true`. With it on every row is `false`, and nothing fails. Declining plans that use these functions fixes it, with `InputFileName`, `InputFileBlockStart` and `InputFileBlockLength` added to the expressions import: ```scala if (occurrences.length < 2) return plan val readsInputFile = plan.collectWithSubqueries { case p => p }.exists(_.expressions.exists( _.exists { case _: InputFileName | _: InputFileBlockStart | _: InputFileBlockLength => true case _ => false })) if (readsInputFile) return plan ``` With it, the query above returns `true` for every row with the config on, and both suites pass. It declines the whole query, which is coarse. A narrower check would look only above the references. Either way, a test with the query above would pin it. ########## 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 Review Comment: **Finding 14.** Since finding 1, canonical equality of the bodies is the guard against divergent occurrences. Canonicalization drops two things in which two occurrences can still differ: - **The file listing.** The relation keeps its file index, but `InMemoryFileIndex` compares equal by root paths. - **Hints.** `ResolvedHint.doCanonicalize` returns `child.canonicalized`. The file-listing case, run on `44c52ac` with AQE off: ```scala spark.range(0, 10).write.parquet(path) sql(s"CREATE OR REPLACE TEMP VIEW pv AS SELECT id FROM parquet.`$path`") val held = spark.table("pv") spark.range(100, 105).write.mode("append").parquet(path) held.union(spark.table("pv")).count() // config off: 25, config on: 20 ``` The two occurrences group, and the first one's body serves both, so one side reads the other's listing. The hint case: `held` comes from `v = SELECT /*+ BROADCAST(t) */ id, k FROM t`, then `v` is replaced without the hint. In `big.join(held, "id").join(spark.table("v"), "id")` only the `held` join carries `BROADCAST` with the config off, and both do with it on, because the comment at line 125 assumes every occurrence shares one set of hints. Comparing the hints of every occurrence in `qualifies` closes that half. I ran it, and both suites pass: ```scala private def hintsOf(v: View): Seq[HintInfo] = v.child.collectWithSubqueries { case h: ResolvedHint => h.hints } ``` with `hintsOf(v) == hintsOf(first)` added to the conditions. I haven't found a fix for the file listing that works from catalyst. Canonicalization folds the listing away, and catalyst code can't read what a file relation lists. The comparison needs input from sql/core, which can, for example the relations' input files. For context, the stale listing also misleads exchange reuse on base: `held.join(spark.table("pv"), Seq("id"), "full_outer").count()` returns 10 instead of 15 there, with AQE on and off. The conversion extends that to shapes with no exchange, like the `UNION ALL` above, which base answers correctly. ########## 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: **Finding 15.** The conversion trades recomputation for a shuffle plus exchange reuse. With `spark.sql.exchange.reuse=false` nothing deduplicates the added round-robin shuffles, so the body still runs once per reference and the conversion is pure overhead. With `v` defined as `SELECT id, k FROM t`, AQE and broadcast off, `SELECT count(*) FROM v a JOIN v b ON a.id = b.id` plans 3 shuffles with the config off and 5 with it on, with no reuse in either. A one-line gate fixes it: ```suggestion if (!SQLConf.get.getConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE) || !SQLConf.get.exchangeReuseEnabled) return plan ``` With it, the reuse-off self-join keeps its 3 shuffles and both suites pass. ########## 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) Review Comment: **Finding 16.** `MergeSubplans.apply` returns any plan that contains a CTE unchanged (`MergeSubplans.scala:164`). The definitions attached here stay in the plan until `ReplaceCTERefWithRepartition`, which runs later. So once a view converts, scalar-subquery merging is off for the whole query, not only around the view. For example, in `SELECT (SELECT count(*) FROM t WHERE k = 1), (SELECT max(id) FROM t WHERE k = 1), (SELECT count(*) FROM v a JOIN v b ON a.id = b.id)` the two subqueries over `t` stop merging once `v` converts. Could the config doc say so? A follow-up could let `MergeSubplans` work with an existing `WithCTE` rather than bail out. -- 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]
