You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ur...@apache.org on 2026/10/22 20:09:10 UTC
(spark) branch master updated: [SPARK-60462][SQL][CONNECT] Keep re-selected hidden join keys regular in DataFrame column resolution
This is an automated email from the ASF dual-hosted git repository.
uros-b pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 98cdf480fc31 [SPARK-60462][SQL][CONNECT] Keep re-selected hidden join keys regular in DataFrame column resolution
98cdf480fc31 is described below
commit 98cdf480fc31c4cc7f249184a4ec9b7b02a0c1a2
Author: Dejan Krakovic <de...@databricks.com>
AuthorDate: Thu Oct 22 20:08:54 2026 +0000
[SPARK-60462][SQL][CONNECT] Keep re-selected hidden join keys regular in DataFrame column resolution
### What changes were proposed in this pull request?
`resolveDataFrameColumnRecursively` tags each plan-id candidate as `hidden` or regular, and `resolveDataFrameColumn` prefers regular candidates (SPARK-56632). A candidate was tagged `hidden` when its references were a subset of `p.metadataOutput`. But an attribute can be in both `p.output` and `p.metadataOutput`: a Project that re-selects the hidden right key of a USING join inherits that key in its `metadataOutput` (`Project.metadataOutput` falls back to `child.metadataOutput`). So a [...]
This PR tags a candidate `hidden` at a node `p` when its references are not all visible at `p`, and keeps it hidden further up, as before. An attribute is visible at `p` if it is in `p.output` and `p` does not hide it itself, i.e. it is not in the part of `p.metadataOutput` that `p` introduces rather than inherits from its children:
```scala
val introducedHidden = AttributeSet(p.metadataOutput) --
p.children.flatMap(_.metadataOutput)
val visible = p.outputSet -- introducedHidden
...
val hidden = c.hidden || !c.expr.references.subsetOf(visible)
```
So a Project that re-selects a USING join's hidden right key, and a USING join wrapper above it, only inherit the key in their `metadataOutput`, and a candidate there stays regular. The USING join wrapper that hides the key keeps it hidden, also after `AddMetadataColumns` adds it to the wrapper's output for a select above, and so does a relation for a metadata column it exposes the same way. A rule based on `p.outputSet` alone would make those widened nodes count as visible, which cha [...]
Which candidates survive is unchanged (still `p.output ++ p.metadataOutput`), and so is the SPARK-56632 behavior of preferring a direct candidate over one hidden under a natural/USING join wrapper.
### Why are the changes needed?
With Spark Connect, this fails since SPARK-56632 (4.2.0):
```python
fact = spark.createDataFrame([(1, 10), (2, 20)], ["fid", "dk"])
dim = spark.createDataFrame([(10, "a"), (20, "b")], ["dk", "name"])
other = spark.createDataFrame([(1, 10, 7), (2, 30, 8)], ["fid", "dk", "qty"])
sel = fact.join(dim, "dk", "left").select(fact["fid"], dim["dk"]) # re-selects the hidden key
q = sel.join(other.select("fid", "qty"), "fid")
r = other.select("dk", "qty").join(sel, "dk") # sel's dk is hidden here
q.join(r, "fid", "left").sort("fid").select(sel["dk"]).collect()
# [AMBIGUOUS_COLUMN_REFERENCE] Column "dk" is ambiguous. ...
```
Before SPARK-56632, `sel["dk"]` resolved to the `q`-side `dk`.
### Does this PR introduce _any_ user-facing change?
Yes, a bug fix. In Spark Connect, the query above fails with `AMBIGUOUS_COLUMN_REFERENCE` in 4.2.0; with this change it resolves `sel["dk"]` to the visible candidate and returns `[10, 20]`.
### How was this patch tested?
Added `ColumnTestsMixin.test_select_reselected_hidden_join_key_with_reused_dataframe`, which runs in classic and Spark Connect. It asserts the exact rows `[10, 20]`; resolving to the hidden candidate would give `[10, NULL]`. Without the fix, the Spark Connect test fails with `AMBIGUOUS_COLUMN_REFERENCE`; the classic one passes either way, since classic resolves `df["dk"]` eagerly.
```
python/run-tests --testnames 'pyspark.sql.tests.test_column,pyspark.sql.tests.connect.test_parity_column'
```
This includes the SPARK-56632 regression test `test_select_regular_column_with_reused_dataframe_hidden_in_natural_join`. Also added two Spark Connect tests, in strict and non-strict DataFrame column resolution, for shapes where `AddMetadataColumns` widens the USING join wrapper of a re-selected hidden key: they resolve as before SPARK-56632 rather than to a different candidate or with `AMBIGUOUS_COLUMN_REFERENCE`, which a rule based on `p.outputSet` alone would give. In classic mode t [...]
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Isaac
This pull request and its description were written by Isaac.
Closes #59694 from dejankrak-db/fix-reselected-hidden-join-key.
Authored-by: Dejan Krakovic <de...@databricks.com>
Signed-off-by: Uros Bojanic <22...@users.noreply.github.com>
---
.../sql/tests/connect/test_parity_column.py | 37 ++++++++++++++++++++++
python/pyspark/sql/tests/test_column.py | 23 ++++++++++++++
.../catalyst/analysis/ColumnResolutionHelper.scala | 26 ++++++++++++---
3 files changed, 81 insertions(+), 5 deletions(-)
diff --git a/python/pyspark/sql/tests/connect/test_parity_column.py b/python/pyspark/sql/tests/connect/test_parity_column.py
index f480294b933d..48c1d847b99f 100644
--- a/python/pyspark/sql/tests/connect/test_parity_column.py
+++ b/python/pyspark/sql/tests/connect/test_parity_column.py
@@ -40,6 +40,43 @@ class ColumnParityTests(ColumnTestsMixin, ReusedConnectTestCase):
def test_validate_column_types(self):
super().test_validate_column_types()
+ # Plan-id resolution of hidden USING join keys that `AddMetadataColumns` adds to
+ # the output of the wrapper that hides them. Classic rejects these self-joins
+ # through its own ambiguity check.
+
+ def test_select_hidden_join_key_widened_by_reuse(self):
+ # A hidden USING join key that a later select re-selects is added to the
+ # output of the USING join wrapper that hides it. Such a candidate must stay
+ # hidden, so the direct candidate is still preferred.
+ a = self.spark.createDataFrame([(1, 1), (2, 1)], ["k", "x"])
+ b = self.spark.createDataFrame([(1, "p"), (2, "q")], ["k", "y"])
+ other = self.spark.createDataFrame([(1, "z")], ["x", "w"])
+ w = a.join(b.alias("b"), "k")
+ s = w.select(w["x"], w["b.k"]).join(other, "x")
+ # `w["b.k"]` resolves to the direct `w`, so both `s` rows join its row with k 1.
+ # Resolving it to the `k` in `s` would instead keep only the `s` row with k 1
+ # and join it to both `w` rows.
+ result = s.join(w, s["x"] == w["b.k"])
+ self.assertEqual(
+ sorted(tuple(r) for r in result.collect()),
+ [(1, 1, "z", 1, 1, "p"), (1, 2, "z", 1, 1, "p")],
+ )
+
+ def test_select_reused_dataframe_hidden_key_reselected_in_natural_join(self):
+ # The SPARK-56632 shape, where `enriched` re-selects the hidden `df2["id"]`,
+ # which adds it to the output of the USING join wrapper that hides it.
+ df1 = self.spark.createDataFrame([(10, "T1"), (20, "T2")], ["key", "val"])
+ df2 = self.spark.createDataFrame([(10,), (20,), (30,)], ["id"])
+ df3 = self.spark.createDataFrame([(10, "T1"), (99, "T2")], ["id", "val"])
+ enriched = df3.join(df2, "id", "left").select(df3["val"], df2["id"])
+ result = (
+ df1.join(df2, df1["key"] == df2["id"], "left")
+ .join(enriched, "val", "full_outer")
+ .sort("val")
+ .select(df2["id"])
+ )
+ self.assertEqual([r["id"] for r in result.collect()], [10, 20])
+
# Connect resolves tagged DataFrame columns by plan id, while Classic tracks
# attribute ids. Keep these tests explicit so the resulting public error
# differences are not hidden by inheritance.
diff --git a/python/pyspark/sql/tests/test_column.py b/python/pyspark/sql/tests/test_column.py
index 410a9eeb1159..a3c1d680cbd7 100644
--- a/python/pyspark/sql/tests/test_column.py
+++ b/python/pyspark/sql/tests/test_column.py
@@ -586,6 +586,29 @@ class ColumnTestsMixin:
[10, 20],
)
+ def test_select_reselected_hidden_join_key_with_reused_dataframe(self):
+ # `sel` re-selects `dim["dk"]`, the hidden right key of a USING join, so in
+ # `sel` that attribute is in both `p.output` and the `p.metadataOutput`
+ # inherited from the USING-join wrapper. `sel` is then reused: once directly
+ # (in `q`) and once as the right side of another USING join on `dk` (in `r`),
+ # where its `dk` is hidden. Resolving `sel["dk"]` must treat the `q` candidate
+ # as regular and prefer it over the hidden `r` one, not throw
+ # AMBIGUOUS_COLUMN_REFERENCE.
+ fact = self.spark.createDataFrame([(1, 10), (2, 20)], ["fid", "dk"])
+ dim = self.spark.createDataFrame([(10, "a"), (20, "b")], ["dk", "name"])
+ # No `other` row has dk 20, so `r` drops fid 2 and the left join below yields
+ # NULL on the `r` side for it. If `sel["dk"]` were resolved to the hidden `r`
+ # candidate, the second row would be NULL instead of 20.
+ other = self.spark.createDataFrame([(1, 10, 7), (2, 30, 8)], ["fid", "dk", "qty"])
+ sel = fact.join(dim, "dk", "left").select(fact["fid"], dim["dk"])
+ q = sel.join(other.select("fid", "qty"), "fid")
+ r = other.select("dk", "qty").join(sel, "dk")
+ result = q.join(r, "fid", "left").sort("fid").select(sel["dk"])
+ self.assertEqual(
+ [row["dk"] for row in result.collect()],
+ [10, 20],
+ )
+
def test_drop_notexistent_col(self):
df1 = self.spark.createDataFrame(
[("a", "b", "c")],
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 08bfcc446029..2c8daf0d27cb 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
@@ -604,9 +604,10 @@ trait ColumnResolutionHelper extends Logging with DataTypeErrorsBase {
// Candidate threaded through the plan walk:
// - expr: the resolved expression
// - depth: the depth at which it was matched
- // - hidden: whether the candidate's references live in `p.metadataOutput`
- // at some plan node along the way (e.g. a natural/USING join
- // wrapper that hides a join key via `Project.hiddenOutputTag`).
+ // - hidden: whether the candidate's references are not all visible at some
+ // plan node along the way, e.g. at a natural/USING join wrapper that
+ // hides a join key via `Project.hiddenOutputTag` (see the filter in
+ // `resolveDataFrameColumnRecursively`).
private case class Candidate(expr: NamedExpression, depth: Int, hidden: Boolean)
private def resolveDataFrameColumnByPlanId(
@@ -640,7 +641,7 @@ trait ColumnResolutionHelper extends Logging with DataTypeErrorsBase {
// case, return an empty Seq to delay the failure, so it is possible to be
// resolved in the next iteration.
// Always initialize `hidden = false` here; the filter below will set it
- // correctly based on whether the references live in `p.metadataOutput`.
+ // correctly based on whether the references are all visible at `p`.
(resolved.map(Candidate(_, currentDepth, hidden = false)).toSeq, true)
} else {
val children = p match {
@@ -688,8 +689,23 @@ trait ColumnResolutionHelper extends Logging with DataTypeErrorsBase {
// 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 candidate becomes hidden at `p` when its references are not all visible at `p`,
+ // and once hidden it stays hidden further up (`c.hidden`). An attribute is visible
+ // at `p` if it is in `p.output` and `p` does not hide it itself, i.e. it is not in
+ // the part of `p.metadataOutput` that `p` introduces rather than inherits from its
+ // children. An attribute can be in both `p.output` and `p.metadataOutput`:
+ // - A Project that re-selects a USING join's hidden right key inherits the key in
+ // its `metadataOutput` (`Project.metadataOutput` falls back to the child's), and
+ // so does a USING join wrapper above it. The key is visible at both.
+ // - `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.
+ val introducedHidden = AttributeSet(p.metadataOutput) --
+ p.children.flatMap(_.metadataOutput)
+ val visible = p.outputSet -- introducedHidden
val filtered = candidates.flatMap { c =>
- val hidden = c.hidden || c.expr.references.subsetOf(AttributeSet(p.metadataOutput))
+ val hidden = c.hidden || !c.expr.references.subsetOf(visible)
if (c.expr.references.subsetOf(AttributeSet(p.output ++ p.metadataOutput))) {
Some(c.copy(hidden = hidden))
} else {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org