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/05 06:55:30 UTC
(spark) branch branch-4.x updated: [SPARK-58575][PYTHON] Inline redundant error_class parameter in worker.py verify helpers
This is an automated email from the ASF dual-hosted git repository.
Yicong-Huang pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new cb8ada68010d [SPARK-58575][PYTHON] Inline redundant error_class parameter in worker.py verify helpers
cb8ada68010d is described below
commit cb8ada68010dddb9d39c570d50c14d8f817b4745
Author: Yicong Huang <17...@users.noreply.github.com>
AuthorDate: Tue Aug 4 23:55:13 2026 -0700
[SPARK-58575][PYTHON] Inline redundant error_class parameter in worker.py verify helpers
### What changes were proposed in this pull request?
This is a follow-up to SPARK-58529, which consolidated the row-count verification helpers in `python/pyspark/worker.py` but left two of them over-parameterized with an `error_class` argument that is always passed the same constant. This PR inlines the constant and drops the parameter:
- `verify_output_row_limit`: both call sites (`SQL_SCALAR_ARROW_ITER_UDF` and `SQL_SCALAR_PANDAS_ITER_UDF`) pass `error_class="OUTPUT_EXCEEDS_INPUT_ROWS"`. The constant is inlined into the `raise` and the parameter is removed from the signature and both call sites.
- `verify_iterator_exhausted`: both call sites pass `error_class="INPUT_NOT_FULLY_CONSUMED"`. Same treatment.
Both error classes are semantically bound to these two helpers, which together enforce the iterator-UDF contract that the number of output rows equals the number of input rows. No other code path can reuse them: for example, UDTFs have no such contract (one input row may produce many output rows) and use a different set of verifiers, so parameterizing the error class buys no reuse.
### Why are the changes needed?
Addresses a non-blocking review comment on SPARK-58529. Removing the always-constant parameter makes the helpers read more clearly: the helper name states the contract and the inlined error class is the fixed violation signal, with no indirection to trace through.
### Does this PR introduce _any_ user-facing change?
No. The raised error class and message are unchanged, so there is no behavior change, no `error-conditions.json` change, and no change to existing test assertions.
### How was this patch tested?
Existing test `ScalarPandasUDFTests.test_vectorized_udf_invalid_length` already exercises the `INPUT_NOT_FULLY_CONSUMED` path through `verify_iterator_exhausted`. This PR additionally adds a case to that test covering an iterator UDF that emits more output rows than input rows, which exercises the `OUTPUT_EXCEEDS_INPUT_ROWS` fail-fast path through `verify_output_row_limit` (previously untested).
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #57773 from Yicong-Huang/inline-redundant-verify-error-class.
Authored-by: Yicong Huang <17...@users.noreply.github.com>
Signed-off-by: Yicong Huang <17...@users.noreply.github.com>
(cherry picked from commit c7c8bcef218b1bf51c7ce0fcce2028181e5a897c)
Signed-off-by: Yicong Huang <17...@users.noreply.github.com>
---
.../sql/tests/pandas/test_pandas_udf_scalar.py | 12 ++++++++++++
python/pyspark/worker.py | 19 +++++--------------
2 files changed, 17 insertions(+), 14 deletions(-)
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 4fa4571c6998..326dd9d53726 100644
--- a/python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py
+++ b/python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py
@@ -704,6 +704,18 @@ class ScalarPandasUDFTestsMixin:
with self.assertRaisesRegex(Exception, "The input iterator must be fully consumed"):
df1.select(iter_udf_not_reading_all_input(col("id"))).collect()
+ @pandas_udf(LongType(), PandasUDFType.SCALAR_ITER)
+ def iter_udf_too_many_output_rows(it):
+ for batch in it:
+ yield pd.Series([1] * (len(batch) + 1))
+
+ with self.sql_conf({"spark.sql.execution.arrow.maxRecordsPerBatch": 3}):
+ df1 = self.spark.range(10).repartition(1)
+ with self.assertRaisesRegex(
+ Exception, "The number of output rows must not exceed the number of input rows"
+ ):
+ df1.select(iter_udf_too_many_output_rows(col("id"))).collect()
+
def test_vectorized_udf_chained(self):
df = self.spark.range(10)
scalar_f = pandas_udf(lambda x: x + 1, LongType())
diff --git a/python/pyspark/worker.py b/python/pyspark/worker.py
index 3a5ae680e56c..9ac8abbfa485 100644
--- a/python/pyspark/worker.py
+++ b/python/pyspark/worker.py
@@ -329,27 +329,26 @@ def verify_scalar_result(result: Any, num_rows: int) -> Any:
return result
-def verify_iterator_exhausted(iterator: Iterator, error_class: str) -> None:
+def verify_iterator_exhausted(iterator: Iterator) -> None:
"""Verify that an iterator has been fully consumed."""
try:
next(iterator)
except StopIteration:
pass
else:
- raise PySparkRuntimeError(errorClass=error_class, messageParameters={})
+ raise PySparkRuntimeError(errorClass="INPUT_NOT_FULLY_CONSUMED", messageParameters={})
def verify_output_row_limit(
iterator: Iterator,
max_rows: Union[int, Callable[[], int]],
- error_class: str,
) -> Iterator:
"""Yield elements while verifying total rows do not exceed a limit (fail-fast)."""
total_rows = 0
for element in iterator:
total_rows += len(element)
if total_rows > (max_rows() if callable(max_rows) else max_rows):
- raise PySparkRuntimeError(errorClass=error_class, messageParameters={})
+ raise PySparkRuntimeError(errorClass="OUTPUT_EXCEEDS_INPUT_ROWS", messageParameters={})
yield element
@@ -2010,7 +2009,6 @@ def read_udfs(pickleSer, udf_info_list, eval_type, runner_conf, eval_conf):
limited = verify_output_row_limit(
process_results(),
lambda: num_input_rows,
- error_class="OUTPUT_EXCEEDS_INPUT_ROWS",
)
# Apply row count match check (final)
@@ -2023,10 +2021,7 @@ def read_udfs(pickleSer, udf_info_list, eval_type, runner_conf, eval_conf):
yield from matched
# Verify iterator consumed
- verify_iterator_exhausted(
- args_iter,
- error_class="INPUT_NOT_FULLY_CONSUMED",
- )
+ verify_iterator_exhausted(args_iter)
# profiling is not supported for UDF
return func, None, ser, ser
@@ -3126,7 +3121,6 @@ def read_udfs(pickleSer, udf_info_list, eval_type, runner_conf, eval_conf):
limited = verify_output_row_limit(
process_results(),
lambda: num_input_rows,
- error_class="OUTPUT_EXCEEDS_INPUT_ROWS",
)
# Apply row count match check (final)
@@ -3139,10 +3133,7 @@ def read_udfs(pickleSer, udf_info_list, eval_type, runner_conf, eval_conf):
yield from matched
# Verify iterator consumed
- verify_iterator_exhausted(
- args_iter,
- error_class="INPUT_NOT_FULLY_CONSUMED",
- )
+ verify_iterator_exhausted(args_iter)
# profiling is not supported for UDF
return func, None, ser, ser
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org