You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "xumingming (via GitHub)" <gi...@apache.org> on 2026/08/23 15:25:31 UTC
[PR] [SPARK-58950][SQL] Materialize repeated view references as one CTE [spark]
xumingming opened a new pull request, #58229:
URL: https://github.com/apache/spark/pull/58229
### What changes were proposed in this pull request?
Introduce a new optimizer rule, `ConvertViewToMaterializedCTE`, that runs in
`FinishAnalysis` (immediately before `EliminateView`). When the same view is
referenced two or more times in a query, the rule groups the occurrences (by
catalog identity plus canonicalized body) and rewrites each qualifying group
into a single `forceSkipInline` `CTERelationDef` with one `CTERelationRef`
per occurrence. Downstream, `ReplaceCTERefWithRepartition` wraps each
reference site with a shuffle boundary and exchange reuse deduplicates the
computation, so the view's body is evaluated once instead of once per
reference.
Details worth noting for reviewers:
- Occurrences are grouped by `(identifier, canonicalized body)`; the bottom-up
rewrite matches by identifier, so views nested inside another view's body
convert too, with definitions emitted in topological order (referenced
definitions first).
- Views whose bodies contain internally correlated subqueries
(e.g. `t WHERE x IN (SELECT y FROM s WHERE s.k = t.k)`) convert as well: a
view body is analyzed standalone at creation, so correlated references
always resolve inside the body and never escape the converted definition.
- The rule is gated behind a new config,
`spark.sql.optimizer.convertViewToMaterializedCTE`, off by default, named
per the configuration naming guideline in `ConfigEntry.scala`.
- Eligibility is conservative: the view must be deterministic and
non-streaming, all occurrences must agree on the effective view SQL config
and output schema, and single-reference views are skipped.
### Why are the changes needed?
In ETL workloads, views are almost always costly: they encapsulate large
scans, joins, and aggregations over big base tables. When such a view is
referenced more than once in one query, the current plan inlines the view
body at every reference site, so the costly body executes once per
reference. Materializing the view once and sharing the result across all
references avoids that repeated work, for example:
```sql
CREATE VIEW serving_log_view AS
SELECT user_id, tier, COUNT(*) AS requests
FROM serving_log GROUP BY user_id, tier;
SELECT t1.user_id, t1.requests
FROM serving_log_view t1 JOIN serving_log_view t2
ON t1.user_id = t2.user_id AND t1.tier = 'priority';
```
Today the whole `serving_log` scan and aggregation behind the view run twice.
### Does this PR introduce _any_ user-facing change?
Yes, in the form of a new, opt-in configuration:
`spark.sql.optimizer.convertViewToMaterializedCTE` (default `false`). When
enabled, repeated references to the same view are materialized as one CTE
with a shuffle boundary per reference site and exchange reuse; when disabled
(the default), the plan is unchanged.
### How was this patch tested?
Added and ran unit tests in both suites:
- `ConvertViewToMaterializedCTESuite` (catalyst, 16 tests) - rule-level
coverage: conversion of self-joined views, single-reference and
non-deterministic/streaming/config-mismatch/schema-mismatch rejection,
references inside scalar subqueries, internally correlated bodies,
view-over-view (nested) conversion with topological definition order,
survival of converted definitions through the `InlineCTE` batch, and
idempotency.
- `ConvertViewToMaterializedCTEQuerySuite` (sql/core, 6 tests) - end-to-end
coverage: results identical to the un-converted baseline, one shuffle
boundary per reference site, non-deterministic views not converted,
`INSERT INTO` from a self-referenced view, and internally correlated and
nested views executing correctly with the config enabled.
Commands:
```
build/sbt -Phive 'catalyst/testOnly *ConvertViewToMaterializedCTESuite' \
'sql/testOnly *ConvertViewToMaterializedCTEQuerySuite'
build/sbt 'catalyst/scalastyle' 'catalyst/Test/scalastyle' 'sql/scalastyle' 'sql/Test/scalastyle'
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Pi
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58950][SQL] Materialize repeated view references as one CTE [spark]
Posted by "xumingming (via GitHub)" <gi...@apache.org>.
xumingming commented on PR #58229:
URL: https://github.com/apache/spark/pull/58229#issuecomment-5775428658
@cloud-fan @peter-toth Thanks for the thorough review, really helpful. I have made the following changes:
- Use view identifier to group and then check whether all the canonicalized view body in the same group are the same, so it refuses views with different view bodies.
- For the View with inner CTE case, I added logic to normalize the cte id/refIds, so views with CTEs can be converted too.
- Fix a Subquery related bug: make the rule only fire when the plan root is NOT a subquery, otherwise the optimization would throw.
- Added a conservative check: if the views contains a top level 'ORDER BY', the rule will not fire, because it would change the result if the view is used like 'SELECT * FROM view LIMIT 10'. Note that this is not strictly a correctness bug, there is no guarantee about the order of the result when no outer 'ORDER BY' is specified, but let's be conservative to not surprise users.
- Added checks that the view body is computed only once when materialized.
- Added more tests for the corresponding changes.
- Also fixed outdated docs, removed trivial nits like tests count.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58950][SQL] Materialize repeated view references as one CTE [spark]
Posted by "xumingming (via GitHub)" <gi...@apache.org>.
xumingming commented on PR #58229:
URL: https://github.com/apache/spark/pull/58229#issuecomment-5537849622
Gentle ping.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58950][SQL] Materialize repeated view references as one CTE [spark]
Posted by "xumingming (via GitHub)" <gi...@apache.org>.
xumingming commented on PR #58229:
URL: https://github.com/apache/spark/pull/58229#issuecomment-6061172491
Hi @peter-toth, thanks for both findings — both fixed on the branch.
**Finding 8:** Done as suggested — `ConvertViewToMaterializedCTE` now runs first in the `FinishAnalysis` batch; your repro is added to the query suite (plus a hinted variant, pinned on the plan), and the class doc is updated.
**Finding 9:** We replaced the check rather than adding the two cases, and would like to discuss.
Before patching we swept other candidate wrappers (your repro, bodies over `v1 = ... ORDER BY id DESC`) — with your two cases already applied, these **still convert and change rows**:
```
SELECT DISTINCT id FROM v1 off: 52,53,61 on: 19,61,14
SELECT id FROM v1 TABLESAMPLE(50 PERCENT) off: 86,83,82 on: 84,85,82
WITH c AS (SELECT id FROM t ORDER BY id DESC) SELECT id FROM c off: 50,51,52 on: 19,61,14
(SELECT id FROM t ORDER BY id DESC) UNION ALL (...) off: 99,99,98 on: 90,90,86
SELECT v1.id FROM v1 JOIN t ON v1.id = t.id off: 50,51,52 on: 67,65,73
```
So the pass-through list is whack-a-mole (`Offset` was brand-new in 4.0). Since the shuffle sits above the whole definition, we went to the semantic criterion instead — reject conversion whenever any `Sort` is reachable through operator `children` from the body root:
```scala
private def hasMainRegionSort(plan: LogicalPlan): Boolean =
plan.isInstanceOf[Sort] || plan.children.exists(hasMainRegionSort)
```
Subquery-expression sorts are not `children`, are skipped, and genuinely can't affect the body's exposed rows (pinned by a test: `IN (SELECT ... ORDER BY ...)` bodies still convert). All shapes above are covered by this one criterion.
Cost: a few safe-by-luck shapes don't convert; benefit: the silent-wrong-results class is closed for future operators, not just today's list. Both suites green, your two bodies + the sweep shapes added as regression tests.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58950][SQL] Materialize repeated view references as one CTE [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58229:
URL: https://github.com/apache/spark/pull/58229#discussion_r4057248704
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/ConvertViewToMaterializedCTE.scala:
##########
@@ -0,0 +1,190 @@
+/*
+ * 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
+ val occurrences = plan.collectWithSubqueries { case v: View => v }
+ if (occurrences.length < 2) return plan
+
+ // Group occurrences of the same view by their canonicalized body and captured SQL
+ // configs. Occurrences of one view differ only in renewed expression ids, which
+ // canonicalization normalizes away.
+ val qualifiedGroups = occurrences.groupBy(groupKey).values.filter(qualifies)
+ if (qualifiedGroups.isEmpty) return plan
+ // Match by identifier during the transform, not by the full group key: the key embeds
+ // the canonicalized body, and by the time an outer view is visited in the bottom-up
+ // traversal, nested views inside its body have already been rewritten into
+ // `CTERelationRef`s, so the body no longer canonicalizes to the key computed here.
+ // An identifier mapping to more than one qualified group would mean occurrences whose
+ // canonicalized bodies diverge; refuse conversion rather than rewrite all of them
+ // against whichever definition happens to be visited first.
+ val qualifiedIdentifiers = qualifiedGroups
Review Comment:
**Non-blocking (P2):** `qualifiedIdentifiers` is computed after non-qualifying groups have been discarded, but the later transform rewrites every `View` with an admitted identifier. If two occurrences form a qualifying group and a third same-identifier occurrence has a different or nondeterministic body, that singleton is excluded here but can still be rebound to the pair's CTE definition, changing the query result. This is reachable when repeated temporary-view lookups observe a replacement during analysis. Please admit an identifier only when every occurrence belongs to the one qualified group, and cover the qualifying-pair plus divergent-singleton case.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58950][SQL] Materialize repeated view references as one CTE [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #58229:
URL: https://github.com/apache/spark/pull/58229#discussion_r4057836384
##########
sql/core/src/test/scala/org/apache/spark/sql/ConvertViewToMaterializedCTEQuerySuite.scala:
##########
@@ -0,0 +1,123 @@
+/*
+ * 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.RepartitionByExpression
+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` 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 {
+ 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 =
+ spark.sql(query).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)
Review Comment:
**Finding 3.** This is as far as either suite goes: two `RepartitionByExpression` in the logical plan. The comment on the line above says identical shuffles are then reused, and that reuse is the point of the change, but nothing asserts it. A regression that left the two shuffles unreused would keep every test in this PR green while making every converted query strictly slower than it was.
It does hold today. On `f681edc`, over a 2000-row parquet `serving_log` and the query from the PR description, with AQE both on and off:
| | parquet scans | rows read | `ReusedExchange` |
|---|---|---|---|
| conversion off | 2 | 4000 | 0 |
| conversion on | 1 | 2000 | 1 |
Worth pinning, roughly:
```scala
test("the view body is computed once") {
withSelfJoinedView {
withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") {
val df = spark.sql(selfJoinQuery)
df.collect()
def walk(p: SparkPlan): Seq[SparkPlan] = p match {
case a: AdaptiveSparkPlanExec => a +: walk(a.executedPlan)
case q: QueryStageExec => q +: walk(q.plan)
case r: ReusedExchangeExec => Seq(r)
case other => other +: other.children.flatMap(walk)
}
assert(walk(df.queryExecution.executedPlan).count(_.isInstanceOf[ReusedExchangeExec]) == 1)
}
}
}
```
The `QueryStageExec` hop matters. With AQE on the final plan keeps the reused exchange behind a query stage, and a plain `collect` on `executedPlan` walks straight past it and finds nothing.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ConvertViewToMaterializedCTESuite.scala:
##########
@@ -0,0 +1,428 @@
+/*
+ * 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("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
+ // of the same view whose canonicalized bodies diverge. Each pair qualifies on its
+ // own, but rewriting all four occurrences against whichever definition happens to
+ // be visited first would re-bind the divergent pair positionally against the wrong
+ // schema, so the identifier must be skipped entirely.
+ val a1 = attr("a", 100)
+ val a2 = attr("a", 200)
+ val a3 = attr("a", 300)
+ val a4 = attr("a", 400)
+ def bodyWithOne(a: AttributeReference): LogicalPlan =
+ Project(Seq(a.as("x")), LocalRelation(Seq(a)))
+ def bodyWithTwo(a: AttributeReference): LogicalPlan =
+ Project(Seq(a.as("x"), a.as("y")), LocalRelation(Seq(a)))
+ val left = Join(
+ tempView("v", bodyWithOne(a1)), tempView("v", bodyWithOne(a2)),
+ Inner, None, JoinHint(None, None))
+ val right = Join(
+ tempView("v", bodyWithTwo(a3)), tempView("v", bodyWithTwo(a4)),
+ Inner, None, JoinHint(None, None))
+ val query = Join(left, right, Inner, None, JoinHint(None, None))
+ comparePlans(Optimize.execute(query), query)
+ }
+ }
+
+ test("converts views referenced inside scalar subqueries") {
+ withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") {
+ val (v1, v2, _) = sameViewTwice(
+ "v", Seq(attr("a", 100), attr("b", 101)),
+ (as: Seq[AttributeReference]) => simpleBody(as(0), as(1)))
+ val sq1 = ScalarSubquery(v1)
+ val sq2 = ScalarSubquery(v2)
+ val query = OneRowRelation().select(sq1.as("s1"), sq2.as("s2"))
+
+ val optimized = Optimize.execute(query)
+
+ // The definitions are attached at the top-level scope while the references live
+ // inside the scalar subqueries.
+ val WithCTE(mainPlan, cteDefs) = optimized
+ assert(cteDefs.length == 1)
+ assert(cteDefs.head.forceSkipInline)
+ val refs = optimized.collectWithSubqueries { case r: CTERelationRef => r }
+ assert(refs.length == 2)
+ assert(mainPlan.collect { case r: CTERelationRef => r }.isEmpty,
+ "references must live inside the subqueries, not in the main plan")
+ assert(optimized.output == query.output)
+ }
+ }
+
+ test("converted definition survives the Inline CTE batch") {
+ 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 = OptimizeWithInlineCTE.execute(query)
+
+ val withCTEs = optimized.collect { case w: WithCTE => w }
+ assert(withCTEs.nonEmpty,
+ "deterministic single-ref CTEs get inlined, but the converted def must survive")
+ val defs = optimized.collect { case d: CTERelationDef => d }
+ assert(defs.length == 1)
+ assert(optimized.output == query.output)
+ }
+ }
+
+ test("does not convert views whose body contains non-deterministic inner CTEs") {
+ withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") {
+ // The view body itself evaluates a multi-ref non-deterministic inner CTE, so the view
+ // yields different results per evaluation. Converting it to a compute-once CTE would
+ // change query results, hence it must stay unconverted.
+ val (v1, v2) = sameViewTwice(
+ "v", () => nestedCteBody(987654321L, deterministic = false))
+ val query = Join(v1, v2, Inner, None, JoinHint(None, None))
+ comparePlans(Optimize.execute(query), query)
+ }
+ }
+
+ test("converts a view whose body contains a surviving deterministic inner CTE") {
+ withSQLConf(SQLConf.CONVERT_VIEW_TO_MATERIALIZED_CTE.key -> "true") {
+ // Both occurrences share the same inner CTE definition id, mirroring how the
Review Comment:
**Finding 2.** This comment says the analyzer duplicates a view body while keeping its inner CTE ids intact. It does not for a SQL view, and because of that a view whose body has a `WITH` clause never converts.
`CTESubstitution` runs on every re-analysis of the body and takes a fresh id from `CTERelationDef.newId`, so the two occurrences differ in a field canonicalization does not normalize. Measured on `f681edc` over a 20-row `t`:
| view | inner def ids across the two occurrences | distinct canonicalized bodies | `RepartitionByExpression` with the config on |
|---|---|---|---|
| `CREATE TEMP VIEW plain AS SELECT id, k FROM t WHERE k < 4` | `[]`, `[]` | 1 | 2 |
| `CREATE TEMP VIEW withcte AS WITH c AS (...) SELECT id, k FROM c` | `[2]`, `[3]` | 2 | **0** |
| the same body as a persistent `CREATE VIEW` | `[17]`, `[18]` | 2 | **0** |
The two occurrences land in two groups of one, neither reaches `occs.length >= 2`, and the rule returns the plan unchanged. It converts only when the view stores an analyzed plan, which I confirmed both for `spark.sql(...).createOrReplaceTempView` and for `spark.sql.legacy.storeAnalyzedPlanForView=true`: one distinct body, and it converts.
That outcome is safe rather than wrong, but it is a silent gap and the hand-built id here is what hides it. Worth replacing this comment with one that says a SQL view's inner CTE ids differ per occurrence so the rule skips it, and adding an end-to-end case pinning 0 repartitions for a `WITH`-bodied view.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/ConvertViewToMaterializedCTE.scala:
##########
@@ -0,0 +1,190 @@
+/*
+ * 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
+ val occurrences = plan.collectWithSubqueries { case v: View => v }
+ if (occurrences.length < 2) return plan
+
+ // Group occurrences of the same view by their canonicalized body and captured SQL
+ // configs. Occurrences of one view differ only in renewed expression ids, which
+ // canonicalization normalizes away.
+ val qualifiedGroups = occurrences.groupBy(groupKey).values.filter(qualifies)
+ if (qualifiedGroups.isEmpty) return plan
+ // Match by identifier during the transform, not by the full group key: the key embeds
+ // the canonicalized body, and by the time an outer view is visited in the bottom-up
+ // traversal, nested views inside its body have already been rewritten into
+ // `CTERelationRef`s, so the body no longer canonicalizes to the key computed here.
+ // An identifier mapping to more than one qualified group would mean occurrences whose
+ // canonicalized bodies diverge; refuse conversion rather than rewrite all of them
+ // against whichever definition happens to be visited first.
+ val qualifiedIdentifiers = qualifiedGroups
Review Comment:
**Finding 1.** Measured on `f681edc`. I would put this at Blocking rather than P2: it is reachable from ordinary API calls and it returns wrong rows with no error.
`qualifies` requires `occs.length >= 2`, so a divergent singleton is dropped from `qualifiedGroups` before the one-group-per-identifier filter runs. The identifier then looks unambiguous and the transform rewrites every `View` carrying it. The same hole admits a second group that failed `qualifies` for any other reason: driving the rule directly with a deterministic pair and a non-deterministic pair of one identifier, all four occurrences are rebound to the deterministic definition and the `rand` is gone from the result plan.
Repro of the wrong answer. Three references to `v`, the first two sitting in a DataFrame that was analyzed before the view was replaced:
```scala
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(col("id"))
sql("CREATE OR REPLACE TEMP VIEW v AS SELECT id, id + 100 AS k FROM range(4)")
pair.join(spark.table("v"), "id").select(col("id"), col("k")).collect()
```
```
conversion off: [0,100] [1,101] [2,102] [3,103]
conversion on : [0,0] [1,1] [2,2] [3,3]
```
The third reference is rebound to the stale definition and nothing reports it. Widening the replacement instead (`SELECT id, id AS k, id * 10 AS z`, then selecting `z`) fails loudly rather than silently:
```
[PLAN_VALIDATION_FAILED_RULE_IN_BATCH] Rule ...ReplaceCTERefWithRepartition in batch
Replace CTE with Repartition generated an invalid plan: Aliases z#76L are dangling in
the references for plan: ...
```
Grouping by identifier first closes both shapes, and it lets `groupKey`, `groupKeyOf` and the `GroupKey` type go:
```scala
val qualifiedIdentifiers = occurrences.groupBy(_.desc.identifier).collect {
case (identifier, occs)
if occs.map(_.child.canonicalized).distinct.length == 1 && qualifies(occs) => identifier
}.toSet
if (qualifiedIdentifiers.isEmpty) return plan
```
I applied exactly that on `f681edc`. Both repros then agree with the un-converted plan, the non-deterministic pair keeps its `rand`, and `ConvertViewToMaterializedCTESuite` (17) plus `ConvertViewToMaterializedCTEQuerySuite` (6) stay green, the existing divergent-groups test included. A case for the qualifying-pair-plus-divergent-singleton shape is worth adding as well, since 2+2 is the only one pinned today.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58950][SQL] Materialize repeated view references as one CTE [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #58229:
URL: https://github.com/apache/spark/pull/58229#issuecomment-5395405182
I'm a bit reluctant to proceed with this PR. Spark doesn't have a cost model that we could use to estimate the benefits and costs of such materialization. That's why Spark currently materializes only multiply referenced, non-deterministic CTEs when materialization is needed for correctness.
@cloud-fan, any thoughts on this?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58950][SQL] Materialize repeated view references as one CTE [spark]
Posted by "xumingming (via GitHub)" <gi...@apache.org>.
xumingming commented on PR #58229:
URL: https://github.com/apache/spark/pull/58229#issuecomment-6053498344
@peter-toth Good catch on Finding 6 & 7, fixed.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58950][SQL] Materialize repeated view references as one CTE [spark]
Posted by "xumingming (via GitHub)" <gi...@apache.org>.
xumingming commented on PR #58229:
URL: https://github.com/apache/spark/pull/58229#issuecomment-5395660595
@peter-toth Thanks for the review. I agree with you that we don’t have a cost model to check whether materialization is the right call. Here we do not do any view materialization by default, only when user enabled the switch, we will do it. In production query we have many views that should be materialized to get better performance, so we want to add this ability.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58950][SQL] Materialize repeated view references as one CTE [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #58229:
URL: https://github.com/apache/spark/pull/58229#discussion_r4221085479
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/ConvertViewToMaterializedCTE.scala:
##########
@@ -0,0 +1,276 @@
+/*
+ * 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 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)
Review Comment:
**Finding 11.** Running this rule first in `FinishAnalysis` (the fix I suggested for finding 8) drops join hints written at the top of a view body.
`EliminateResolvedHint.extractHintsFromPlan` walks down through any unary node, `View` included. So today a hint like the one in `SELECT /*+ BROADCAST(t) */ id, k FROM t WHERE k < 4` reaches the outer join the view takes part in. This rule now runs before it and replaces the `View` with a `CTERelationRef`, which is a leaf. The walk stops at the reference. The hint stays in the definition body, where no join sits above it. The last step of `EliminateResolvedHint` (`EliminateResolvedHint.scala:50-53`) then drops it with a "join hint ... is not part of a join relation" warning.
I ran it on this head with that view, `spark.sql.autoBroadcastJoinThreshold=-1` and AQE off:
| query | conversion off | conversion on |
|---|---|---|
| `SELECT t1.id FROM v t1 JOIN v t2 ON t1.id = t2.id` | `broadcast` hint on both sides, 1 `BroadcastHashJoin` | no hints, 1 `SortMergeJoin` |
| `SELECT b.id FROM big b JOIN v t1 ON b.id = t1.id JOIN v t2 ON b.id = t2.id` | 2 `BroadcastHashJoin` | 2 `SortMergeJoin` |
With the old order (rule right before `EliminateView`), conversion keeps the hints and the broadcast joins in both queries.
One fix I tried takes the top-of-body hints off each occurrence and puts them back above its reference. The definition is built from the hint-free body. `EliminateResolvedHint` then pulls the hints into the outer join as before:
```scala
case None =>
val (hintFreeBody, _) = EliminateResolvedHint.extractHintsFromPlan(v.child)
val cteDef = CTERelationDef(hintFreeBody)
...
withBodyHints(v.child, CTERelationRef(...))
private def withBodyHints(body: LogicalPlan, site: LogicalPlan): LogicalPlan = {
val (_, hints) = EliminateResolvedHint.extractHintsFromPlan(body)
hints.foldRight(site)((h, acc) => ResolvedHint(acc, h))
}
```
The later-occurrence branch wraps its rebinding `Project` the same way. With it, both queries keep their hints and broadcast joins when converted, and both suites stay green. Declining views whose body carries such a hint would be a smaller fix. The self-join above makes a regression test. Assert the `Join` hints in the optimized plan, or the `BroadcastHashJoinExec` count.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1001,6 +1001,22 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val CONVERT_VIEW_TO_MATERIALIZED_CTE =
+ buildConf("spark.sql.optimizer.convertViewToMaterializedCTE.enabled")
+ .doc("When true, the optimizer rewrites multiple references to the same view " +
+ "into a single materialized CTE definition with multiple references, so the " +
+ "view's underlying plan is computed once (through exchange reuse) instead of " +
+ "once per reference. Views that contain non-deterministic expressions, " +
+ "streaming sources, or a top-level ORDER BY (whose ordering would be lost " +
+ "across the added shuffle boundary, the same as MATERIALIZED CTEs) are never " +
Review Comment:
**Finding 10.** The rule now rejects any `Sort` outside subquery expressions, not only a top-level one:
```suggestion
"streaming sources, or an ORDER BY outside subquery expressions (its order " +
"would be lost across the added shuffle boundary) are never " +
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58950][SQL] Materialize repeated view references as one CTE [spark]
Posted by "xumingming (via GitHub)" <gi...@apache.org>.
xumingming commented on PR #58229:
URL: https://github.com/apache/spark/pull/58229#issuecomment-6079424163
@peter-toth Thanks for all the thorough reviews, really helpful. BTW, the AI augmented review(I guess) result is very well organized and helpful, is it some script or skill or something?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58950][SQL] Materialize repeated view references as one CTE [spark]
Posted by "xumingming (via GitHub)" <gi...@apache.org>.
xumingming commented on PR #58229:
URL: https://github.com/apache/spark/pull/58229#issuecomment-5602439097
Add more cross engine info here, I investigated popular compute engine like PostgreSQL and DuckDB:
**PostgreSQL**
- < 12: **every CTE was an unconditional optimization fence — always materialized**, no exceptions.
- ≥ 12 (release notes (https://www.postgresql.org/docs/12/release-12.html)): flipped default — a WITH query is inlined if it's non-recursive, has no volatile functions, and is referenced exactly once. **Referenced ≥2 times → materialized by default.** Volatile or recursive → always materialized (can't be folded safely). Explicit MATERIALIZED/NOT MATERIALIZED keywords let you override in either direction.
**DuckDB**
- < 1.4: inlined by default (any reference count), which caused both duplicated computation and real correctness bugs.
**- ≥ 1.4 ("Andium", Sept 2025): flipped the other way — materialized by default.** Only inlines when: referenced once, no volatile function, no grouped aggregation, and no NOT MATERIALIZED override forbidding it. DuckDB explicitly cites both performance and correctness bugs as the reason. Same MATERIALIZED/NOT MATERIALIZED syntax as Postgres exists but docs discourage using it manually.
**Although for views all the engines inline by default. But from the performance point of view, IMO View and CTE are similar, they both captures a common shared snippet of SQL that could be referenced multiple times in a query. Add a switch to be able to materialize the View/CTE will give us real performance benefit**
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58950][SQL] Materialize repeated view references as one CTE [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58950][SQL] Materialize repeated view references as one CTE [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #58229:
URL: https://github.com/apache/spark/pull/58229#issuecomment-5750609724
The off-by-default switch and the documented shuffle tradeoff are sufficient for me on the performance side. Users enabling it are explicitly choosing materialization for workloads where repeated view computation is expected to cost more than the shuffle.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"conversation:5395405182","thread_id":"conversation:5395405182","verdict_sha256":"8b38ebc6f9add1f5914fc133f5e01e342df23c9c815d0003cf272bfaf0fea4b0"} -->
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58950][SQL] Materialize repeated view references as one CTE [spark]
Posted by "uros-b (via GitHub)" <gi...@apache.org>.
uros-b commented on PR #58229:
URL: https://github.com/apache/spark/pull/58229#issuecomment-5539259635
I think the overall direction here makes sense, and opt-in approach seems valid in this case, but @peter-toth's point above is also fair. Let's loop in @cloud-fan and consider whether to move forward with this
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58950][SQL] Materialize repeated view references as one CTE [spark]
Posted by "uros-b (via GitHub)" <gi...@apache.org>.
uros-b commented on PR #58229:
URL: https://github.com/apache/spark/pull/58229#issuecomment-5388143008
Thank you @xumingming! LGTM, adding @peter-toth for further review
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58950][SQL] Materialize repeated view references as one CTE [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #58229:
URL: https://github.com/apache/spark/pull/58229#issuecomment-5750609944
I think the opt-in direction is acceptable. The default remains unchanged, and the config documentation makes clear that materialization adds shuffles and is only beneficial when the repeated view subtree is sufficiently expensive.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"conversation:5539259635","thread_id":"conversation:5539259635","verdict_sha256":"8b38ebc6f9add1f5914fc133f5e01e342df23c9c815d0003cf272bfaf0fea4b0"} -->
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58950][SQL] Materialize repeated view references as one CTE [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #58229:
URL: https://github.com/apache/spark/pull/58229#issuecomment-5614643845
How about https://github.com/apache/spark/pull/58661?
cc @pan3793
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58950][SQL] Materialize repeated view references as one CTE [spark]
Posted by "pan3793 (via GitHub)" <gi...@apache.org>.
pan3793 commented on PR #58229:
URL: https://github.com/apache/spark/pull/58229#issuecomment-5615263383
> Add a switch to be able to materialize the View/CTE will give us real performance benefit
I'm not sure about the View part, but the CTE part does match the work I'm currently doing. (I use the PostgreSQL 18 behavior as the reference, wasn't aware that DuckDB also supports it and has a different behavior.)
> Spark doesn't have a cost model that we could use to estimate the benefits and costs of such materialization.
Fully agree with this. In addition, data source statistics are also unreliable/missing in many cases. As a result, many of Spark's built-in optimization rules are very conservative. That is why I am trying to add some HINT-like functionality, allowing users to control the behavior manually.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58950][SQL] Materialize repeated view references as one CTE [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #58229:
URL: https://github.com/apache/spark/pull/58229#issuecomment-6079147226
@cloud-fan, do you have any more comments?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58950][SQL] Materialize repeated view references as one CTE [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #58229:
URL: https://github.com/apache/spark/pull/58229#issuecomment-6095911909
Thanks @xumingming, yes, I have a custom review skill that focuses on adding exact repros to blocking issues at least. Also, findings are organized into categories, each one is identified by a unique number and contains a link to the inline thread. I believe this ease the review process for those who don't use agents.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58950][SQL] Materialize repeated view references as one CTE [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58950][SQL] Materialize repeated view references as one CTE [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #58229:
URL: https://github.com/apache/spark/pull/58229#issuecomment-6323366031
@cloud-fan, are you ok with merging this?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org