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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
