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/02/25 01:39:25 UTC
[spark] branch branch-3.4 updated: [SPARK-42568][CONNECT] Fix SparkConnectStreamHandler to handle configs properly while planning
This is an automated email from the ASF dual-hosted git repository.
gurwls223 pushed a commit to branch branch-3.4
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.4 by this push:
new 40e6b9d3623 [SPARK-42568][CONNECT] Fix SparkConnectStreamHandler to handle configs properly while planning
40e6b9d3623 is described below
commit 40e6b9d3623d81b6fdb0ac8de16aab7abb3dd3b5
Author: Takuya UESHIN <ue...@databricks.com>
AuthorDate: Sat Feb 25 10:38:53 2023 +0900
[SPARK-42568][CONNECT] Fix SparkConnectStreamHandler to handle configs properly while planning
### What changes were proposed in this pull request?
Fixes `SparkConnectStreamHandler` to handle configs properly while planning.
The whole process should be done in `session.withActive` to take the proper `SQLConf` into account.
### Why are the changes needed?
Some components for planning need to check configs in `SQLConf.get` while building the plan, but currently it's unavailable.
For example, `spark.sql.legacy.allowNegativeScaleOfDecimal` needs to check when construct `DecimalType` but it's not set while planning, thus it causes an error when trying to cast to `DecimalType(1, -1)` with the config set to `"true"`:
```
[INTERNAL_ERROR] Negative scale is not allowed: -1. Set the config "spark.sql.legacy.allowNegativeScaleOfDecimal" to "true" to allow it.
```
### Does this PR introduce _any_ user-facing change?
The configs will take effect while planning.
### How was this patch tested?
Enabled a related test.
Closes #40165 from ueshin/issues/SPARK-42568/withActive.
Authored-by: Takuya UESHIN <ue...@databricks.com>
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
(cherry picked from commit b99435328f76febd36dadd93649d1b3c10fe03da)
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
---
.../sql/connect/service/SparkConnectStreamHandler.scala | 12 +++++++-----
python/pyspark/sql/connect/types.py | 3 ++-
python/pyspark/sql/tests/connect/test_parity_types.py | 5 -----
python/pyspark/sql/tests/test_types.py | 4 ++--
4 files changed, 11 insertions(+), 13 deletions(-)
diff --git a/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectStreamHandler.scala b/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectStreamHandler.scala
index cff3a451ceb..f46aca9c8cf 100644
--- a/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectStreamHandler.scala
+++ b/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectStreamHandler.scala
@@ -44,11 +44,13 @@ class SparkConnectStreamHandler(responseObserver: StreamObserver[ExecutePlanResp
SparkConnectService
.getOrCreateIsolatedSession(v.getUserContext.getUserId, v.getClientId)
.session
- v.getPlan.getOpTypeCase match {
- case proto.Plan.OpTypeCase.COMMAND => handleCommand(session, v)
- case proto.Plan.OpTypeCase.ROOT => handlePlan(session, v)
- case _ =>
- throw new UnsupportedOperationException(s"${v.getPlan.getOpTypeCase} not supported.")
+ session.withActive {
+ v.getPlan.getOpTypeCase match {
+ case proto.Plan.OpTypeCase.COMMAND => handleCommand(session, v)
+ case proto.Plan.OpTypeCase.ROOT => handlePlan(session, v)
+ case _ =>
+ throw new UnsupportedOperationException(s"${v.getPlan.getOpTypeCase} not supported.")
+ }
}
}
diff --git a/python/pyspark/sql/connect/types.py b/python/pyspark/sql/connect/types.py
index c73aa43d0ca..33ddd4e9929 100644
--- a/python/pyspark/sql/connect/types.py
+++ b/python/pyspark/sql/connect/types.py
@@ -87,7 +87,8 @@ def pyspark_types_to_proto_types(data_type: DataType) -> pb2.DataType:
elif isinstance(data_type, DoubleType):
ret.double.CopyFrom(pb2.DataType.Double())
elif isinstance(data_type, DecimalType):
- ret.decimal.CopyFrom(pb2.DataType.Decimal())
+ ret.decimal.scale = data_type.scale
+ ret.decimal.precision = data_type.precision
elif isinstance(data_type, DateType):
ret.date.CopyFrom(pb2.DataType.Date())
elif isinstance(data_type, TimestampType):
diff --git a/python/pyspark/sql/tests/connect/test_parity_types.py b/python/pyspark/sql/tests/connect/test_parity_types.py
index e966986c152..3d54d488a5d 100644
--- a/python/pyspark/sql/tests/connect/test_parity_types.py
+++ b/python/pyspark/sql/tests/connect/test_parity_types.py
@@ -113,11 +113,6 @@ class TypesParityTests(TypesTestsMixin, ReusedConnectTestCase):
def test_infer_schema_with_udt(self):
super().test_infer_schema_with_udt()
- # TODO(SPARK-41834): Implement SparkSession.conf
- @unittest.skip("Fails in Spark Connect, should enable.")
- def test_negative_decimal(self):
- super().test_negative_decimal()
-
# TODO(SPARK-42020): createDataFrame with UDT
@unittest.skip("Fails in Spark Connect, should enable.")
def test_nested_udt_in_df(self):
diff --git a/python/pyspark/sql/tests/test_types.py b/python/pyspark/sql/tests/test_types.py
index 9db090fa810..68424cad386 100644
--- a/python/pyspark/sql/tests/test_types.py
+++ b/python/pyspark/sql/tests/test_types.py
@@ -379,13 +379,13 @@ class TypesTestsMixin:
def test_negative_decimal(self):
try:
- self.spark.sql("set spark.sql.legacy.allowNegativeScaleOfDecimal=true")
+ self.spark.sql("set spark.sql.legacy.allowNegativeScaleOfDecimal=true").collect()
df = self.spark.createDataFrame([(1,), (11,)], ["value"])
ret = df.select(col("value").cast(DecimalType(1, -1))).collect()
actual = list(map(lambda r: int(r.value), ret))
self.assertEqual(actual, [0, 10])
finally:
- self.spark.sql("set spark.sql.legacy.allowNegativeScaleOfDecimal=false")
+ self.spark.sql("set spark.sql.legacy.allowNegativeScaleOfDecimal=false").collect()
def test_create_dataframe_from_objects(self):
data = [MyObject(1, "1"), MyObject(2, "2")]
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org