You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ru...@apache.org on 2022/10/17 02:28:28 UTC
[spark] branch master updated: [SPARK-40534][CONNECT][FOLLOW-UP] Support all join types in python client
This is an automated email from the ASF dual-hosted git repository.
ruifengz 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 116f6a719ea [SPARK-40534][CONNECT][FOLLOW-UP] Support all join types in python client
116f6a719ea is described below
commit 116f6a719eaf1220d8eed98433f724fca672acc6
Author: Rui Wang <ru...@databricks.com>
AuthorDate: Mon Oct 17 10:27:58 2022 +0800
[SPARK-40534][CONNECT][FOLLOW-UP] Support all join types in python client
### What changes were proposed in this pull request?
Following up https://github.com/apache/spark/pull/38157, add all join types support for python. Please note that in the PR we decided to not support CROSS Join now.
### Why are the changes needed?
Finish the join type support in Connect.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
UT
Closes #38243 from amaliujia/python_join_types.
Authored-by: Rui Wang <ru...@databricks.com>
Signed-off-by: Ruifeng Zheng <ru...@apache.org>
---
python/pyspark/sql/connect/dataframe.py | 2 +-
python/pyspark/sql/connect/plan.py | 32 +++++++++++++++++++---
.../sql/tests/connect/test_connect_select_ops.py | 18 +++++++++++-
3 files changed, 46 insertions(+), 6 deletions(-)
diff --git a/python/pyspark/sql/connect/dataframe.py b/python/pyspark/sql/connect/dataframe.py
index bb74282cfda..31215b4da79 100644
--- a/python/pyspark/sql/connect/dataframe.py
+++ b/python/pyspark/sql/connect/dataframe.py
@@ -183,7 +183,7 @@ class DataFrame(object):
return self.toPandas()
# TODO(martin.grund) fix mypu
- def join(self, other: "DataFrame", on: Any, how: Any = None) -> "DataFrame":
+ def join(self, other: "DataFrame", on: Any, how: Optional[str] = None) -> "DataFrame":
if self._plan is None:
raise Exception("Cannot join when self._plan is empty.")
if other._plan is None:
diff --git a/python/pyspark/sql/connect/plan.py b/python/pyspark/sql/connect/plan.py
index 67ed6b964fa..486778b9d37 100644
--- a/python/pyspark/sql/connect/plan.py
+++ b/python/pyspark/sql/connect/plan.py
@@ -368,21 +368,45 @@ class Join(LogicalPlan):
left: Optional["LogicalPlan"],
right: "LogicalPlan",
on: "ColumnOrString",
- how: proto.Join.JoinType.ValueType = proto.Join.JoinType.JOIN_TYPE_INNER,
+ how: Optional[str],
) -> None:
super().__init__(left)
self.left = cast(LogicalPlan, left)
self.right = right
self.on = on
if how is None:
- how = proto.Join.JoinType.JOIN_TYPE_INNER
- self.how = how
+ join_type = proto.Join.JoinType.JOIN_TYPE_INNER
+ elif how == "inner":
+ join_type = proto.Join.JoinType.JOIN_TYPE_INNER
+ elif how in ["outer", "full", "fullouter"]:
+ join_type = proto.Join.JoinType.JOIN_TYPE_FULL_OUTER
+ elif how in ["leftouter", "left"]:
+ join_type = proto.Join.JoinType.JOIN_TYPE_LEFT_OUTER
+ elif how in ["rightouter", "right"]:
+ join_type = proto.Join.JoinType.JOIN_TYPE_RIGHT_OUTER
+ elif how in ["leftsemi", "semi"]:
+ join_type = proto.Join.JoinType.JOIN_TYPE_LEFT_SEMI
+ elif how in ["leftanti", "anti"]:
+ join_type = proto.Join.JoinType.JOIN_TYPE_LEFT_ANTI
+ else:
+ raise NotImplementedError(
+ """
+ Unsupported join type: %s. Supported join types include:
+ "inner", "outer", "full", "fullouter", "full_outer",
+ "leftouter", "left", "left_outer", "rightouter",
+ "right", "right_outer", "leftsemi", "left_semi",
+ "semi", "leftanti", "left_anti", "anti",
+ """
+ % how
+ )
+ self.how = join_type
def plan(self, session: Optional["RemoteSparkSession"]) -> proto.Relation:
rel = proto.Relation()
rel.join.left.CopyFrom(self.left.plan(session))
rel.join.right.CopyFrom(self.right.plan(session))
- rel.join.on.CopyFrom(self.to_attr_or_expression(self.on, session))
+ rel.join.join_condition.CopyFrom(self.to_attr_or_expression(self.on, session))
+ rel.join.join_type = self.how
return rel
def print(self, indent: int = 0) -> str:
diff --git a/python/pyspark/sql/tests/connect/test_connect_select_ops.py b/python/pyspark/sql/tests/connect/test_connect_select_ops.py
index f7f164c11db..37a64abcc5e 100644
--- a/python/pyspark/sql/tests/connect/test_connect_select_ops.py
+++ b/python/pyspark/sql/tests/connect/test_connect_select_ops.py
@@ -18,14 +18,30 @@ from pyspark.testing.connectutils import PlanOnlyTestFixture
from pyspark.sql.connect import DataFrame
from pyspark.sql.connect.functions import col
from pyspark.sql.connect.plan import Read, InputValidationError
+import pyspark.sql.connect.proto as proto
-class SparkConnectSelectOpsSuite(PlanOnlyTestFixture):
+class SparkConnectToProtoSuite(PlanOnlyTestFixture):
def test_select_with_literal(self):
df = DataFrame.withPlan(Read("table"))
self.assertIsNotNone(df.select(col("name"))._plan.collect())
self.assertRaises(InputValidationError, df.select, "name")
+ def test_join_with_join_type(self):
+ df_left = DataFrame.withPlan(Read("table"))
+ df_right = DataFrame.withPlan(Read("table"))
+ for (join_type_str, join_type) in [
+ (None, proto.Join.JoinType.JOIN_TYPE_INNER),
+ ("inner", proto.Join.JoinType.JOIN_TYPE_INNER),
+ ("outer", proto.Join.JoinType.JOIN_TYPE_FULL_OUTER),
+ ("leftouter", proto.Join.JoinType.JOIN_TYPE_LEFT_OUTER),
+ ("rightouter", proto.Join.JoinType.JOIN_TYPE_RIGHT_OUTER),
+ ("leftanti", proto.Join.JoinType.JOIN_TYPE_LEFT_ANTI),
+ ("leftsemi", proto.Join.JoinType.JOIN_TYPE_LEFT_SEMI),
+ ]:
+ joined_df = df_left.join(df_right, on=col("name"), how=join_type_str)._plan.collect()
+ self.assertEqual(joined_df.root.join.join_type, join_type)
+
if __name__ == "__main__":
import unittest
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org