boobpoop opened a new issue, #57109: URL: https://github.com/apache/spark/issues/57109
# Summary When a persisted VIEW is referenced in both branches of a UNION ALL through a CTE, the shuffle of the underlying table that the VIEW scans should be reused across the two branches. Instead, ReusedExchange fails to match the two equivalent exchanges, and the table is scanned and shuffled twice. This is a performance regression introduced by the SPARK-39887 fix (PRs [#37334](https://github.com/apache/spark/pull/37334) and follow-up [#37534](https://github.com/apache/spark/pull/37534)), which added a first-child protection to RemoveRedundantAliases for Union. That protection is required for correctness and must not be removed, but it produces a structural asymmetry between the two UNION branches that defeats ReuseExchange. # Root Cause (one line) SPARK-39887's Union first-child protection in RemoveRedundantAliases preserves the view's schema-compat Alias on the first branch but allows the second branch's no-op Alias to be stripped, leaving the two ProjectExecs structurally asymmetric — one carries an Alias node, the other a bare AttributeReference — so their canonical forms differ and ReuseExchange cannot match them. # Root Cause Chain (4 steps) ## Step 1 — VIEW construction (SPARK-34269) When a persisted VIEW is resolved, each output column is wrapped in a schema-compatibility Alias: `Alias(UpCast(col, field.dataType), field.name)(explicitMetadata = Some(field.metadata))` So the view's Project project list contains Alias(a#scan AS a#catalog) where #scan is the ExprId of the underlying scan column and #catalog is the fixed ExprId from the view's catalog schema. These two ExprIds are different — it is a genuine rename, not a no-op. ## Step 2 — InlineCTE Both UNION branches reference the same cte body, so the two branches share identical ExprIds → Union.duplicateResolved = false. ## Step 3 — DeduplicateRelations DeduplicateRelations resolves the duplicate by wrapping the non-first branch with an outer Project(Alias(attr, attr.name)), and via newInstance() + rewriteAttrs it rewrites the view's inner schema-compat Alias so that its child ExprId becomes equal to its output ExprId — turning it into a no-op Alias(a#X AS a#X). ## Step 4 — RemoveRedundantAliases (SPARK-39887 first-child protection) This is where the asymmetry is produced. The SPARK-39887 fix added special Union handling: ``` case u: Union => var first = true plan.mapChildren { child => if (first) { first = false removeRedundantAliases(child, excluded ++ child.outputSet) // first child PROTECTED } else { removeRedundantAliases(child, excluded -- u.children.head.outputSet) // non-first NOT protected } } ``` Result: the two branches are structurally asymmetric — Branch1 carries an Alias node (1 alias), Branch2 has none (0 aliases, bare AttributeReference) — even though they are semantically equivalent. ## Step5 -- ReuseExchangeAndSubquery So Branch1 and Branch2 produce different ProjectExec canonical forms, which makes the upstream Exchange canonical forms differ. Both ReuseExchangeAndSubquery (non-AQE, keyed on canonicalized) and AQE stageCache (keyed on Exchange.canonicalized) fail to match them. The same table shuffle is computed twice instead of being reused. # Reproduction ``` CREATE TABLE t1 (a BIGINT, b INT, c INT) USING parquet; CREATE TABLE t2 (a BIGINT, d BIGINT) USING parquet; INSERT INTO t1 VALUES (1, 2, 3), (4, 5, 6); -- c=6 is filtered out by the view INSERT INTO t2 VALUES (1, 7), (4, 8); CREATE OR REPLACE VIEW v AS SELECT a, b FROM t1 WHERE c <> 6; WITH cte AS (SELECT ta.d, tv.b FROM t2 ta JOIN v tv ON ta.a = tv.a) SELECT d, b, count(1) FROM cte GROUP BY 1, 2 UNION ALL SELECT d, b, count(1) FROM cte GROUP BY 1, 2; ``` ## Expected vs Actual Expected: the t1 shuffle is reused across the two UNION branches; the final plan contains a ReusedExchangeExec. Actual: no ReusedExchangeExec exists; t1 is scanned and shuffled twice. # Fix Fix it in the canonicalization layer rather than in RemoveRedundantAliases. Make the two structurally asymmetric but semantically equivalent forms canonicalize equal, so that ReuseExchange can match them — without touching any of the four rules above and without risking correctness. Concretely: a no-op rename Alias(attr, attr.name) (same condition RemoveRedundantAliases uses to strip it: name == attr.name and matching metadata) can be folded into the bare attr during ProjectExec canonicalization, so that Alias(a#scan AS a#catalog) (Branch1, preserved) and AttributeReference(a#X) (Branch2, stripped) both canonicalize to AttributeReference(ExprId(position)) where position is the column's index in child.output. Because the underlying scan column sits at the same position in both branches, the two ProjectExec canonical forms become identical, and the upstream Exchange becomes reusable. # Request for Feedback I implemented a version of the above approach on Spark 4.2, and it fixes the issue. I would appreciate any guidance or feedback. -- 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]
