You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2025/05/09 01:20:10 UTC
(spark) branch branch-4.0 updated: [SPARK-52040][PYTHON][SQL][CONNECT] ResolveLateralColumnAliasReference should retain the plan id
This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new b1b9bbf1447f [SPARK-52040][PYTHON][SQL][CONNECT] ResolveLateralColumnAliasReference should retain the plan id
b1b9bbf1447f is described below
commit b1b9bbf1447f72034ecf982fd8c196824ba75d81
Author: Ruifeng Zheng <ru...@apache.org>
AuthorDate: Fri May 9 09:18:19 2025 +0800
[SPARK-52040][PYTHON][SQL][CONNECT] ResolveLateralColumnAliasReference should retain the plan id
ResolveLateralColumnAliasReference should retain the plan id
bug fix
before:
```
In [1]: from pyspark.sql import functions as sf
In [2]: df1 = spark.range(10).select((sf.col("id") + sf.lit(1)).alias("x"), (sf.col("x") + sf.lit(1)).alias("y"))
In [3]: df2 = spark.range(10).select(sf.col("id").alias("x"))
In [4]: df1.join(df2, df1.x == df2.x).select(df1.y)
Out[4]: 25/05/08 16:38:28 ERROR ErrorUtils: Spark Connect RPC error during: analyze. UserId: ruifeng.zheng. SessionId: af3deba7-1e48-49fd-adad-2046a72ed341.
org.apache.spark.sql.AnalysisException: [CANNOT_RESOLVE_DATAFRAME_COLUMN] Cannot resolve dataframe column "y". It's probably because of illegal references like `df1.select(df2.col("a"))`. SQLSTATE: 42704
at org.apache.spark.sql.errors.QueryCompilationErrors$.cannotResolveDataFrameColumn(QueryCompilationErrors.scala:4147)
at org.apache.spark.sql.catalyst.analysis.ColumnResolutionHelper.resolveDataFrameColumn(ColumnResolutionHelper.scala:562)
at org.apache.spark.sql.catalyst.analysis.ColumnResolutionHelper.tryResolveDataFrameColumns(ColumnResolutionHelper.scala:537)
```
after:
```
In [1]: from pyspark.sql import functions as sf
In [2]: df1 = spark.range(10).select((sf.col("id") + sf.lit(1)).alias("x"), (sf.col("x") + sf.lit(1)).alias("y"))
In [3]: df2 = spark.range(10).select(sf.col("id").alias("x"))
In [4]: df1.join(df2, df1.x == df2.x).select(df1.y).show()
+---+
| y|
+---+
| 2|
| 3|
| 4|
| 5|
| 6|
| 7|
| 8|
| 9|
| 10|
+---+
```
yes, above query works after this change
added test
no
Closes #50831 from zhengruifeng/fix_lca.
Authored-by: Ruifeng Zheng <ru...@apache.org>
Signed-off-by: Wenchen Fan <we...@databricks.com>
(cherry picked from commit 688281ae13ecfaf5111e28ef54d6c8163c2fda99)
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
python/pyspark/sql/tests/test_dataframe.py | 9 +++++++++
.../catalyst/analysis/ResolveLateralColumnAliasReference.scala | 8 ++++++--
2 files changed, 15 insertions(+), 2 deletions(-)
diff --git a/python/pyspark/sql/tests/test_dataframe.py b/python/pyspark/sql/tests/test_dataframe.py
index e20ab09d8d1e..890ae56ffa52 100644
--- a/python/pyspark/sql/tests/test_dataframe.py
+++ b/python/pyspark/sql/tests/test_dataframe.py
@@ -148,6 +148,15 @@ class DataFrameTestsMixin:
self.assertTrue(df3.columns, ["id", "value", "id", "value"])
self.assertTrue(df3.count() == 20)
+ def test_lateral_column_alias(self):
+ df1 = self.spark.range(10).select(
+ (col("id") + lit(1)).alias("x"), (col("x") + lit(1)).alias("y")
+ )
+ df2 = self.spark.range(10).select(col("id").alias("x"))
+ df3 = df1.join(df2, df1.x == df2.x).select(df1.y)
+ self.assertTrue(df3.columns, ["y"])
+ self.assertTrue(df3.count() == 9)
+
def test_duplicated_column_names(self):
df = self.spark.createDataFrame([(1, 2)], ["c", "c"])
row = df.select("*").first()
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveLateralColumnAliasReference.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveLateralColumnAliasReference.scala
index cb26820a0c79..931f9bda78a1 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveLateralColumnAliasReference.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveLateralColumnAliasReference.scala
@@ -190,10 +190,12 @@ object ResolveLateralColumnAliasReference extends Rule[LogicalPlan] {
outerProjectList.update(idx, alias.toAttribute)
innerProjectList += alias
}
- p.copy(
+ val newProject = p.copy(
projectList = outerProjectList.toSeq,
child = Project(innerProjectList.toSeq, child)
)
+ newProject.copyTagsFrom(p)
+ newProject
}
case aggOriginal: Aggregate
@@ -264,10 +266,12 @@ object ResolveLateralColumnAliasReference extends Rule[LogicalPlan] {
}
val projectExprs = aggregateExpressions.map(
extractExpressions(_).asInstanceOf[NamedExpression])
- Project(
+ val newProject = Project(
projectList = projectExprs,
child = agg.copy(aggregateExpressions = newAggExprs.asScala.toSeq)
)
+ newProject.copyTagsFrom(agg)
+ newProject
}
case p: LogicalPlan =>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org