You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by yi...@apache.org on 2026/08/03 20:32:08 UTC
(spark) branch master updated: [SPARK-58161][PYTHON] Standardize PySpark UDF row count mismatch error to use RESULT_ROWS_MISMATCH
This is an automated email from the ASF dual-hosted git repository.
Yicong-Huang 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 fd9412dd236f [SPARK-58161][PYTHON] Standardize PySpark UDF row count mismatch error to use RESULT_ROWS_MISMATCH
fd9412dd236f is described below
commit fd9412dd236f673e73de5193c7592e9fcc0dd41d
Author: gaurav kumar pandey <ga...@gmail.com>
AuthorDate: Mon Aug 3 20:31:54 2026 +0000
[SPARK-58161][PYTHON] Standardize PySpark UDF row count mismatch error to use RESULT_ROWS_MISMATCH
### What changes were proposed in this pull request?
This PR standardizes the PySpark Pandas and Arrow UDF row-count mismatch errors to use the consolidated error class `RESULT_ROWS_MISMATCH` instead of `SCHEMA_MISMATCH_FOR_PANDAS_UDF`.
Specifically, the changes:
1. Migrate verification checks in `worker.py` to use `RESULT_ROWS_MISMATCH` with input/output length parameters.
2. Update `error-conditions.json`, development debugging documentation, and associated unit tests.
### Why are the changes needed?
To consolidate and standardize PySpark's internal error class definitions under a unified error handling system.
### Does this PR introduce _any_ user-facing change?
Yes. The error class for PySpark UDF row count mismatch errors has been updated to `RESULT_ROWS_MISMATCH`. Exceptions raised when a UDF returns a row count that does not match the input length will now carry the `RESULT_ROWS_MISMATCH` error class.
### How was this patch tested?
Updated and executed unit tests in `python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py`.
Closes #57292 from Alwaysgaurav1/pyspark-udf-error-standardization-master.
Authored-by: gaurav kumar pandey <ga...@gmail.com>
Signed-off-by: Yicong-Huang <17...@users.noreply.github.com>
---
python/docs/source/development/debugging.rst | 4 +-
python/pyspark/errors/error-conditions.json | 6 +--
.../sql/tests/pandas/test_pandas_udf_scalar.py | 2 +-
python/pyspark/worker.py | 46 ++++++++++++++--------
4 files changed, 34 insertions(+), 24 deletions(-)
diff --git a/python/docs/source/development/debugging.rst b/python/docs/source/development/debugging.rst
index 4b7e2e288301..e69055b410a2 100644
--- a/python/docs/source/development/debugging.rst
+++ b/python/docs/source/development/debugging.rst
@@ -575,7 +575,7 @@ Solution:
1 6
dtype: int64
-**RuntimeError: Result vector from pandas_udf was not the required length**
+**PySparkRuntimeError: [RESULT_ROWS_MISMATCH] The number of output rows must match the number of input rows**
Exception:
@@ -588,7 +588,7 @@ Exception:
22/04/12 13:46:39 ERROR Executor: Exception in task 2.0 in stage 16.0 (TID 88)
org.apache.spark.api.python.PythonException: Traceback (most recent call last):
...
- RuntimeError: Result vector from pandas_udf was not the required length: expected 1, got 0
+ pyspark.errors.exceptions.base.PySparkRuntimeError: [RESULT_ROWS_MISMATCH] The number of output rows (0) must match the number of input rows (1).
Solution:
diff --git a/python/pyspark/errors/error-conditions.json b/python/pyspark/errors/error-conditions.json
index 580f5a8f16e9..07ce4314763a 100644
--- a/python/pyspark/errors/error-conditions.json
+++ b/python/pyspark/errors/error-conditions.json
@@ -861,11 +861,7 @@
"An Observation can be used with a DataFrame only once."
]
},
- "SCHEMA_MISMATCH_FOR_PANDAS_UDF": {
- "message": [
- "Result vector from <udf_type> was not the required length: expected <expected>, got <actual>."
- ]
- },
+
"SESSION_ALREADY_EXIST": {
"message": [
"Cannot start a remote Spark session because there is a regular Spark session already running."
diff --git a/python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py b/python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py
index 72d9fa566dee..4fa4571c6998 100644
--- a/python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py
+++ b/python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py
@@ -678,7 +678,7 @@ class ScalarPandasUDFTestsMixin:
df = self.spark.range(10)
raise_exception = pandas_udf(lambda _: pd.Series(1), LongType())
with self.assertRaisesRegex(
- Exception, "Result vector from pandas_udf was not the required length"
+ Exception, "The number of output rows.*must match the number of input rows"
):
df.select(raise_exception(col("id"))).collect()
diff --git a/python/pyspark/worker.py b/python/pyspark/worker.py
index 2db77d73148e..0db7e7151038 100644
--- a/python/pyspark/worker.py
+++ b/python/pyspark/worker.py
@@ -301,6 +301,8 @@ def verify_result_row_count(result_length: int, expected: int) -> None:
"output_length": str(result_length),
"input_length": str(expected),
},
+ message=f"The number of output rows ({result_length}) must match the number of input rows ({expected}). "
+ f"Result vector from pandas_udf was not the required length: expected {expected}, got {result_length}.",
)
@@ -326,14 +328,14 @@ def verify_scalar_result(result: Any, num_rows: int) -> Any:
},
)
if result_length != num_rows:
- # TODO: change error class to RESULT_ROWS_MISMATCH
raise PySparkRuntimeError(
- errorClass="SCHEMA_MISMATCH_FOR_PANDAS_UDF",
+ errorClass="RESULT_ROWS_MISMATCH",
messageParameters={
- "udf_type": "arrow_udf",
- "expected": str(num_rows),
- "actual": str(result_length),
+ "output_length": str(result_length),
+ "input_length": str(num_rows),
},
+ message=f"The number of output rows ({result_length}) must match the number of input rows ({num_rows}). "
+ f"Result vector from pandas_udf was not the required length: expected {num_rows}, got {result_length}.",
)
return result
@@ -375,13 +377,24 @@ def verify_output_row_count(
expected = expected_rows() if callable(expected_rows) else expected_rows
if actual_rows != expected:
- raise PySparkRuntimeError(
- errorClass=error_class,
- messageParameters={
- "output_length": str(actual_rows),
- "input_length": str(expected),
- },
- )
+ if error_class == "RESULT_ROWS_MISMATCH":
+ raise PySparkRuntimeError(
+ errorClass=error_class,
+ messageParameters={
+ "output_length": str(actual_rows),
+ "input_length": str(expected),
+ },
+ message=f"The number of output rows ({actual_rows}) must match the number of input rows ({expected}). "
+ f"Result vector from pandas_udf was not the required length: expected {expected}, got {actual_rows}.",
+ )
+ else:
+ raise PySparkRuntimeError(
+ errorClass=error_class,
+ messageParameters={
+ "output_length": str(actual_rows),
+ "input_length": str(expected),
+ },
+ )
def wrap_udf(f, args_offsets, kwargs_offsets, return_type):
@@ -3059,12 +3072,13 @@ def read_udfs(pickleSer, udf_info_list, eval_type, runner_conf, eval_conf):
)
if len(result) != num_rows:
raise PySparkRuntimeError(
- errorClass="SCHEMA_MISMATCH_FOR_PANDAS_UDF",
+ errorClass="RESULT_ROWS_MISMATCH",
messageParameters={
- "udf_type": "pandas_udf",
- "expected": str(num_rows),
- "actual": str(len(result)),
+ "output_length": str(len(result)),
+ "input_length": str(num_rows),
},
+ message=f"The number of output rows ({len(result)}) must match the number of input rows ({num_rows}). "
+ f"Result vector from pandas_udf was not the required length: expected {num_rows}, got {len(result)}.",
)
# struct_in_pandas="dict": UDF must return DataFrame for struct types
if isinstance(udf_return_type, StructType) and not isinstance(
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org