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/01/29 01:03:26 UTC
[spark] branch branch-3.4 updated: [SPARK-42224][CONNECT] Migrate `TypeError` into error framework for Spark Connect functions
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 1923dfb8a71 [SPARK-42224][CONNECT] Migrate `TypeError` into error framework for Spark Connect functions
1923dfb8a71 is described below
commit 1923dfb8a71aab50c246505481c54c58a4a57591
Author: itholic <ha...@databricks.com>
AuthorDate: Sun Jan 29 10:03:06 2023 +0900
[SPARK-42224][CONNECT] Migrate `TypeError` into error framework for Spark Connect functions
### What changes were proposed in this pull request?
This PR proposes to migrate `TypeError` into error framework for Spark Connect functions.
### Why are the changes needed?
To improve errors by leveraging the PySpark error framework for Spark Connect functions.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Fixed & added UTs.
Closes #39782 from itholic/SPARK-42224.
Authored-by: itholic <ha...@databricks.com>
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
(cherry picked from commit dbdb06ccbcd00d6e3254151e1e27f55cfc796aea)
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
---
python/pyspark/errors/error_classes.py | 27 ++++++++---
python/pyspark/sql/connect/functions.py | 49 ++++++++++++++------
.../sql/tests/connect/test_connect_function.py | 54 ++++++++++++++++++++++
python/pyspark/sql/tests/test_functions.py | 26 +++++------
4 files changed, 123 insertions(+), 33 deletions(-)
diff --git a/python/pyspark/errors/error_classes.py b/python/pyspark/errors/error_classes.py
index 77f74097880..dd86f06505b 100644
--- a/python/pyspark/errors/error_classes.py
+++ b/python/pyspark/errors/error_classes.py
@@ -21,7 +21,7 @@ ERROR_CLASSES_JSON = """
{
"COLUMN_IN_LIST": {
"message": [
- "<func_name> does not allow a column in a list."
+ "<func_name> does not allow a Column in a list."
]
},
"HIGHER_ORDER_FUNCTION_SHOULD_RETURN_COLUMN" : {
@@ -31,27 +31,42 @@ ERROR_CLASSES_JSON = """
},
"NOT_A_COLUMN" : {
"message" : [
- "Argument `<arg_name>` should be a column, got <arg_type>."
+ "Argument `<arg_name>` should be a Column, got <arg_type>."
+ ]
+ },
+ "NOT_A_DATAFRAME" : {
+ "message" : [
+ "Argument `<arg_name>` must be a DataFrame, got <arg_type>."
]
},
"NOT_A_STRING" : {
"message" : [
- "Argument `<arg_name>` should be a string, got <arg_type>."
+ "Argument `<arg_name>` should be a str, got <arg_type>."
+ ]
+ },
+ "NOT_COLUMN_OR_DATATYPE_OR_STRING" : {
+ "message" : [
+ "Argument `<arg_name>` should be a Column or str or DataType, but got <arg_type>."
]
},
"NOT_COLUMN_OR_INTEGER" : {
"message" : [
- "Argument `<arg_name>` should be a column or integer, got <arg_type>."
+ "Argument `<arg_name>` should be a Column or int, got <arg_type>."
]
},
"NOT_COLUMN_OR_INTEGER_OR_STRING" : {
"message" : [
- "Argument `<arg_name>` should be a column or integer or string, got <arg_type>."
+ "Argument `<arg_name>` should be a Column, int or str, got <arg_type>."
]
},
"NOT_COLUMN_OR_STRING" : {
"message" : [
- "Argument `<arg_name>` should be a column or string, got <arg_type>."
+ "Argument `<arg_name>` should be a Column or str, got <arg_type>."
+ ]
+ },
+ "UNSUPPORTED_NUMPY_ARRAY_SCALAR" : {
+ "message" : [
+ "The type of array scalar '<dtype>' is not supported."
]
},
"UNSUPPORTED_PARAM_TYPE_FOR_HIGHER_ORDER_FUNCTION" : {
diff --git a/python/pyspark/sql/connect/functions.py b/python/pyspark/sql/connect/functions.py
index 7c21f9280c2..3b3036f3be8 100644
--- a/python/pyspark/sql/connect/functions.py
+++ b/python/pyspark/sql/connect/functions.py
@@ -223,7 +223,10 @@ def lit(col: Any) -> Column:
return array(*[lit(c) for c in col])
elif isinstance(col, np.ndarray) and col.ndim == 1:
if _from_numpy_type(col.dtype) is None:
- raise TypeError("The type of array scalar '%s' is not supported" % (col.dtype))
+ raise PySparkTypeError(
+ error_class="UNSUPPORTED_NUMPY_ARRAY_SCALAR",
+ message_parameters={"dtype": col.dtype.name},
+ )
# NumpyArrayConverter for Py4J can not support ndarray with int8 values.
# Actually this is not a problem for Connect, but here still convert it
@@ -258,7 +261,10 @@ def broadcast(df: "DataFrame") -> "DataFrame":
from pyspark.sql.connect.dataframe import DataFrame
if not isinstance(df, DataFrame):
- raise TypeError(f"'df' must be a DataFrame, but got {type(df).__name__} {df}")
+ raise PySparkTypeError(
+ error_class="NOT_A_DATAFRAME",
+ message_parameters={"arg_name": "df", "arg_type": type(df).__name__},
+ )
return df.hint("broadcast")
@@ -1376,8 +1382,9 @@ def from_json(
elif isinstance(schema, str):
_schema = lit(schema)
else:
- raise TypeError(
- f"schema should be a Column or str or DataType, but got {type(schema).__name__}"
+ raise PySparkTypeError(
+ error_class="NOT_COLUMN_OR_DATATYPE_OR_STRING",
+ message_parameters={"arg_name": "schema", "arg_type": type(schema).__name__},
)
if options is None:
@@ -1592,14 +1599,20 @@ def slice(
elif isinstance(start, int):
_start = lit(start)
else:
- raise TypeError(f"start should be a Column, str or int, but got {type(start).__name__}")
+ raise PySparkTypeError(
+ error_class="NOT_COLUMN_OR_INTEGER_OR_STRING",
+ message_parameters={"arg_name": "start", "arg_type": type(start).__name__},
+ )
if isinstance(length, (Column, str)):
_length = length
elif isinstance(length, int):
_length = lit(length)
else:
- raise TypeError(f"start should be a Column, str or int, but got {type(length).__name__}")
+ raise PySparkTypeError(
+ error_class="NOT_COLUMN_OR_INTEGER_OR_STRING",
+ message_parameters={"arg_name": "length", "arg_type": type(length).__name__},
+ )
return _invoke_function_over_columns("slice", col, _start, _length)
@@ -2218,13 +2231,17 @@ def window(
},
)
if slideDuration is not None and not isinstance(slideDuration, str):
- raise TypeError(
- f"slideDuration should be as a string, "
- f"but got {type(slideDuration).__name__} {slideDuration}"
+ raise PySparkTypeError(
+ error_class="NOT_A_STRING",
+ message_parameters={
+ "arg_name": "slideDuration",
+ "arg_type": type(slideDuration).__name__,
+ },
)
if startTime is not None and not isinstance(startTime, str):
- raise TypeError(
- f"startTime should be as a string, " f"but got {type(startTime).__name__} {startTime}"
+ raise PySparkTypeError(
+ error_class="NOT_A_STRING",
+ message_parameters={"arg_name": "startTime", "arg_type": type(startTime).__name__},
)
time_col = _to_col(timeColumn)
@@ -2330,7 +2347,10 @@ def assert_true(col: "ColumnOrName", errMsg: Optional[Union[Column, str]] = None
if errMsg is None:
return _invoke_function_over_columns("assert_true", col)
if not isinstance(errMsg, (str, Column)):
- raise TypeError("errMsg should be a Column or a str, got {}".format(type(errMsg)))
+ raise PySparkTypeError(
+ error_class="NOT_COLUMN_OR_STRING",
+ message_parameters={"arg_name": "errMsg", "arg_type": type(errMsg).__name__},
+ )
_err_msg = lit(errMsg) if isinstance(errMsg, str) else _to_col(errMsg)
return _invoke_function("assert_true", _to_col(col), _err_msg)
@@ -2340,7 +2360,10 @@ assert_true.__doc__ = pysparkfuncs.assert_true.__doc__
def raise_error(errMsg: Union[Column, str]) -> Column:
if not isinstance(errMsg, (str, Column)):
- raise TypeError("errMsg should be a Column or a str, got {}".format(type(errMsg)))
+ raise PySparkTypeError(
+ error_class="NOT_COLUMN_OR_STRING",
+ message_parameters={"arg_name": "errMsg", "arg_type": type(errMsg).__name__},
+ )
_err_msg = lit(errMsg) if isinstance(errMsg, str) else _to_col(errMsg)
return _invoke_function("raise_error", _err_msg)
diff --git a/python/pyspark/sql/tests/connect/test_connect_function.py b/python/pyspark/sql/tests/connect/test_connect_function.py
index 50fadb49ed4..0832c9739d6 100644
--- a/python/pyspark/sql/tests/connect/test_connect_function.py
+++ b/python/pyspark/sql/tests/connect/test_connect_function.py
@@ -164,6 +164,15 @@ class SparkConnectFunctionTests(ReusedConnectTestCase, PandasOnSparkTestUtils, S
SF.broadcast(sdf1).join(SF.broadcast(sdf2), on="a").toPandas(),
)
+ with self.assertRaises(PySparkTypeError) as pe:
+ CF.broadcast(cdf.a)
+
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_A_DATAFRAME",
+ message_parameters={"arg_name": "df", "arg_type": "Column"},
+ )
+
def test_normal_functions(self):
from pyspark.sql import functions as SF
from pyspark.sql.connect import functions as CF
@@ -1168,6 +1177,24 @@ class SparkConnectFunctionTests(ReusedConnectTestCase, PandasOnSparkTestUtils, S
sdf.select(SF.slice(sdf.a, 1, 2), SF.slice("c", 2, 3)).toPandas(),
)
+ with self.assertRaises(PySparkTypeError) as pe:
+ CF.slice(cdf.a, 1.0, 2)
+
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_COLUMN_OR_INTEGER_OR_STRING",
+ message_parameters={"arg_name": "start", "arg_type": "float"},
+ )
+
+ with self.assertRaises(PySparkTypeError) as pe:
+ CF.slice(cdf.a, 1, 2.0)
+
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_COLUMN_OR_INTEGER_OR_STRING",
+ message_parameters={"arg_name": "length", "arg_type": "float"},
+ )
+
# test sort_array
self.assert_eq(
cdf.select(CF.sort_array(cdf.a, True), CF.sort_array("c", False)).toPandas(),
@@ -1759,6 +1786,15 @@ class SparkConnectFunctionTests(ReusedConnectTestCase, PandasOnSparkTestUtils, S
sdf.select(SF.from_json("a", s_schema, {"mode": "FAILFAST"})),
)
+ with self.assertRaises(PySparkTypeError) as pe:
+ CF.from_json("a", [c_schema])
+
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_COLUMN_OR_DATATYPE_OR_STRING",
+ message_parameters={"arg_name": "schema", "arg_type": "list"},
+ )
+
# test get_json_object
self.assert_eq(
cdf.select(
@@ -2155,6 +2191,24 @@ class SparkConnectFunctionTests(ReusedConnectTestCase, PandasOnSparkTestUtils, S
truncate=100,
)
+ with self.assertRaises(PySparkTypeError) as pe:
+ CF.window("date", "15 seconds", 10, "5 seconds")
+
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_A_STRING",
+ message_parameters={"arg_name": "slideDuration", "arg_type": "int"},
+ )
+
+ with self.assertRaises(PySparkTypeError) as pe:
+ CF.window("date", "15 seconds", "10 seconds", 5)
+
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_A_STRING",
+ message_parameters={"arg_name": "startTime", "arg_type": "int"},
+ )
+
# test session_window
self.compare_by_show(
cdf.select(CF.session_window("date", "15 seconds")),
diff --git a/python/pyspark/sql/tests/test_functions.py b/python/pyspark/sql/tests/test_functions.py
index f7f15d8e2c1..6af45124db8 100644
--- a/python/pyspark/sql/tests/test_functions.py
+++ b/python/pyspark/sql/tests/test_functions.py
@@ -1021,15 +1021,14 @@ class FunctionsTestsMixin:
with self.assertRaisesRegex((Py4JJavaError, SparkConnectException), "2000000"):
df.select(assert_true(df.id < 2, df.id * 1e6)).toDF("val").collect()
- with self.assertRaises((PySparkTypeError, TypeError)) as pe:
+ with self.assertRaises(TypeError) as pe:
df.select(assert_true(df.id < 2, 5))
- if isinstance(pe, PySparkTypeError):
- self.check_error(
- exception=pe.exception,
- error_class="NOT_COLUMN_OR_STRING",
- message_parameters={"arg_name": "errMsg", "arg_type": "int"},
- )
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_COLUMN_OR_STRING",
+ message_parameters={"arg_name": "errMsg", "arg_type": "int"},
+ )
def test_raise_error(self):
from pyspark.sql.functions import raise_error
@@ -1042,15 +1041,14 @@ class FunctionsTestsMixin:
with self.assertRaisesRegex((Py4JJavaError, SparkConnectException), "barfoo"):
df.select(raise_error("barfoo")).collect()
- with self.assertRaises((PySparkTypeError, TypeError)) as pe:
+ with self.assertRaises(PySparkTypeError) as pe:
df.select(raise_error(None))
- if isinstance(pe, PySparkTypeError):
- self.check_error(
- exception=pe.exception,
- error_class="NOT_COLUMN_OR_STRING",
- message_parameters={"arg_name": "errMsg", "arg_type": "NoneType"},
- )
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_COLUMN_OR_STRING",
+ message_parameters={"arg_name": "errMsg", "arg_type": "NoneType"},
+ )
def test_sum_distinct(self):
self.spark.range(10).select(
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org