peter-toth commented on code in PR #58229:
URL: https://github.com/apache/spark/pull/58229#discussion_r4217883538
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -353,6 +353,7 @@ abstract class Optimizer(catalogManager: CatalogManager)
EliminateResolvedHint,
EliminateSubqueryAliases,
EliminatePipeOperators,
+ ConvertViewToMaterializedCTE,
Review Comment:
**Finding 8.** `FinishAnalysis` runs its rules on the main plan first, and
only then on each subquery plan (`Optimizer.scala:377-384`). So when this rule
runs, `EliminateResolvedHint`, `EliminateSubqueryAliases` and
`EliminatePipeOperators` have cleaned the main plan but no subquery plan yet.
The rule takes the definition body from whichever occurrence
`transformUpWithSubqueries` visits first. If that occurrence sits inside a
subquery, its body still holds those nodes. The body then moves into the
top-level `WithCTE`, where the three rules have already run, so nothing removes
them.
I ran this on this head, with `t` a parquet table:
```sql
CREATE TEMP VIEW v AS SELECT id, k FROM t WHERE k < 4;
SELECT (SELECT count(*) FROM v) AS c, (SELECT max(id) FROM v) AS m;
```
- Conversion off returns `[8,13]`.
- Conversion on fails with `[INTERNAL_ERROR] The Spark SQL phase planning
failed`, caused by `No plan for SubqueryAlias spark_catalog.default.t`.
- A view body with a `BROADCAST` hint keeps its `ResolvedHint` node the same
way.
- `SELECT count(*) FROM v WHERE id > (SELECT avg(id) FROM v)` works, because
the main-plan occurrence is visited first.
The scalar-subquery test in `ConvertViewToMaterializedCTESuite` doesn't
catch this because its body has no `SubqueryAlias`.
Running the rule first in the batch fixes it. Wherever the body comes from,
it then sits in the top-level plan before the three rules run:
```suggestion
ConvertViewToMaterializedCTE,
EliminateResolvedHint,
EliminateSubqueryAliases,
EliminatePipeOperators,
```
With that change both the plain and the hinted view return the same rows as
with conversion off, and `ConvertViewToMaterializedCTEQuerySuite` stays green.
The class doc at `ConvertViewToMaterializedCTE.scala:33` would need the new
position. The query above in `ConvertViewToMaterializedCTEQuerySuite` would pin
it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/ConvertViewToMaterializedCTE.scala:
##########
@@ -0,0 +1,274 @@
+/*
+ * 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 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]
+
+ 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. `forceSkipInline` is decided
after the
+ // rewrite, by the number of surviving references.
+ val cteDef = CTERelationDef(v.child)
+ 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 {
+ // 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 &&
+ !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)
+ // ORDER BY ... OFFSET n wraps the Sort in Offset, possibly under the two
limit nodes
+ case Offset(_, child) => hasTopLevelSort(child)
Review Comment:
**Finding 9.** `hasTopLevelSort` now looks through `Offset`, but it still
stops at two nodes that can sit above the `Sort`:
- A view whose body has a `WITH` clause analyzes to `WithCTE(Sort(...),
defs)` under the view's cast `Project`.
- A view that selects from another view reaches the inner view's `View`
node, so an `ORDER BY` in the inner view is not seen.
Both reach `case _ => false`, so the view converts. I ran your ORDER BY
test's query, `SELECT t1.id FROM v t1 JOIN v t2 ON t1.id = t2.id LIMIT 3`, on
this head:
| view body | repartitions | rows, conversion off | rows, conversion on |
|---|---|---|---|
| `WITH c AS (SELECT id FROM range(100)) SELECT id FROM c ORDER BY id DESC`
| 2 | `0, 1, 2` | `14, 19, 61` |
| `SELECT id FROM v1`, where `v1` is `SELECT id FROM range(100) ORDER BY id
DESC` | 2 | `0, 1, 2` | `14, 19, 61` |
`(SELECT id FROM v LIMIT 3) UNION ALL (SELECT id FROM v LIMIT 3)` changes
the same way. With conversion off each branch returns `97, 98, 99`, with it on
other rows.
```suggestion
case Offset(_, child) => hasTopLevelSort(child)
case WithCTE(child, _) => hasTopLevelSort(child)
case v: View => hasTopLevelSort(v.child)
```
With it applied, both bodies give 0 repartitions and the same rows as with
conversion off, and both suites stay green. Adding the two bodies to the ORDER
BY test would pin it.
--
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]
