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 523ab88c1791 [SPARK-59942][PYTHON] Use real RunnerConf in Arrow eval
type handler tests
523ab88c1791 is described below
commit 523ab88c17914e59ceed7094b2b2ae004c8a9178
Author: Yicong Huang <[email protected]>
AuthorDate: Fri Oct 2 05:26:32 2026 +0000
[SPARK-59942][PYTHON] Use real RunnerConf in Arrow eval type handler tests
### What changes were proposed in this pull request?
The Arrow eval type handler tests built handlers with a hand-written
`_RunnerConf` stand-in that exposed only a few hard-coded fields. This replaces
it with the worker's real `RunnerConf`, constructed from a plain conf dict
(`RunnerConf({...})`), matching the pattern already used by the incremental
aggregate/window handler tests added in SPARK-59727. Window-frame tests now
pass the bound types through the real conf key
(`RunnerConf({"window_bound_types": "unbounded"})` / `"bounded"`) [...]
### Why are the changes needed?
The fake config duplicated field defaults that can silently drift from the
real `RunnerConf`, and it hid the actual conf keys the handlers read. Building
the real config from a conf dict exercises the same code path as production and
keeps the tests honest. This is a test improvement following the Arrow eval
handler migrations under SPARK-59415.
### Does this PR introduce _any_ user-facing change?
No. Test-only change.
### How was this patch tested?
`pyspark.eval_handlers.tests.test_arrow_eval_type_handlers` and
`pyspark.eval_handlers.tests.test_base_eval_type_handlers` pass (38 tests).
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 4.8
This pull request and its description were written by Isaac.
Closes #59195 from Yicong-Huang/arrow-eval-conf-test.
Authored-by: Yicong Huang <[email protected]>
Signed-off-by: Yicong-Huang <[email protected]>
---
.../tests/test_arrow_eval_type_handlers.py | 35 ++++++++--------------
.../tests/test_base_eval_type_handlers.py | 22 ++++++--------
2 files changed, 22 insertions(+), 35 deletions(-)
diff --git
a/python/pyspark/eval_handlers/tests/test_arrow_eval_type_handlers.py
b/python/pyspark/eval_handlers/tests/test_arrow_eval_type_handlers.py
index 5c47ae609f41..e8a6cc24e5f8 100644
--- a/python/pyspark/eval_handlers/tests/test_arrow_eval_type_handlers.py
+++ b/python/pyspark/eval_handlers/tests/test_arrow_eval_type_handlers.py
@@ -57,17 +57,6 @@ if have_pyarrow:
from pyspark.sql.conversion import ArrowBatchTransformer
-class _RunnerConf:
- """Minimal stand-in for the worker's RunnerConf, exposing only the fields
- the handlers under test read."""
-
- use_large_var_types = False
- assign_cols_by_name = True
- map_in_batch_legacy_accept_any_iterable = False
- # One bound type per window UDF; the bounded test overrides this per
instance.
- window_bound_types = ["unbounded"]
-
-
class _SumAggregator:
"""Incremental ``Aggregator`` stub: sums non-null inputs, with a ``(sum,
count)`` buffer.
@@ -136,7 +125,9 @@ def _scalar_handler(handler_cls, udf):
The scalar UDF tuple is ``(func, args_offsets, kwargs_offsets,
return_type)``.
"""
- return handler_cls(udfs=[(udf, [0], {}, LongType())],
runner_conf=_RunnerConf(), eval_conf=None)
+ return handler_cls(
+ udfs=[(udf, [0], {}, LongType())], runner_conf=RunnerConf({}),
eval_conf=None
+ )
def _grouped_handler(handler_cls, udf, arg_offsets, num_udf_args):
@@ -146,7 +137,7 @@ def _grouped_handler(handler_cls, udf, arg_offsets,
num_udf_args):
"""
return handler_cls(
udfs=[(udf, arg_offsets, _RETURN_TYPE, num_udf_args)],
- runner_conf=_RunnerConf(),
+ runner_conf=RunnerConf({}),
eval_conf=None,
)
@@ -240,7 +231,7 @@ class ArrowMapUDFHandlerTests(unittest.TestCase):
yield _batch(v=[c.as_py() * 2 for c in batch.column("v")])
handler = ArrowMapUDFHandler(
- udfs=[(double_v, None, None, None)], runner_conf=_RunnerConf(),
eval_conf=None
+ udfs=[(double_v, None, None, None)], runner_conf=RunnerConf({}),
eval_conf=None
)
out = list(handler.run(0, iter([_struct_batch(v=[1, 2, 3])])))
self.assertEqual(out[0].column(0).field("v").to_pylist(), [2, 4, 6])
@@ -320,7 +311,7 @@ class ArrowGroupedAggUDFHandlerTests(unittest.TestCase):
return sum(c.as_py() for c in col)
handler = ArrowGroupedAggUDFHandler(
- udfs=[(sum_udf, [0], {}, LongType())], runner_conf=_RunnerConf(),
eval_conf=None
+ udfs=[(sum_udf, [0], {}, LongType())], runner_conf=RunnerConf({}),
eval_conf=None
)
out = list(handler.run(0, _one_group(_batch(v=[1, 2, 3]),
_batch(v=[4]))))
self.assertEqual(out[0].column("_0").to_pylist(), [10])
@@ -334,7 +325,7 @@ class ArrowGroupedAggIterUDFHandlerTests(unittest.TestCase):
return sum(c.as_py() for col in col_iter for c in col)
handler = ArrowGroupedAggIterUDFHandler(
- udfs=[(sum_iter_udf, [0], {}, LongType())],
runner_conf=_RunnerConf(), eval_conf=None
+ udfs=[(sum_iter_udf, [0], {}, LongType())],
runner_conf=RunnerConf({}), eval_conf=None
)
out = list(handler.run(0, _one_group(_batch(v=[10]), _batch(v=[20]))))
self.assertEqual(out[0].column("_0").to_pylist(), [30])
@@ -347,10 +338,10 @@ class ArrowWindowAggUDFHandlerTests(unittest.TestCase):
def sum_udf(col):
return sum(c.as_py() for c in col)
- conf = _RunnerConf()
- conf.window_bound_types = ["unbounded"]
handler = ArrowWindowAggUDFHandler(
- udfs=[(sum_udf, [0], {}, LongType())], runner_conf=conf,
eval_conf=None
+ udfs=[(sum_udf, [0], {}, LongType())],
+ runner_conf=RunnerConf({"window_bound_types": "unbounded"}),
+ eval_conf=None,
)
out = list(handler.run(0, _one_group(_batch(v=[1, 2, 3]))))
self.assertEqual(out[0].column("_0").to_pylist(), [6, 6, 6])
@@ -360,10 +351,10 @@ class ArrowWindowAggUDFHandlerTests(unittest.TestCase):
def sum_udf(col):
return sum(c.as_py() for c in col)
- conf = _RunnerConf()
- conf.window_bound_types = ["bounded"]
handler = ArrowWindowAggUDFHandler(
- udfs=[(sum_udf, [0, 1, 2], {}, LongType())], runner_conf=conf,
eval_conf=None
+ udfs=[(sum_udf, [0, 1, 2], {}, LongType())],
+ runner_conf=RunnerConf({"window_bound_types": "bounded"}),
+ eval_conf=None,
)
# Row 0 frame [0, 1) -> [10]; row 1 frame [0, 2) -> [10, 20].
out = list(handler.run(0, _one_group(_batch(begin=[0, 0], end=[1, 2],
v=[10, 20]))))
diff --git a/python/pyspark/eval_handlers/tests/test_base_eval_type_handlers.py
b/python/pyspark/eval_handlers/tests/test_base_eval_type_handlers.py
index c5fff82312ac..5fe45119bc88 100644
--- a/python/pyspark/eval_handlers/tests/test_base_eval_type_handlers.py
+++ b/python/pyspark/eval_handlers/tests/test_base_eval_type_handlers.py
@@ -21,7 +21,9 @@ Handler-flavor tests live alongside their module, e.g. the
Arrow handlers in
``test_arrow_eval_type_handlers``.
"""
+import os
import unittest
+from unittest.mock import patch
from pyspark.eval_handlers._base import (
BatchEvalTypeHandler,
@@ -37,14 +39,8 @@ from pyspark.sql.pandas.serializers import (
ArrowStreamSerializer,
)
-
-class _RunnerConf:
- """Minimal stand-in for the worker's RunnerConf, exposing only the fields
- the handlers under test read."""
-
- use_large_var_types = False
- assign_cols_by_name = True
- map_in_batch_legacy_accept_any_iterable = False
+with patch.dict(os.environ, {"SPARK_PYTHON_RUNTIME": "PYTHON_WORKER"}):
+ from pyspark.worker_util import RunnerConf
class EvalTypeHandlerTests(unittest.TestCase):
@@ -58,7 +54,7 @@ class EvalTypeHandlerTests(unittest.TestCase):
CoGroupedEvalTypeHandler,
):
with self.assertRaises(TypeError):
- base([], _RunnerConf(), None)
+ base([], RunnerConf({}), None)
def test_category_bases_are_not_registered(self):
# Only concrete subclasses that declare an eval type are registered.
@@ -84,12 +80,12 @@ class EvalTypeHandlerTests(unittest.TestCase):
def run(self, split_index, data):
return data
- self.assertIsInstance(_Batch([], _RunnerConf(), None).serializer,
ArrowStreamSerializer)
+ self.assertIsInstance(_Batch([], RunnerConf({}), None).serializer,
ArrowStreamSerializer)
self.assertIsInstance(
- _Grouped([], _RunnerConf(), None).serializer,
ArrowStreamGroupSerializer
+ _Grouped([], RunnerConf({}), None).serializer,
ArrowStreamGroupSerializer
)
self.assertIsInstance(
- _CoGrouped([], _RunnerConf(), None).serializer,
ArrowStreamCoGroupSerializer
+ _CoGrouped([], RunnerConf({}), None).serializer,
ArrowStreamCoGroupSerializer
)
def test_run_produces_output(self):
@@ -98,7 +94,7 @@ class EvalTypeHandlerTests(unittest.TestCase):
for item in data:
yield item * 2
- handler = _Doubler([], _RunnerConf(), None)
+ handler = _Doubler([], RunnerConf({}), None)
self.assertEqual(list(handler.run(0, iter([1, 2, 3]))), [2, 4, 6])
def test_duplicate_eval_type_rejected(self):
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
