cloud-fan commented on code in PR #58760:
URL: https://github.com/apache/spark/pull/58760#discussion_r4057040519
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -2085,6 +2086,45 @@ 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).
+ *
+ * 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 {
+ case s: UnresolvedStarExceptOrReplace =>
s.retainExceptedColumnsAsHidden
+ case _ => false
+ }
+ 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)
+ .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(
Review Comment:
**Non-blocking (P2):** This tag becomes correctness-critical across analyzer
iterations: `ResolveReferences` can resolve `t.a` from metadata only after
`AddMetadataColumns` has already run, so the attribute is materialized on a
later iteration. Generic `copyTagsFrom` calls do not guarantee that
`hiddenOutputTag` survives every intervening Project replacement. Please keep
the retained source-row state at a durable owner, or materialize referenced
source attributes within a bounded operation, so a documented qualified SET
reference cannot lose its producing child.
See **Shared repair plan 1** in the review body.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -2085,6 +2086,45 @@ 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).
+ *
+ * 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 {
+ case s: UnresolvedStarExceptOrReplace =>
s.retainExceptedColumnsAsHidden
+ case _ => false
+ }
+ 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)
+ .filter(_.qualifier.nonEmpty)
+ .distinctBy(_.exprId)
Review Comment:
**Blocking (P1):** Project output is positional, so repeated projections can
legitimately share an ExprId. Here `distinctBy(_.exprId)` drops one such
position, and the later ExprId set removes every matching visible occurrence.
For example, `VALUES (1, 2) AS s(a, b) |> SELECT a, a, b |> AS t |> SET b = 3
|> SELECT t.*` expands to `[a, b]` instead of `[a, a, b]`, silently changing
the schema and row shape. Please preserve source-row occurrences rather than
using ExprId as positional identity.
See **Shared repair plan 1** in the review body.
##########
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) &&
Review Comment:
**Non-blocking (P2):** The new test stops at tag coexistence and never
constructs a plan-ID-bound column reference that reaches this visible/hidden
overlap. Reverting this predicate would therefore leave the added assertions
green while allowing the Spark Connect/DataFrame-column failure to return.
Please add a focused analyzer case that invokes plan-ID resolution for an
attribute present in both outputs and verifies that the visible candidate wins.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -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(
Review Comment:
Verified on the current commit: the original Project tags are copied before
hiddenOutputTag is installed, and the focused analyzer test preserves both
PLAN_ID_TAG and hiddenOutputTag. This issue is resolved.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4047867586","thread_id":"inline:4047867586","verdict_sha256":"7645db3e2352c29f0760c1b9e840cd6ebca5f4c507fcea390aca07845cab8561"}
-->
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -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)
Review Comment:
Verified on the current commit: t.* now follows the original source-row
order, including across repeated SET operations. This issue is resolved.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4047867596","thread_id":"inline:4047867596","verdict_sha256":"7645db3e2352c29f0760c1b9e840cd6ebca5f4c507fcea390aca07845cab8561"}
-->
--
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]
