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 c7c8bcef218b [SPARK-58575][PYTHON] Inline redundant error_class
parameter in worker.py verify helpers
c7c8bcef218b is described below
commit c7c8bcef218b1bf51c7ce0fcce2028181e5a897c
Author: Yicong Huang <[email protected]>
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 <[email protected]>
Signed-off-by: Yicong Huang <[email protected]>
---
.../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: [email protected]
For additional commands, e-mail: [email protected]
