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 3909ebe6f254 [SPARK-60530][PYTHON] Drop redundant run-level
fail_on_stopiteration in ArrowElementwiseUDFHandler
3909ebe6f254 is described below
commit 3909ebe6f25462ff103930f66b518553d535bf29
Author: Yicong Huang <[email protected]>
AuthorDate: Mon Oct 26 18:50:05 2026 +0000
[SPARK-60530][PYTHON] Drop redundant run-level fail_on_stopiteration in
ArrowElementwiseUDFHandler
### What changes were proposed in this pull request?
Remove the `fail_on_stopiteration` decorator on `_evaluate_elementwise_udf`
in `ArrowElementwiseUDFHandler.run` (`SQL_ARROW_ELEMENTWISE_UDF`), and the
now-unused import.
### Why are the changes needed?
The handler keeps the default `_fail_on_stopiteration = True`, so
`EvalTypeHandler.__init__` already wraps each UDF in `fail_on_stopiteration`. A
`StopIteration` from the UDF becomes `STOP_ITERATION_OCCURRED` at that inner
layer, on both the sequential path and inside the `ThreadPoolExecutor` path, so
the outer decorator never sees a `StopIteration`.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Extended `test_stop_iteration_raises_runtime_error` in
`test_elementwise_eval_type_handlers.py` to cover both the sequential and the
thread-pool (`arrow.concurrency.level > 0`) paths and assert the
`STOP_ITERATION_OCCURRED` condition.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
This pull request and its description were written by Isaac.
Closes #59775 from Yicong-Huang/elementwise-drop-double-stopiteration-wrap.
Authored-by: Yicong Huang <[email protected]>
Signed-off-by: Yicong-Huang <[email protected]>
(cherry picked from commit 6c9dcb9b354d378e39b6321af58d64e86f26441e)
Signed-off-by: Yicong-Huang <[email protected]>
---
python/pyspark/eval_handlers/_elementwise.py | 3 +--
.../tests/test_elementwise_eval_type_handlers.py | 12 +++++++++---
2 files changed, 10 insertions(+), 5 deletions(-)
diff --git a/python/pyspark/eval_handlers/_elementwise.py
b/python/pyspark/eval_handlers/_elementwise.py
index c5a1ce4e3dd3..66d5994039b4 100644
--- a/python/pyspark/eval_handlers/_elementwise.py
+++ b/python/pyspark/eval_handlers/_elementwise.py
@@ -57,7 +57,7 @@ from pyspark.sql.conversion import (
from pyspark.sql.pandas.types import to_arrow_type
from pyspark.sql.pandas.utils import require_minimum_pyarrow_version
from pyspark.sql.types import ArrayType, DataType, StructField, StructType
-from pyspark.util import PythonEvalType, fail_on_stopiteration
+from pyspark.util import PythonEvalType
if TYPE_CHECKING:
import pyarrow as pa
@@ -240,7 +240,6 @@ class
ArrowElementwiseUDFHandler(BatchEvalTypeHandler["pa.RecordBatch"]):
def run(self, split_index: int, data: Iterator[pa.RecordBatch]) ->
Iterator[pa.RecordBatch]:
import pyarrow as pa
- @fail_on_stopiteration
def _evaluate_elementwise_udf(udf_func: Any, rows: Iterator[tuple]) ->
list[Any]:
if self._runner_conf.arrow_concurrency_level <= 0:
return [udf_func(*row) for row in rows]
diff --git
a/python/pyspark/eval_handlers/tests/test_elementwise_eval_type_handlers.py
b/python/pyspark/eval_handlers/tests/test_elementwise_eval_type_handlers.py
index 9dbf346607bf..2b018efc46c7 100644
--- a/python/pyspark/eval_handlers/tests/test_elementwise_eval_type_handlers.py
+++ b/python/pyspark/eval_handlers/tests/test_elementwise_eval_type_handlers.py
@@ -216,9 +216,15 @@ class
ArrowElementwiseUDFHandlerTests(ElementwiseEvalTypeHandlerTestCase):
def stop(x):
raise StopIteration()
- handler = self._handler(ArrowElementwiseUDFHandler, (stop, [0], {},
LongType()))
- with self.assertRaises(PySparkRuntimeError):
- list(handler.run(0, iter([self._batch()])))
+ for level in ("0", "4"):
+ with self.subTest(concurrency_level=level):
+ conf =
{"spark.sql.execution.pythonUDF.arrow.concurrency.level": level}
+ handler = self._handler(
+ ArrowElementwiseUDFHandler, (stop, [0], {}, LongType()),
runner_conf=conf
+ )
+ with self.assertRaises(PySparkRuntimeError) as ctx:
+ list(handler.run(0, iter([self._batch()])))
+ self.assertEqual(ctx.exception.getCondition(),
"STOP_ITERATION_OCCURRED")
class ScalarElementwiseUDFHandlerTestsMixin:
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
