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: [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]
