You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by gu...@apache.org on 2023/08/03 02:58:00 UTC
[spark] branch branch-3.5 updated: [SPARK-44620][SQL][PS][CONNECT] Make `ResolvePivot` retain the `Plan_ID_TAG`
This is an automated email from the ASF dual-hosted git repository.
gurwls223 pushed a commit to branch branch-3.5
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.5 by this push:
new 5cfd85fcf59 [SPARK-44620][SQL][PS][CONNECT] Make `ResolvePivot` retain the `Plan_ID_TAG`
5cfd85fcf59 is described below
commit 5cfd85fcf5989cd9a3a083d712b0b800a2a92409
Author: Ruifeng Zheng <ru...@apache.org>
AuthorDate: Thu Aug 3 11:57:34 2023 +0900
[SPARK-44620][SQL][PS][CONNECT] Make `ResolvePivot` retain the `Plan_ID_TAG`
### What changes were proposed in this pull request?
Make `ResolvePivot` retain the `Plan_ID_TAG`
### Why are the changes needed?
to resolve the `AnalysisException` in Pandas APIs on Connect
### Does this PR introduce _any_ user-facing change?
yes, new APIs enabled:
1. `frame.pivot_table`
2. `frame.transpose`
3. `series.unstack`
### How was this patch tested?
enabled UTs
Closes #42261 from zhengruifeng/ps_connect_analyze_pivot.
Authored-by: Ruifeng Zheng <ru...@apache.org>
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
(cherry picked from commit 1a89bdc60d55394a1a9d94d4fa69fa5ab8041671)
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
---
.../tests/connect/computation/test_parity_pivot.py | 17 +----------------
.../pandas/tests/connect/frame/test_parity_reshaping.py | 11 +----------
.../pandas/tests/connect/series/test_parity_compute.py | 6 +-----
.../pandas/tests/connect/test_parity_categorical.py | 6 ------
.../apache/spark/sql/catalyst/analysis/Analyzer.scala | 10 +++++++---
5 files changed, 10 insertions(+), 40 deletions(-)
diff --git a/python/pyspark/pandas/tests/connect/computation/test_parity_pivot.py b/python/pyspark/pandas/tests/connect/computation/test_parity_pivot.py
index d2c4f9ae607..c8ec48eb06a 100644
--- a/python/pyspark/pandas/tests/connect/computation/test_parity_pivot.py
+++ b/python/pyspark/pandas/tests/connect/computation/test_parity_pivot.py
@@ -16,28 +16,13 @@
#
import unittest
-from pyspark import pandas as ps
from pyspark.pandas.tests.computation.test_pivot import FramePivotMixin
from pyspark.testing.connectutils import ReusedConnectTestCase
from pyspark.testing.pandasutils import PandasOnSparkTestUtils
class FrameParityPivotTests(FramePivotMixin, PandasOnSparkTestUtils, ReusedConnectTestCase):
- @property
- def psdf(self):
- return ps.from_pandas(self.pdf)
-
- @unittest.skip(
- "TODO(SPARK-43611): Fix unexpected `AnalysisException` from Spark Connect client."
- )
- def test_pivot_table(self):
- super().test_pivot_table()
-
- @unittest.skip(
- "TODO(SPARK-43611): Fix unexpected `AnalysisException` from Spark Connect client."
- )
- def test_pivot_table_dtypes(self):
- super().test_pivot_table_dtypes()
+ pass
if __name__ == "__main__":
diff --git a/python/pyspark/pandas/tests/connect/frame/test_parity_reshaping.py b/python/pyspark/pandas/tests/connect/frame/test_parity_reshaping.py
index 98ebf3ca44a..e4bac7b078e 100644
--- a/python/pyspark/pandas/tests/connect/frame/test_parity_reshaping.py
+++ b/python/pyspark/pandas/tests/connect/frame/test_parity_reshaping.py
@@ -16,22 +16,13 @@
#
import unittest
-from pyspark import pandas as ps
from pyspark.pandas.tests.frame.test_reshaping import FrameReshapingMixin
from pyspark.testing.connectutils import ReusedConnectTestCase
from pyspark.testing.pandasutils import PandasOnSparkTestUtils
class FrameParityReshapingTests(FrameReshapingMixin, PandasOnSparkTestUtils, ReusedConnectTestCase):
- @property
- def psdf(self):
- return ps.from_pandas(self.pdf)
-
- @unittest.skip(
- "TODO(SPARK-43611): Fix unexpected `AnalysisException` from Spark Connect client."
- )
- def test_transpose(self):
- super().test_transpose()
+ pass
if __name__ == "__main__":
diff --git a/python/pyspark/pandas/tests/connect/series/test_parity_compute.py b/python/pyspark/pandas/tests/connect/series/test_parity_compute.py
index f757d19ca69..8876fcb1398 100644
--- a/python/pyspark/pandas/tests/connect/series/test_parity_compute.py
+++ b/python/pyspark/pandas/tests/connect/series/test_parity_compute.py
@@ -22,11 +22,7 @@ from pyspark.testing.pandasutils import PandasOnSparkTestUtils
class SeriesParityComputeTests(SeriesComputeMixin, PandasOnSparkTestUtils, ReusedConnectTestCase):
- @unittest.skip(
- "TODO(SPARK-43611): Fix unexpected `AnalysisException` from Spark Connect client."
- )
- def test_unstack(self):
- super().test_unstack()
+ pass
if __name__ == "__main__":
diff --git a/python/pyspark/pandas/tests/connect/test_parity_categorical.py b/python/pyspark/pandas/tests/connect/test_parity_categorical.py
index 3e05eb2c0f3..210cfce8ddb 100644
--- a/python/pyspark/pandas/tests/connect/test_parity_categorical.py
+++ b/python/pyspark/pandas/tests/connect/test_parity_categorical.py
@@ -53,12 +53,6 @@ class CategoricalParityTests(
def test_set_categories(self):
super().test_set_categories()
- @unittest.skip(
- "TODO(SPARK-43611): Fix unexpected `AnalysisException` from Spark Connect client."
- )
- def test_unstack(self):
- super().test_unstack()
-
if __name__ == "__main__":
from pyspark.pandas.tests.connect.test_parity_categorical import * # noqa: F401
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 7f2471c9e19..09467c22e2b 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
@@ -759,7 +759,7 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
case p: Pivot if !p.childrenResolved || !p.aggregates.forall(_.resolved)
|| (p.groupByExprsOpt.isDefined && !p.groupByExprsOpt.get.forall(_.resolved))
|| !p.pivotColumn.resolved || !p.pivotValues.forall(_.resolved) => p
- case Pivot(groupByExprsOpt, pivotColumn, pivotValues, aggregates, child) =>
+ case p @ Pivot(groupByExprsOpt, pivotColumn, pivotValues, aggregates, child) =>
if (!RowOrdering.isOrderable(pivotColumn.dataType)) {
throw QueryCompilationErrors.unorderablePivotColError(pivotColumn)
}
@@ -823,7 +823,9 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
Alias(ExtractValue(pivotAtt, Literal(i), resolver), outputName(value, aggregate))()
}
}
- Project(groupByExprsAttr ++ pivotOutputs, secondAgg)
+ val newProject = Project(groupByExprsAttr ++ pivotOutputs, secondAgg)
+ newProject.copyTagsFrom(p)
+ newProject
} else {
val pivotAggregates: Seq[NamedExpression] = pivotValues.flatMap { value =>
def ifExpr(e: Expression) = {
@@ -857,7 +859,9 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
Alias(filteredAggregate, outputName(value, aggregate))()
}
}
- Aggregate(groupByExprs, groupByExprs ++ pivotAggregates, child)
+ val newAggregate = Aggregate(groupByExprs, groupByExprs ++ pivotAggregates, child)
+ newAggregate.copyTagsFrom(p)
+ newAggregate
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org