From fb466db388533a8fdcc1fdda6be1f0a6485933c0 Mon Sep 17 00:00:00 2001 From: AMC-hawk Date: Fri, 11 Sep 2026 20:13:34 +0530 Subject: [PATCH 1/6] [SPARK-59146][SQL] Retain qualified access to source columns affected by pipe SET The pipe SET operator is built as a star expansion that excludes the assigned column and appends a replacement of the same name, so the original attribute is dropped from the project list and a later qualified reference such as `t.a` has nothing to resolve to. This contradicts the documented behavior that table aliases keep referring to the original row values after an assignment. Retain the excluded attribute as hidden output on the Project that SET builds, via the existing Project.hiddenOutputTag mechanism that USING joins use for duplicated join keys. AddMetadataColumns splices it back only when referenced, so the output schema of SET and the plan for `|> SELECT *` are unchanged. The retention is flagged from the parser because pipe SET and regular `SELECT * EXCEPT` / `* REPLACE` produce an identical node at star-expansion time, and the latter must not be affected. --- docs/sql-pipe-syntax.md | 15 ++--- .../sql/catalyst/analysis/Analyzer.scala | 26 +++++++- .../sql/catalyst/analysis/unresolved.scala | 11 +++- .../sql/catalyst/parser/AstBuilder.scala | 7 ++- .../analyzer-results/pipe-operators.sql.out | 59 +++++++++++++++++++ .../sql-tests/inputs/pipe-operators.sql | 27 +++++++++ .../sql-tests/results/pipe-operators.sql.out | 53 +++++++++++++++++ 7 files changed, 188 insertions(+), 10 deletions(-) diff --git a/docs/sql-pipe-syntax.md b/docs/sql-pipe-syntax.md index 3d757db966239..024c77edf4683 100644 --- a/docs/sql-pipe-syntax.md +++ b/docs/sql-pipe-syntax.md @@ -289,14 +289,15 @@ VALUES (0), (1) tab(col) +---+ VALUES (0), (1) tab(col) -|> SET col = col * 2; +|> SET col = col * 2 +|> SELECT col, tab.col; -+---+ -|col| -+---+ -| 0| -| 2| -+---+ ++---+---+ +|col|col| ++---+---+ +| 0| 0| +| 2| 1| ++---+---+ ``` #### DROP diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala index 35b9052686dcf..df032ba79f9d4 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala @@ -49,7 +49,7 @@ import org.apache.spark.sql.catalyst.trees.AlwaysProcess import org.apache.spark.sql.catalyst.trees.CurrentOrigin.withOrigin import org.apache.spark.sql.catalyst.trees.TreePattern._ import org.apache.spark.sql.catalyst.types.DataTypeUtils -import org.apache.spark.sql.catalyst.util.{toPrettySQL, trimTempResolvedColumn, CharVarcharUtils, GeneratedColumn} +import org.apache.spark.sql.catalyst.util.{toPrettySQL, trimTempResolvedColumn, CharVarcharUtils, GeneratedColumn, MetadataColumnHelper} import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns._ // `View` is aliased to `V2View` to avoid clashing with the logical-plan `View` imported via // `org.apache.spark.sql.catalyst.plans.logical._`. @@ -1701,6 +1701,7 @@ class Analyzer( if (expanded.projectList.size < p.projectList.size) { checkTrailingCommaInSelect(expanded, starRemoved = true) } + retainExceptedColumnsAsHiddenOutput(p, expanded) expanded // If the filter list contains Stars, expand it. case p: Filter if containsStar(Seq(p.condition)) => @@ -2085,6 +2086,29 @@ class Analyzer( }.map(_.asInstanceOf[NamedExpression]) } + /** + * The SQL pipe SET operator is implemented as a star expansion that excludes the assigned + * column and appends a replacement of the same name. That drops the original attribute from + * the project list, which would also make it unreachable through its table alias, contradicting + * the documented behavior that table aliases keep referring to the original row values after an + * assignment. Retain the excluded attributes as hidden output instead, the same way USING joins + * hide their duplicated join keys (SPARK-59146). + */ + private def retainExceptedColumnsAsHiddenOutput(original: Project, expanded: Project): Unit = { + val retain = original.projectList.exists { + case s: UnresolvedStarExceptOrReplace => s.retainExceptedColumnsAsHidden + case _ => false + } + if (retain) { + val excepted = expanded.child.output.filterNot(expanded.outputSet.contains) + if (excepted.nonEmpty) { + expanded.setTagValue( + Project.hiddenOutputTag, + excepted.map(_.markAsQualifiedAccessOnly()) ++ expanded.child.metadataOutput) + } + } + } + /** * Returns true if `exprs` contains a [[Star]]. */ diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala index 79215115c7031..3cb95de2123db 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala @@ -640,11 +640,20 @@ trait UnresolvedStarBase extends Star with Unevaluable { * expressions removed by EXCEPT. If present, the length of this list must * be the same as the length of the EXCEPT list. This supports replacing * expressions instead of excluding them from the original SELECT list. + * + * @param retainExceptedColumnsAsHidden if true, the excluded attributes are kept as the hidden + * output of the enclosing [[Project]], so that they remain + * reachable through their table alias. The SQL pipe SET + * operator sets this, since it documents that table aliases + * keep referring to the original row values after an + * assignment. It stays false for SELECT * EXCEPT, where the + * excluded columns must not be reachable at all. */ case class UnresolvedStarExceptOrReplace( target: Option[Seq[String]], excepts: Seq[Seq[String]], - replacements: Option[Seq[NamedExpression]]) + replacements: Option[Seq[NamedExpression]], + retainExceptedColumnsAsHidden: Boolean = false) extends LeafExpression with UnresolvedStarBase { final override val nodePatterns: Seq[TreePattern] = Seq(UNRESOLVED_STAR_EXCEPT_OR_REPLACE) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala index 9a102f0d0b8d0..62981cc0cfec5 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala @@ -8158,11 +8158,16 @@ class AstBuilder extends DataTypeAstBuilder // Add an UnresolvedStarExceptOrReplace to exclude the SET expression name from the relation // and add the new SET expression to the projection list. // Use a PipeSelect expression to make sure it does not contain any aggregate functions. + // Retain the excluded source column as hidden output so that qualified references such as + // `t.a` keep returning the original row value after `|> SET a = ...` (SPARK-59146). val replacement = Alias(PipeExpression(target, isAggregate = false, PipeOperators.setClause), ident)() val projectList: Seq[NamedExpression] = Seq(UnresolvedStarExceptOrReplace( - target = None, excepts = Seq(Seq(ident)), replacements = Some(Seq(replacement)))) + target = None, + excepts = Seq(Seq(ident)), + replacements = Some(Seq(replacement)), + retainExceptedColumnsAsHidden = true)) // Add a projection to implement the SET operator using the UnresolvedStarExceptOrReplace // expression. We do this once per SET assignment to allow for multiple SET assignments with // optional lateral references to previous ones. diff --git a/sql/core/src/test/resources/sql-tests/analyzer-results/pipe-operators.sql.out b/sql/core/src/test/resources/sql-tests/analyzer-results/pipe-operators.sql.out index 94fbe261ca29c..abc81eb2c8416 100644 --- a/sql/core/src/test/resources/sql-tests/analyzer-results/pipe-operators.sql.out +++ b/sql/core/src/test/resources/sql-tests/analyzer-results/pipe-operators.sql.out @@ -992,6 +992,65 @@ Project [a#x, a#x, z2#x] +- LocalRelation [a#x] +-- !query +values (1, 10) as t(a, b) +|> set a = a + 1 +|> select t.a +-- !query analysis +Project [a#x] ++- Project [(a#x + 1) AS a#x, b#x, a#x] + +- SubqueryAlias t + +- LocalRelation [a#x, b#x] + + +-- !query +values (1, 10) as t(a, b) +|> set a = a + 1 +|> select a, t.a, t.b +-- !query analysis +Project [a#x, a#x, b#x] ++- Project [(a#x + 1) AS a#x, b#x, a#x] + +- SubqueryAlias t + +- LocalRelation [a#x, b#x] + + +-- !query +values (1, 10) as t(a, b) +|> set a = a + 1 +|> select * +-- !query analysis +Project [a#x, b#x] ++- Project [(a#x + 1) AS a#x, b#x] + +- SubqueryAlias t + +- LocalRelation [a#x, b#x] + + +-- !query +table t +|> as u +|> set x = x + 1 +|> select x, u.x +-- !query analysis +Project [x#x, x#x] ++- Project [(x#x + 1) AS x#x, y#x, x#x] + +- SubqueryAlias u + +- SubqueryAlias spark_catalog.default.t + +- Relation spark_catalog.default.t[x#x,y#x] csv + + +-- !query +values (1) as t(a) +|> set a = a + 1 +|> set a = a + 1 +|> select a, t.a +-- !query analysis +Project [a#x, a#x] ++- Project [(a#x + 1) AS a#x, a#x] + +- Project [(a#x + 1) AS a#x, a#x] + +- SubqueryAlias t + +- LocalRelation [a#x] + + -- !query table t |> set z = 1 diff --git a/sql/core/src/test/resources/sql-tests/inputs/pipe-operators.sql b/sql/core/src/test/resources/sql-tests/inputs/pipe-operators.sql index 916b114f17f57..f9faacfb8fa68 100644 --- a/sql/core/src/test/resources/sql-tests/inputs/pipe-operators.sql +++ b/sql/core/src/test/resources/sql-tests/inputs/pipe-operators.sql @@ -361,6 +361,33 @@ values (0), (1) lhs(a) |> limit 2 |> select lhs.a, rhs.a, z2; +-- A table alias still refers to the original row value of a column affected by SET. +values (1, 10) as t(a, b) +|> set a = a + 1 +|> select t.a; + +-- The unqualified name refers to the assigned value and the qualified name to the original one. +values (1, 10) as t(a, b) +|> set a = a + 1 +|> select a, t.a, t.b; + +-- The retained source column stays out of the output schema unless it is named explicitly. +values (1, 10) as t(a, b) +|> set a = a + 1 +|> select *; + +-- Qualified access to a column affected by SET, through an alias added by the AS operator. +table t +|> as u +|> set x = x + 1 +|> select x, u.x; + +-- The alias keeps pointing at the original source value across a sequence of SET operators. +values (1) as t(a) +|> set a = a + 1 +|> set a = a + 1 +|> select a, t.a; + -- SET operators: negative tests. --------------------------------- diff --git a/sql/core/src/test/resources/sql-tests/results/pipe-operators.sql.out b/sql/core/src/test/resources/sql-tests/results/pipe-operators.sql.out index 66cf72698a2d1..222d9fef38dab 100644 --- a/sql/core/src/test/resources/sql-tests/results/pipe-operators.sql.out +++ b/sql/core/src/test/resources/sql-tests/results/pipe-operators.sql.out @@ -909,6 +909,59 @@ struct 1 1 4 +-- !query +values (1, 10) as t(a, b) +|> set a = a + 1 +|> select t.a +-- !query schema +struct +-- !query output +1 + + +-- !query +values (1, 10) as t(a, b) +|> set a = a + 1 +|> select a, t.a, t.b +-- !query schema +struct +-- !query output +2 1 10 + + +-- !query +values (1, 10) as t(a, b) +|> set a = a + 1 +|> select * +-- !query schema +struct +-- !query output +2 10 + + +-- !query +table t +|> as u +|> set x = x + 1 +|> select x, u.x +-- !query schema +struct +-- !query output +1 0 +2 1 + + +-- !query +values (1) as t(a) +|> set a = a + 1 +|> set a = a + 1 +|> select a, t.a +-- !query schema +struct +-- !query output +3 1 + + -- !query table t |> set z = 1 From d092103b273205d1e9c31472cebff75e32939df0 Mon Sep 17 00:00:00 2001 From: AMC-hawk Date: Sat, 19 Sep 2026 01:19:21 +0530 Subject: [PATCH 2/6] [SPARK-59146][SQL] Preserve column order and Project tags for pipe SET hidden output Address review comments: - Keep the original column order for a qualified star after pipe SET. The SET Project now retains the whole qualified source row, in order, as hidden output, and the qualified star expansion emits an attribute that is both hidden and visible once, at its hidden position. Previously `VALUES (1, 2, 3) AS t(a, b, c) |> SET b = 20 |> SELECT t.*` returned (2, 1, 3). DataFrame column resolution treats an attribute present in both output and metadataOutput as a regular match. - Copy the original Project tags before setting the hidden output tag, since tag forwarding only applies to a node without tags. This preserves the Spark Connect plan id of a relation rooted at pipe SET. --- .../sql/catalyst/analysis/Analyzer.scala | 22 +++++++-- .../analysis/ColumnResolutionHelper.scala | 7 ++- .../sql/catalyst/analysis/unresolved.scala | 11 ++++- .../sql/catalyst/analysis/AnalysisSuite.scala | 17 +++++++ .../analyzer-results/pipe-operators.sql.out | 49 +++++++++++++++++++ .../sql-tests/inputs/pipe-operators.sql | 21 ++++++++ .../sql-tests/results/pipe-operators.sql.out | 42 ++++++++++++++++ 7 files changed, 162 insertions(+), 7 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala index df032ba79f9d4..eacee8129897d 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala @@ -2093,6 +2093,12 @@ class Analyzer( * the documented behavior that table aliases keep referring to the original row values after an * assignment. Retain the excluded attributes as hidden output instead, the same way USING joins * hide their duplicated join keys (SPARK-59146). + * + * The hidden output holds the whole qualified source row in its original order, not just the + * excluded attributes, because a qualified star expands hidden output ahead of the visible + * output. Retaining only the excluded attributes would move them to the front, so that + * `VALUES (1, 2, 3) AS t(a, b, c) |> SET b = 20 |> SELECT t.*` returned `(b, a, c)`. Attributes + * that are also visible are emitted once by the star expansion, in their hidden position. */ private def retainExceptedColumnsAsHiddenOutput(original: Project, expanded: Project): Unit = { val retain = original.projectList.exists { @@ -2100,11 +2106,21 @@ class Analyzer( case _ => false } if (retain) { - val excepted = expanded.child.output.filterNot(expanded.outputSet.contains) - if (excepted.nonEmpty) { + val child = expanded.child + if (child.output.exists(!expanded.outputSet.contains(_))) { + // The row a qualified star sees on the child: its qualified-only hidden output first, + // then its output. Only qualified attributes are reachable through a qualified star. + val sourceRow = (child.metadataOutput.filter(_.qualifiedAccessOnly) ++ child.output) + .filter(_.qualifier.nonEmpty) + .distinctBy(_.exprId) + val sourceRowIds = sourceRow.map(_.exprId).toSet + // The rule forwards the original tags, such as the Spark Connect plan id, only onto a + // node without tags, so copy them here before adding the hidden output tag. + expanded.copyTagsFrom(original) expanded.setTagValue( Project.hiddenOutputTag, - excepted.map(_.markAsQualifiedAccessOnly()) ++ expanded.child.metadataOutput) + sourceRow.map(_.markAsQualifiedAccessOnly()) ++ + child.metadataOutput.filterNot(a => sourceRowIds.contains(a.exprId))) } } } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ColumnResolutionHelper.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ColumnResolutionHelper.scala index 08bfcc446029d..1b2ce6b964086 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ColumnResolutionHelper.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ColumnResolutionHelper.scala @@ -687,9 +687,12 @@ trait ColumnResolutionHelper extends Logging with DataTypeErrorsBase { // ancestor plan (e.g. a natural/USING join wrapper that hides a join key // via `Project.hiddenOutputTag`). We accept that here but tag the candidate // as `hidden` so the top-level merge in `resolveDataFrameColumn` can prefer - // a regular (p.output) match over hidden (p.metadataOutput) ones. + // a regular (p.output) match over hidden (p.metadataOutput) ones. An + // attribute can be in both, e.g. under the SQL pipe SET operator, and then + // counts as a regular match. val filtered = candidates.flatMap { c => - val hidden = c.hidden || c.expr.references.subsetOf(AttributeSet(p.metadataOutput)) + val hidden = c.hidden || (!c.expr.references.subsetOf(p.outputSet) && + c.expr.references.subsetOf(AttributeSet(p.metadataOutput))) if (c.expr.references.subsetOf(AttributeSet(p.output ++ p.metadataOutput))) { Some(c.copy(hidden = hidden)) } else { diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala index 3cb95de2123db..c34e70d5a0459 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala @@ -584,8 +584,15 @@ trait UnresolvedStarBase extends Star with Unevaluable { // keep any restrictions that may break column resolution for normal attributes. // See SPARK-42084 for more details. .map(_.markAsAllowAnyAccess()) - val expandedAttributes = (hiddenOutput ++ parameters.childOperatorOutput) - .filter(matchedQualifier(_, target.get, parameters.resolver)) + // Hidden output may also contain visible attributes to fix their position in the expansion, + // as the SQL pipe SET operator does to keep the original column order (SPARK-59146). Emit + // each such attribute once, at its hidden position, using the visible attribute itself. + val visibleById = parameters.childOperatorOutput.map(a => a.exprId -> a).toMap + val hiddenIds = hiddenOutput.map(_.exprId).toSet + val expandedAttributes = + (hiddenOutput.map(a => visibleById.getOrElse(a.exprId, a)) ++ + parameters.childOperatorOutput.filterNot(a => hiddenIds.contains(a.exprId))) + .filter(matchedQualifier(_, target.get, parameters.resolver)) if (expandedAttributes.nonEmpty) return expandedAttributes diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala index 4a7adcc050a00..58fc4e835052e 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala @@ -1894,6 +1894,23 @@ class AnalysisSuite extends AnalysisTest with Matchers { val expectedPlan = Project(Seq(UnresolvedAttribute("i")), addColumnF).analyze checkAnalysis(inputPlan, expectedPlan) } + + test("SPARK-59146: pipe SET keeps the original Project tags alongside the hidden output") { + // The Project that the SQL pipe SET operator builds, as a Spark Connect relation. + val set = Project( + Seq(UnresolvedStarExceptOrReplace( + target = None, + excepts = Seq(Seq("a")), + replacements = Some(Seq(Alias(Literal(1), "a")())), + retainExceptedColumnsAsHidden = true)), + testRelation.subquery("t")) + set.setTagValue(LogicalPlan.PLAN_ID_TAG, 42L) + + val analyzed = getAnalyzer.execute(set) + assert(analyzed.isInstanceOf[Project], analyzed) + assert(analyzed.getTagValue(LogicalPlan.PLAN_ID_TAG).contains(42L)) + assert(analyzed.getTagValue(Project.hiddenOutputTag).exists(_.map(_.name) == Seq("a"))) + } } /** diff --git a/sql/core/src/test/resources/sql-tests/analyzer-results/pipe-operators.sql.out b/sql/core/src/test/resources/sql-tests/analyzer-results/pipe-operators.sql.out index abc81eb2c8416..a251379300e5a 100644 --- a/sql/core/src/test/resources/sql-tests/analyzer-results/pipe-operators.sql.out +++ b/sql/core/src/test/resources/sql-tests/analyzer-results/pipe-operators.sql.out @@ -1051,6 +1051,55 @@ Project [a#x, a#x] +- LocalRelation [a#x] +-- !query +values (1, 2, 3) as t(a, b, c) +|> set b = 20 +|> select t.* +-- !query analysis +Project [a#x, b#x, c#x] ++- Project [a#x, 20 AS b#x, c#x, b#x] + +- SubqueryAlias t + +- LocalRelation [a#x, b#x, c#x] + + +-- !query +values (1, 2, 3) as t(a, b, c) +|> set b = 20 +|> set a = 10 +|> select t.* +-- !query analysis +Project [a#x, b#x, c#x] ++- Project [10 AS a#x, b#x, c#x, a#x, b#x] + +- Project [a#x, 20 AS b#x, c#x, b#x] + +- SubqueryAlias t + +- LocalRelation [a#x, b#x, c#x] + + +-- !query +values (1, 2, 3) as t(a, b, c) +|> set a = 10 +|> set c = 30 +|> select a, b, c, t.* +-- !query analysis +Project [a#x, b#x, c#x, a#x, b#x, c#x] ++- Project [a#x, b#x, 30 AS c#x, a#x, c#x] + +- Project [10 AS a#x, b#x, c#x, a#x] + +- SubqueryAlias t + +- LocalRelation [a#x, b#x, c#x] + + +-- !query +values (1, 2, 3) as t(a, b, c) +|> set c = 30, a = 10 +|> select t.* +-- !query analysis +Project [a#x, b#x, c#x] ++- Project [10 AS a#x, b#x, c#x, a#x, c#x] + +- Project [a#x, b#x, 30 AS c#x, c#x] + +- SubqueryAlias t + +- LocalRelation [a#x, b#x, c#x] + + -- !query table t |> set z = 1 diff --git a/sql/core/src/test/resources/sql-tests/inputs/pipe-operators.sql b/sql/core/src/test/resources/sql-tests/inputs/pipe-operators.sql index f9faacfb8fa68..092bb308c0718 100644 --- a/sql/core/src/test/resources/sql-tests/inputs/pipe-operators.sql +++ b/sql/core/src/test/resources/sql-tests/inputs/pipe-operators.sql @@ -388,6 +388,27 @@ values (1) as t(a) |> set a = a + 1 |> select a, t.a; +-- A qualified star returns the original row in its original column order. +values (1, 2, 3) as t(a, b, c) +|> set b = 20 +|> select t.*; + +-- The order is preserved across SET operators on different columns, in either order. +values (1, 2, 3) as t(a, b, c) +|> set b = 20 +|> set a = 10 +|> select t.*; + +values (1, 2, 3) as t(a, b, c) +|> set a = 10 +|> set c = 30 +|> select a, b, c, t.*; + +-- Assigning several columns in one SET operator. +values (1, 2, 3) as t(a, b, c) +|> set c = 30, a = 10 +|> select t.*; + -- SET operators: negative tests. --------------------------------- diff --git a/sql/core/src/test/resources/sql-tests/results/pipe-operators.sql.out b/sql/core/src/test/resources/sql-tests/results/pipe-operators.sql.out index 222d9fef38dab..cc78590d8cee9 100644 --- a/sql/core/src/test/resources/sql-tests/results/pipe-operators.sql.out +++ b/sql/core/src/test/resources/sql-tests/results/pipe-operators.sql.out @@ -962,6 +962,48 @@ struct 3 1 +-- !query +values (1, 2, 3) as t(a, b, c) +|> set b = 20 +|> select t.* +-- !query schema +struct +-- !query output +1 2 3 + + +-- !query +values (1, 2, 3) as t(a, b, c) +|> set b = 20 +|> set a = 10 +|> select t.* +-- !query schema +struct +-- !query output +1 2 3 + + +-- !query +values (1, 2, 3) as t(a, b, c) +|> set a = 10 +|> set c = 30 +|> select a, b, c, t.* +-- !query schema +struct +-- !query output +10 2 30 1 2 3 + + +-- !query +values (1, 2, 3) as t(a, b, c) +|> set c = 30, a = 10 +|> select t.* +-- !query schema +struct +-- !query output +1 2 3 + + -- !query table t |> set z = 1 From 262b18bbe7dd2c213e2d7a7e8380bcabb5eebc62 Mon Sep 17 00:00:00 2001 From: AMC-hawk Date: Sun, 20 Sep 2026 21:51:57 +0530 Subject: [PATCH 3/6] [SPARK-59146][SQL] Preserve repeated source-row positions for pipe SET hidden output Address review comments: - Keep every position of the source row that pipe SET retains. A projection may repeat a column, and the repeated positions share an ExprId, so using the ExprId as positional identity dropped one of them: `VALUES (1, 2) AS s(a, b) |> SELECT a, a, b |> AS t |> SET b = 3 |> SELECT t.*` returned (a, b) instead of (a, a, b). Retention and qualified star expansion now match occurrences through the new UnresolvedStarBase.mergeHiddenAndVisibleOutput, where a hidden attribute consumes at most one visible occurrence and is emitted in its place. - Replace the AnalysisSuite case for DataFrame column resolution, which only asserted that the two Project tags coexist. It now resolves a plan id bound column across a visible and hidden overlap, which fails if the visible candidate is classified as hidden. - Add golden coverage for a qualified SET reference under the rules that rebuild the projection above it: filter, sort, aggregate, join, union, an enclosing subquery and a CTE. --- .../sql/catalyst/analysis/Analyzer.scala | 20 +-- .../sql/catalyst/analysis/unresolved.scala | 42 +++++-- .../sql/catalyst/analysis/AnalysisSuite.scala | 25 ++++ .../analyzer-results/pipe-operators.sql.out | 114 ++++++++++++++++++ .../sql-tests/inputs/pipe-operators.sql | 38 ++++++ .../sql-tests/results/pipe-operators.sql.out | 84 +++++++++++++ 6 files changed, 305 insertions(+), 18 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala index eacee8129897d..abf36d0975a93 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala @@ -2094,11 +2094,12 @@ class Analyzer( * assignment. Retain the excluded attributes as hidden output instead, the same way USING joins * hide their duplicated join keys (SPARK-59146). * - * The hidden output holds the whole qualified source row in its original order, not just the - * excluded attributes, because a qualified star expands hidden output ahead of the visible - * output. Retaining only the excluded attributes would move them to the front, so that - * `VALUES (1, 2, 3) AS t(a, b, c) |> SET b = 20 |> SELECT t.*` returned `(b, a, c)`. Attributes - * that are also visible are emitted once by the star expansion, in their hidden position. + * The hidden output holds the whole qualified source row, with every position in its original + * order, not just the excluded attributes, because a qualified star expands hidden output + * ahead of the visible output. Retaining only the excluded attributes would move them to the + * front, so that `VALUES (1, 2, 3) AS t(a, b, c) |> SET b = 20 |> SELECT t.*` returned + * `(b, a, c)`. Attributes that are also visible are emitted once by the star expansion, in + * their hidden position. */ private def retainExceptedColumnsAsHiddenOutput(original: Project, expanded: Project): Unit = { val retain = original.projectList.exists { @@ -2108,11 +2109,12 @@ class Analyzer( if (retain) { val child = expanded.child if (child.output.exists(!expanded.outputSet.contains(_))) { - // The row a qualified star sees on the child: its qualified-only hidden output first, - // then its output. Only qualified attributes are reachable through a qualified star. - val sourceRow = (child.metadataOutput.filter(_.qualifiedAccessOnly) ++ child.output) + // The row a qualified star sees on the child, keeping every one of its positions. Only + // qualified attributes are reachable through a qualified star. + val sourceRow = UnresolvedStarBase + .mergeHiddenAndVisibleOutput( + child.metadataOutput.filter(_.qualifiedAccessOnly), child.output) .filter(_.qualifier.nonEmpty) - .distinctBy(_.exprId) val sourceRowIds = sourceRow.map(_.exprId).toSet // The rule forwards the original tags, such as the Spark Connect plan id, only onto a // node without tags, so copy them here before adding the hidden output tag. diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala index c34e70d5a0459..230436c95d8c6 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala @@ -17,6 +17,8 @@ package org.apache.spark.sql.catalyst.analysis +import scala.collection.mutable + import org.apache.spark.SparkException import org.apache.spark.sql.AnalysisException import org.apache.spark.sql.catalyst.{FunctionIdentifier, InternalRow, TableIdentifier} @@ -584,15 +586,9 @@ trait UnresolvedStarBase extends Star with Unevaluable { // keep any restrictions that may break column resolution for normal attributes. // See SPARK-42084 for more details. .map(_.markAsAllowAnyAccess()) - // Hidden output may also contain visible attributes to fix their position in the expansion, - // as the SQL pipe SET operator does to keep the original column order (SPARK-59146). Emit - // each such attribute once, at its hidden position, using the visible attribute itself. - val visibleById = parameters.childOperatorOutput.map(a => a.exprId -> a).toMap - val hiddenIds = hiddenOutput.map(_.exprId).toSet - val expandedAttributes = - (hiddenOutput.map(a => visibleById.getOrElse(a.exprId, a)) ++ - parameters.childOperatorOutput.filterNot(a => hiddenIds.contains(a.exprId))) - .filter(matchedQualifier(_, target.get, parameters.resolver)) + val expandedAttributes = UnresolvedStarBase + .mergeHiddenAndVisibleOutput(hiddenOutput, parameters.childOperatorOutput) + .filter(matchedQualifier(_, target.get, parameters.resolver)) if (expandedAttributes.nonEmpty) return expandedAttributes @@ -633,6 +629,34 @@ trait UnresolvedStarBase extends Star with Unevaluable { override def toString: String = target.map(_.mkString("", ".", ".")).getOrElse("") + "*" } +object UnresolvedStarBase { + + /** + * Merges the hidden output of a plan with its visible output, in the order a qualified star + * expands them. + * + * Hidden output usually holds attributes that are not visible at all, such as the duplicated + * join keys of a USING join. It may also repeat an attribute that is visible, to pin that + * attribute's position in the expansion, as the SQL pipe SET operator does to keep the original + * column order of the row its table alias refers to (SPARK-59146). + * + * Every occurrence of the source row is therefore preserved: a hidden attribute consumes at + * most one visible occurrence, which is emitted in the hidden attribute's place, because a + * projection may repeat the same attribute. Visible attributes that no hidden attribute + * consumed keep their relative order, after the hidden output. + */ + def mergeHiddenAndVisibleOutput( + hiddenOutput: Seq[Attribute], + visibleOutput: Seq[Attribute]): Seq[Attribute] = { + val unconsumed = visibleOutput.to(mutable.ArrayBuffer) + val merged = hiddenOutput.map { hidden => + val visible = unconsumed.indexWhere(_.exprId == hidden.exprId) + if (visible >= 0) unconsumed.remove(visible) else hidden + } + merged ++ unconsumed.toSeq + } +} + /** * Represents some of the input attributes to a given relational operator, for example in * "SELECT * EXCEPT(a) FROM ...". diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala index 58fc4e835052e..6d9381249cabb 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala @@ -1911,6 +1911,31 @@ class AnalysisSuite extends AnalysisTest with Matchers { assert(analyzed.getTagValue(LogicalPlan.PLAN_ID_TAG).contains(42L)) assert(analyzed.getTagValue(Project.hiddenOutputTag).exists(_.map(_.name) == Seq("a"))) } + + test("SPARK-59146: a DataFrame column keeps a visible candidate of a hidden output visible") { + // Two plan nodes carry the same plan id, as the two sides of a Spark Connect self join do. + // On the left, `b` is visible and also hidden, the way pipe SET retains a source column it + // did not assign; on the right it is only visible. Both candidates are regular ones, so the + // reference is ambiguous. Treating the left one as hidden would instead silently resolve it + // to the right side. + val Seq(a, b) = testRelation2.output.take(2) + val set = Project(Seq(Alias(Literal("x"), "a")(), b), testRelation2) + set.setTagValue(Project.hiddenOutputTag, Seq(a, b).map(_.markAsQualifiedAccessOnly())) + set.setTagValue(LogicalPlan.PLAN_ID_TAG, 1L) + + val otherRelation = LocalRelation(AttributeReference("b", StringType)()) + val other = Project(otherRelation.output, otherRelation) + other.setTagValue(LogicalPlan.PLAN_ID_TAG, 1L) + + val column = UnresolvedAttribute("b") + column.setTagValue(LogicalPlan.PLAN_ID_TAG, 1L) + val plan = Project(Seq(column), Join(set, other, Inner, None, JoinHint.NONE)) + + checkError( + exception = intercept[AnalysisException](getAnalyzer.execute(plan)), + condition = "AMBIGUOUS_COLUMN_REFERENCE", + parameters = Map("name" -> "\"b\"")) + } } /** diff --git a/sql/core/src/test/resources/sql-tests/analyzer-results/pipe-operators.sql.out b/sql/core/src/test/resources/sql-tests/analyzer-results/pipe-operators.sql.out index a251379300e5a..d523b293000c5 100644 --- a/sql/core/src/test/resources/sql-tests/analyzer-results/pipe-operators.sql.out +++ b/sql/core/src/test/resources/sql-tests/analyzer-results/pipe-operators.sql.out @@ -1100,6 +1100,120 @@ Project [a#x, b#x, c#x] +- LocalRelation [a#x, b#x, c#x] +-- !query +values (1, 2) as s(a, b) +|> select a, a, b +|> as t +|> set b = 3 +|> select t.* +-- !query analysis +Project [a#x, a#x, b#x] ++- Project [a#x, a#x, 3 AS b#x, b#x] + +- SubqueryAlias t + +- Project [a#x, a#x, b#x] + +- SubqueryAlias s + +- LocalRelation [a#x, b#x] + + +-- !query +values (1, 10) as t(a, b) +|> set a = a + 1 +|> where t.a = 1 +|> select a, t.a +-- !query analysis +Project [a#x, a#x] ++- Filter (a#x = 1) + +- PipeOperator + +- Project [(a#x + 1) AS a#x, b#x, a#x] + +- SubqueryAlias t + +- LocalRelation [a#x, b#x] + + +-- !query +values (1, 10) as t(a, b) +|> set a = a + 1 +|> order by t.a +|> select a, t.a +-- !query analysis +Project [a#x, a#x] ++- Sort [a#x ASC NULLS FIRST], true + +- PipeOperator + +- Project [(a#x + 1) AS a#x, b#x, a#x] + +- SubqueryAlias t + +- LocalRelation [a#x, b#x] + + +-- !query +values (1, 10) as t(a, b) +|> set a = a + 1 +|> aggregate sum(t.a) as original_sum group by a +-- !query analysis +Aggregate [a#x], [a#x, sum(a#x) AS original_sum#xL] ++- Project [(a#x + 1) AS a#x, b#x, a#x] + +- SubqueryAlias t + +- LocalRelation [a#x, b#x] + + +-- !query +values (1, 10) as t(a, b) +|> set a = a + 1 +|> join (values (1, 20) as u(c, d)) on t.a = c +|> select a, t.a, c +-- !query analysis +Project [a#x, a#x, c#x] ++- Join Inner, (a#x = c#x) + :- Project [(a#x + 1) AS a#x, b#x, a#x] + : +- SubqueryAlias t + : +- LocalRelation [a#x, b#x] + +- SubqueryAlias __auto_generated_subquery_name + +- SubqueryAlias u + +- LocalRelation [c#x, d#x] + + +-- !query +values (1, 10) as t(a, b) +|> set a = a + 1 +|> select a, t.a +|> union all (values (1, 10) as t(a, b) |> set a = a + 1 |> select a, t.a) +-- !query analysis +Union false, false +:- Project [a#x, a#x] +: +- Project [(a#x + 1) AS a#x, b#x, a#x] +: +- SubqueryAlias t +: +- LocalRelation [a#x, b#x] ++- Project [a#x, a#x] + +- Project [(a#x + 1) AS a#x, b#x, a#x] + +- SubqueryAlias t + +- LocalRelation [a#x, b#x] + + +-- !query +select * from (values (1, 10) as t(a, b) |> set a = a + 1 |> select a, t.a) +-- !query analysis +Project [a#x, a#x] ++- SubqueryAlias __auto_generated_subquery_name + +- Project [a#x, a#x] + +- Project [(a#x + 1) AS a#x, b#x, a#x] + +- SubqueryAlias t + +- LocalRelation [a#x, b#x] + + +-- !query +with cte as (values (1, 10) as t(a, b) |> set a = a + 1 |> select a, t.a) +select * from cte +-- !query analysis +WithCTE +:- CTERelationDef xxxx, false +: +- SubqueryAlias cte +: +- Project [a#x, a#x] +: +- Project [(a#x + 1) AS a#x, b#x, a#x] +: +- SubqueryAlias t +: +- LocalRelation [a#x, b#x] ++- Project [a#x, a#x] + +- SubqueryAlias cte + +- CTERelationRef xxxx, true, [a#x, a#x], false, false, 1 + + -- !query table t |> set z = 1 diff --git a/sql/core/src/test/resources/sql-tests/inputs/pipe-operators.sql b/sql/core/src/test/resources/sql-tests/inputs/pipe-operators.sql index 092bb308c0718..768d57322d73b 100644 --- a/sql/core/src/test/resources/sql-tests/inputs/pipe-operators.sql +++ b/sql/core/src/test/resources/sql-tests/inputs/pipe-operators.sql @@ -409,6 +409,44 @@ values (1, 2, 3) as t(a, b, c) |> set c = 30, a = 10 |> select t.*; +-- Repeated source columns keep every position in the qualified star. +values (1, 2) as s(a, b) +|> select a, a, b +|> as t +|> set b = 3 +|> select t.*; + +-- The retained source column survives the rules that rebuild the projection above SET: +-- a filter, a sort, an aggregate, a join, a union and an enclosing subquery or CTE. +values (1, 10) as t(a, b) +|> set a = a + 1 +|> where t.a = 1 +|> select a, t.a; + +values (1, 10) as t(a, b) +|> set a = a + 1 +|> order by t.a +|> select a, t.a; + +values (1, 10) as t(a, b) +|> set a = a + 1 +|> aggregate sum(t.a) as original_sum group by a; + +values (1, 10) as t(a, b) +|> set a = a + 1 +|> join (values (1, 20) as u(c, d)) on t.a = c +|> select a, t.a, c; + +values (1, 10) as t(a, b) +|> set a = a + 1 +|> select a, t.a +|> union all (values (1, 10) as t(a, b) |> set a = a + 1 |> select a, t.a); + +select * from (values (1, 10) as t(a, b) |> set a = a + 1 |> select a, t.a); + +with cte as (values (1, 10) as t(a, b) |> set a = a + 1 |> select a, t.a) +select * from cte; + -- SET operators: negative tests. --------------------------------- diff --git a/sql/core/src/test/resources/sql-tests/results/pipe-operators.sql.out b/sql/core/src/test/resources/sql-tests/results/pipe-operators.sql.out index cc78590d8cee9..41eba9e36e03e 100644 --- a/sql/core/src/test/resources/sql-tests/results/pipe-operators.sql.out +++ b/sql/core/src/test/resources/sql-tests/results/pipe-operators.sql.out @@ -1004,6 +1004,90 @@ struct 1 2 3 +-- !query +values (1, 2) as s(a, b) +|> select a, a, b +|> as t +|> set b = 3 +|> select t.* +-- !query schema +struct +-- !query output +1 1 2 + + +-- !query +values (1, 10) as t(a, b) +|> set a = a + 1 +|> where t.a = 1 +|> select a, t.a +-- !query schema +struct +-- !query output +2 1 + + +-- !query +values (1, 10) as t(a, b) +|> set a = a + 1 +|> order by t.a +|> select a, t.a +-- !query schema +struct +-- !query output +2 1 + + +-- !query +values (1, 10) as t(a, b) +|> set a = a + 1 +|> aggregate sum(t.a) as original_sum group by a +-- !query schema +struct +-- !query output +2 1 + + +-- !query +values (1, 10) as t(a, b) +|> set a = a + 1 +|> join (values (1, 20) as u(c, d)) on t.a = c +|> select a, t.a, c +-- !query schema +struct +-- !query output +2 1 1 + + +-- !query +values (1, 10) as t(a, b) +|> set a = a + 1 +|> select a, t.a +|> union all (values (1, 10) as t(a, b) |> set a = a + 1 |> select a, t.a) +-- !query schema +struct +-- !query output +2 1 +2 1 + + +-- !query +select * from (values (1, 10) as t(a, b) |> set a = a + 1 |> select a, t.a) +-- !query schema +struct +-- !query output +2 1 + + +-- !query +with cte as (values (1, 10) as t(a, b) |> set a = a + 1 |> select a, t.a) +select * from cte +-- !query schema +struct +-- !query output +2 1 + + -- !query table t |> set z = 1 From 5545d0d885c5dd4dc49a6d74345ef9782fe6a2d1 Mon Sep 17 00:00:00 2001 From: AMC-hawk Date: Tue, 20 Oct 2026 23:58:30 +0530 Subject: [PATCH 4/6] [SPARK-59146][SQL] Keep qualified star analysis linear for pipe SET hidden output Address review comments: - Make UnresolvedStarBase.mergeHiddenAndVisibleOutput linear. It searched the remaining visible output for every hidden attribute and removed each match from the middle of a buffer, so a qualified star over a wide USING or NATURAL join, or after pipe SET, was quadratic in the schema width. It now indexes the visible occurrences by ExprId in one pass, consumes them in constant expected time and emits the unconsumed ones in one final pass, with the same output. - Look up the metadata output of the children by ExprId in AddMetadataColumns. It scanned that output for every attribute of a node, and pipe SET retains the whole source row as hidden output, so analyzing any wide pipe SET query was quadratic, with or without a qualified star. - Use a consistent counterfactual in the comment on retainExceptedColumnsAsHiddenOutput. - Add AnalysisSuite cases for the merge on disjoint, overlapping and repeated inputs, checking the exact instances it emits, and for a qualified star after repeated pipe SETs. Co-Authored-By: Claude Opus 5.5 --- .../sql/catalyst/analysis/Analyzer.scala | 24 +++++--- .../sql/catalyst/analysis/unresolved.scala | 25 ++++++-- .../sql/catalyst/analysis/AnalysisSuite.scala | 58 +++++++++++++++++++ 3 files changed, 94 insertions(+), 13 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala index abf36d0975a93..474981612a5bf 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala @@ -1081,28 +1081,34 @@ class Analyzer( } private def getMetadataAttributes(plan: LogicalPlan): Seq[Attribute] = { + val childrenMetadataOutput = childrenMetadataOutputById(plan) plan.expressions.flatMap(_.collect { case a: Attribute if a.isMetadataCol => a - case a: Attribute - if plan.children.exists(c => c.metadataOutput.exists(_.exprId == a.exprId)) => - plan.children.collectFirst { - case c if c.metadataOutput.exists(_.exprId == a.exprId) => - c.metadataOutput.find(_.exprId == a.exprId).get - }.get + case a: Attribute if childrenMetadataOutput.contains(a.exprId) => + childrenMetadataOutput(a.exprId) }) } private def hasMetadataCol(plan: LogicalPlan): Boolean = { + lazy val childrenMetadataOutput = childrenMetadataOutputById(plan) plan.expressions.exists(_.exists { case a: Attribute => // If an attribute is resolved before being labeled as metadata // (i.e. from the originating Dataset), we check with expression ID - a.isMetadataCol || - plan.children.exists(c => c.metadataOutput.exists(_.exprId == a.exprId)) + a.isMetadataCol || childrenMetadataOutput.contains(a.exprId) case _ => false }) } + // The metadata output of the children, keeping the first attribute for each expression ID. + // Looking attributes up by ID keeps this rule linear when a child's metadata output is as wide + // as its output, as the hidden source row of the SQL pipe SET operator is. + private def childrenMetadataOutputById(plan: LogicalPlan): collection.Map[ExprId, Attribute] = { + val byId = mutable.HashMap.empty[ExprId, Attribute] + plan.children.foreach(_.metadataOutput.foreach(a => byId.getOrElseUpdate(a.exprId, a))) + byId + } + private def addMetadataCol( plan: LogicalPlan, requiredAttrIds: Set[ExprId], @@ -2097,7 +2103,7 @@ class Analyzer( * The hidden output holds the whole qualified source row, with every position in its original * order, not just the excluded attributes, because a qualified star expands hidden output * ahead of the visible output. Retaining only the excluded attributes would move them to the - * front, so that `VALUES (1, 2, 3) AS t(a, b, c) |> SET b = 20 |> SELECT t.*` returned + * front, so that `VALUES (1, 2, 3) AS t(a, b, c) |> SET b = 20 |> SELECT t.*` would return * `(b, a, c)`. Attributes that are also visible are emitted once by the star expansion, in * their hidden position. */ diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala index 230436c95d8c6..9f00b799a01ce 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala @@ -644,16 +644,33 @@ object UnresolvedStarBase { * most one visible occurrence, which is emitted in the hidden attribute's place, because a * projection may repeat the same attribute. Visible attributes that no hidden attribute * consumed keep their relative order, after the hidden output. + * + * Both outputs can be as wide as the source row, for example under a wide USING join or a pipe + * SET, so the merge runs in time linear in their combined size. */ def mergeHiddenAndVisibleOutput( hiddenOutput: Seq[Attribute], visibleOutput: Seq[Attribute]): Seq[Attribute] = { - val unconsumed = visibleOutput.to(mutable.ArrayBuffer) + if (hiddenOutput.isEmpty) return visibleOutput + val visible = visibleOutput.toIndexedSeq + // The positions of the occurrences of each visible attribute, in order. Built back to front so + // that every list is in ascending order and its head is the first unconsumed occurrence. + val occurrences = mutable.HashMap.empty[ExprId, List[Int]] + visible.indices.reverseIterator.foreach { i => + val exprId = visible(i).exprId + occurrences(exprId) = i :: occurrences.getOrElse(exprId, Nil) + } + val consumed = new Array[Boolean](visible.length) val merged = hiddenOutput.map { hidden => - val visible = unconsumed.indexWhere(_.exprId == hidden.exprId) - if (visible >= 0) unconsumed.remove(visible) else hidden + occurrences.get(hidden.exprId) match { + case Some(i :: remaining) => + occurrences(hidden.exprId) = remaining + consumed(i) = true + visible(i) + case _ => hidden + } } - merged ++ unconsumed.toSeq + merged ++ visible.indices.collect { case i if !consumed(i) => visible(i) } } } diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala index 6d9381249cabb..b250d1bb4c4b1 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala @@ -1936,6 +1936,64 @@ class AnalysisSuite extends AnalysisTest with Matchers { condition = "AMBIGUOUS_COLUMN_REFERENCE", parameters = Map("name" -> "\"b\"")) } + + test("SPARK-59146: merge hidden and visible output for a qualified star") { + // Compares instances, so that the qualifier and metadata of every position are checked too. + def checkMerge( + hidden: Seq[Attribute], + visible: Seq[Attribute], + expected: Seq[Attribute]): Unit = { + val merged = UnresolvedStarBase.mergeHiddenAndVisibleOutput(hidden, visible) + assert(merged.length == expected.length, merged) + merged.zip(expected).foreach { case (actual, instance) => + assert(actual eq instance, s"$actual in $merged is not the instance $instance") + } + } + def attributes(qualifier: String, names: String*): Seq[AttributeReference] = + names.map(name => AttributeReference(name, IntegerType)(qualifier = Seq(qualifier))) + + // Disjoint, as for the join key that a USING join hides: the hidden output goes first. + val Seq(lk, x) = attributes("l", "k", "x") + val Seq(rk, y) = attributes("r", "k", "y") + val hiddenKey = rk.markAsQualifiedAccessOnly() + checkMerge(Seq(hiddenKey), Seq(lk, x, y), Seq(hiddenKey, lk, x, y)) + checkMerge(Nil, Seq(lk, x, y), Seq(lk, x, y)) + + // Overlapping, as for pipe SET: the hidden source row pins the position of every visible + // attribute, which is emitted in place of its hidden copy. + val Seq(a, b, c) = attributes("t", "a", "b", "c") + val assigned = AttributeReference("b", IntegerType)() + val row = Seq(a, b, c).map(_.markAsQualifiedAccessOnly()) + checkMerge(row, Seq(a, c, assigned), Seq(a, row(1), c, assigned)) + + // Repeated, as for `SELECT a, a, b`: the occurrences share an ExprId, and are told apart here + // by their qualifier. A hidden occurrence consumes the first unconsumed visible one, so every + // position is kept exactly once. + val repeatedA = a.withQualifier(Seq("u")) + val repeatedRow = Seq(a, repeatedA, b).map(_.markAsQualifiedAccessOnly()) + checkMerge( + repeatedRow, Seq(a, repeatedA, assigned), Seq(a, repeatedA, repeatedRow(2), assigned)) + checkMerge(Seq(row(1), row(0)), Seq(a, repeatedA), Seq(row(1), a, repeatedA)) + checkMerge(repeatedRow.take(2), Seq(a), Seq(a, repeatedRow(1))) + } + + test("SPARK-59146: a qualified star after repeated pipe SETs returns the original row") { + // The second SET sees the hidden output of the first one overlap its visible output. + def set(column: String, child: LogicalPlan): LogicalPlan = Project( + Seq(UnresolvedStarExceptOrReplace( + target = None, + excepts = Seq(Seq(column)), + replacements = Some(Seq(Alias(Literal("x"), column)())), + retainExceptedColumnsAsHidden = true)), + child) + val plan = Project( + Seq(UnresolvedStar(Some(Seq("t")))), + set("a", set("c", testRelation2.subquery("t")))) + + val output = getAnalyzer.execute(plan).output + assert(output.map(_.exprId) == testRelation2.output.map(_.exprId), output) + assert(output.forall(_.qualifier == Seq("t")), output) + } } /** From fa812e60b52b9c619f00b241d1c67569acbaac1d Mon Sep 17 00:00:00 2001 From: AMC-hawk Date: Wed, 21 Oct 2026 00:03:10 +0530 Subject: [PATCH 5/6] Retrigger CI Co-Authored-By: Claude Opus 5.5 From b7dc511d60141bd418206dd659fcb91c2bdb15d7 Mon Sep 17 00:00:00 2001 From: AMC-hawk Date: Sat, 24 Oct 2026 01:44:05 +0530 Subject: [PATCH 6/6] [SPARK-59146][SQL] Keep unassigned pipe SET columns regular in DataFrame column resolution SPARK-60462 tags a plan id candidate as hidden at a node that introduces its attribute in its metadataOutput, even when the attribute is also in the node's output. Pipe SET introduces its whole source row as hidden output, to keep the column order of a qualified star, so the source columns it does not assign became hidden at the SET Project. A Spark Connect DataFrame used directly and also under a USING join that hides one of its columns then failed with AMBIGUOUS_COLUMN_REFERENCE instead of resolving to the direct column. Record the source columns that the SET Project keeps visible in the new Project.visibleHiddenOutputTag, and leave them out of the hidden attributes the node introduces. The columns it assigns stay hidden, also once AddMetadataColumns adds them to its output. Replace the AnalysisSuite case that built the SET Project by hand with one that analyzes pipe SET and reuses the DataFrame that way. It fails with AMBIGUOUS_COLUMN_REFERENCE without this change. Co-Authored-By: Claude Opus 5.5 --- .../sql/catalyst/analysis/Analyzer.scala | 5 +- .../analysis/ColumnResolutionHelper.scala | 6 ++- .../plans/logical/basicLogicalOperators.scala | 8 ++++ .../sql/catalyst/analysis/AnalysisSuite.scala | 48 +++++++++++-------- 4 files changed, 45 insertions(+), 22 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala index 6ed48667fdd33..b1a176a3d87cd 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala @@ -2202,7 +2202,8 @@ class Analyzer( * ahead of the visible output. Retaining only the excluded attributes would move them to the * front, so that `VALUES (1, 2, 3) AS t(a, b, c) |> SET b = 20 |> SELECT t.*` would return * `(b, a, c)`. Attributes that are also visible are emitted once by the star expansion, in - * their hidden position. + * their hidden position. The Project does not hide them, so they are also recorded in + * [[Project.visibleHiddenOutputTag]], which keeps them regular in DataFrame column resolution. */ private def retainExceptedColumnsAsHiddenOutput(original: Project, expanded: Project): Unit = { val retain = original.projectList.exists { @@ -2226,6 +2227,8 @@ class Analyzer( Project.hiddenOutputTag, sourceRow.map(_.markAsQualifiedAccessOnly()) ++ child.metadataOutput.filterNot(a => sourceRowIds.contains(a.exprId))) + expanded.setTagValue( + Project.visibleHiddenOutputTag, sourceRow.filter(expanded.outputSet.contains)) } } } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ColumnResolutionHelper.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ColumnResolutionHelper.scala index 2c8daf0d27cbb..6084a74ae362a 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ColumnResolutionHelper.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ColumnResolutionHelper.scala @@ -701,8 +701,12 @@ trait ColumnResolutionHelper extends Logging with DataTypeErrorsBase { // - `AddMetadataColumns` adds a hidden key to the `output` of the USING join // wrapper that hides it, and a metadata column to the `output` of a relation. // That node introduces the attribute in its `metadataOutput`, so it stays hidden. + // - The Project of the SQL pipe SET operator introduces the whole source row in its + // `metadataOutput`, to keep the column order of a qualified star, but only hides the + // columns it assigns. The others are in `Project.visibleHiddenOutputTag` and visible. val introducedHidden = AttributeSet(p.metadataOutput) -- - p.children.flatMap(_.metadataOutput) + p.children.flatMap(_.metadataOutput) -- + p.getTagValue(Project.visibleHiddenOutputTag).getOrElse(Nil) val visible = p.outputSet -- introducedHidden val filtered = candidates.flatMap { c => val hidden = c.hidden || !c.expr.references.subsetOf(visible) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala index a08214404c6ea..105cc326b512b 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala @@ -120,6 +120,14 @@ case class Project(projectList: Seq[NamedExpression], child: LogicalPlan) object Project { val hiddenOutputTag: TreeNodeTag[Seq[Attribute]] = TreeNodeTag[Seq[Attribute]]("hidden_output") + /** + * The attributes of [[hiddenOutputTag]] that the Project also keeps visible in its output. They + * are in the hidden output only to keep their positions in a qualified star, as are the source + * columns that the SQL pipe SET operator does not assign, so the Project does not hide them. + */ + val visibleHiddenOutputTag: TreeNodeTag[Seq[Attribute]] = + TreeNodeTag[Seq[Attribute]]("visible_hidden_output") + def matchSchema(plan: LogicalPlan, schema: StructType, conf: SQLConf): Project = { assert(plan.resolved) val projectList = diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala index 96b0a28a64a1a..1e4bb9e260ac5 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala @@ -2020,29 +2020,37 @@ class AnalysisSuite extends AnalysisTest with Matchers { assert(analyzed.getTagValue(Project.hiddenOutputTag).exists(_.map(_.name) == Seq("a"))) } - test("SPARK-59146: a DataFrame column keeps a visible candidate of a hidden output visible") { - // Two plan nodes carry the same plan id, as the two sides of a Spark Connect self join do. - // On the left, `b` is visible and also hidden, the way pipe SET retains a source column it - // did not assign; on the right it is only visible. Both candidates are regular ones, so the - // reference is ambiguous. Treating the left one as hidden would instead silently resolve it - // to the right side. - val Seq(a, b) = testRelation2.output.take(2) - val set = Project(Seq(Alias(Literal("x"), "a")(), b), testRelation2) - set.setTagValue(Project.hiddenOutputTag, Seq(a, b).map(_.markAsQualifiedAccessOnly())) - set.setTagValue(LogicalPlan.PLAN_ID_TAG, 1L) - - val otherRelation = LocalRelation(AttributeReference("b", StringType)()) - val other = Project(otherRelation.output, otherRelation) - other.setTagValue(LogicalPlan.PLAN_ID_TAG, 1L) - + test("SPARK-59146: DataFrame columns that pipe SET did not assign stay visible") { + // The Project of `|> SET a = 'x'` over `t(a, b)`, as a Spark Connect relation. Each use of the + // DataFrame is planned from a new relation, with new expression IDs and the same plan id. + def set(source: LocalRelation): LogicalPlan = { + val project = Project( + Seq(UnresolvedStarExceptOrReplace( + target = None, + excepts = Seq(Seq("a")), + replacements = Some(Seq(Alias(Literal("x"), "a")())), + retainExceptedColumnsAsHidden = true)), + source.subquery("t")) + project.setTagValue(LogicalPlan.PLAN_ID_TAG, 1L) + project + } + // The DataFrame is used directly, and as the right side of a USING join on `b` that hides its + // `b`. SET retains its whole source row as hidden output, but only to keep the column order, + // so the direct candidate is regular and wins. Treating it as hidden too would make the + // reference ambiguous. + val direct = LocalRelation($"a".string, $"b".string) + val reused = Join( + LocalRelation($"b".string, $"c".int), + set(LocalRelation($"a".string, $"b".string)), + UsingJoin(Inner, Seq("b")), + None, + JoinHint.NONE) val column = UnresolvedAttribute("b") column.setTagValue(LogicalPlan.PLAN_ID_TAG, 1L) - val plan = Project(Seq(column), Join(set, other, Inner, None, JoinHint.NONE)) + val plan = Project(Seq(column), Join(set(direct), reused, Inner, None, JoinHint.NONE)) - checkError( - exception = intercept[AnalysisException](getAnalyzer.execute(plan)), - condition = "AMBIGUOUS_COLUMN_REFERENCE", - parameters = Map("name" -> "\"b\"")) + val output = getAnalyzer.execute(plan).output + assert(output.map(_.exprId) == Seq(direct.output(1).exprId), output) } test("SPARK-59146: merge hidden and visible output for a qualified star") {