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 4b682c6005da [SPARK-57645][PYTHON][TEST] Add ASV microbenchmark for
SQL_GROUPED_AGG_PANDAS_ITER_UDF
4b682c6005da is described below
commit 4b682c6005da5a8b0224e5d081ce7d07775d1722
Author: Yicong Huang <[email protected]>
AuthorDate: Wed Jun 24 19:02:38 2026 +0000
[SPARK-57645][PYTHON][TEST] Add ASV microbenchmark for
SQL_GROUPED_AGG_PANDAS_ITER_UDF
### What changes were proposed in this pull request?
Add an ASV microbenchmark for `SQL_GROUPED_AGG_PANDAS_ITER_UDF` to
`python/benchmarks/bench_eval_type.py`, parallel to the existing
`GroupedAggArrowIterUDFTimeBench`. New classes:
`_GroupedAggPandasIterBenchMixin`, `GroupedAggPandasIterUDFTimeBench`, and
`GroupedAggPandasIterUDFPeakmemBench`. The mixin reuses
`_write_scenario`/`_build_scenario`/`_scenario_configs` from the non-iterator
Pandas sibling and only overrides the eval type and the iterator-style UDFs
(`sum_udf`, `mean_multi_ [...]
### Why are the changes needed?
`SQL_GROUPED_AGG_PANDAS_ITER_UDF` had no worker-level microbenchmark. This
fills the coverage gap and provides a before/after baseline for an upcoming
serializer refactor of this eval type.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Existing tests. Benchmark-only addition. The worker output of the new
iterator bench was verified to be byte-identical to the non-iterator Pandas
grouped-agg bench across all scenario/UDF combinations (only the trailing
timing telemetry differs).
ASV results (`COLUMNS=120 asv run --bench GroupedAggPandasIterUDFTimeBench
-a repeat=3 --python=same`):
```text
bench_eval_type.GroupedAggPandasIterUDFTimeBench.time_worker
================ ============ ================
-- udf
---------------- -----------------------------
scenario sum_udf mean_multi_udf
================ ============ ================
few_groups_sm 40.6+-0.3ms 46.9+-0.4ms
few_groups_lg 66.5+-0.4ms 74.7+-0.4ms
many_groups_sm 1.53+-0.01s 1.77+-0.01s
many_groups_lg 428+-3ms 487+-1ms
wide_cols 397+-0.9ms 420+-2ms
================ ============ ================
```
Numbers are stable across two local runs (deltas < 3%).
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #56730 from Yicong-Huang/SPARK-57645/bench/grouped-agg-pandas-iter.
Authored-by: Yicong Huang <[email protected]>
Signed-off-by: Yicong-Huang <[email protected]>
---
python/benchmarks/bench_eval_type.py | 45 ++++++++++++++++++++++++++++++++++++
1 file changed, 45 insertions(+)
diff --git a/python/benchmarks/bench_eval_type.py
b/python/benchmarks/bench_eval_type.py
index 9bf005880378..f220b804be0b 100644
--- a/python/benchmarks/bench_eval_type.py
+++ b/python/benchmarks/bench_eval_type.py
@@ -1041,6 +1041,51 @@ class
GroupedAggPandasUDFPeakmemBench(_GroupedAggPandasBenchMixin, _PeakmemBench
pass
+# -- SQL_GROUPED_AGG_PANDAS_ITER_UDF
-------------------------------------------
+# UDF receives an iterator of ``pd.Series`` columns (or tuples of them) per
+# group, returns scalar.
+
+
+class _GroupedAggPandasIterBenchMixin(_GroupedAggPandasBenchMixin):
+ """Provides _write_scenario for SQL_GROUPED_AGG_PANDAS_ITER_UDF.
+
+ Inherits ``_build_scenario`` and ``_write_scenario`` from the Pandas
+ sibling; only the eval type and the UDFs differ. The UDF consumes the
+ per-group batches lazily through an iterator instead of receiving a single
+ concatenated column.
+ """
+
+ def _grouped_agg_pandas_iter_sum(series_iter):
+ """Sum across batches via iterator."""
+ total = 0
+ for col in series_iter:
+ total += col.sum() or 0
+ return total
+
+ def _grouped_agg_pandas_iter_mean_multi(series_iter):
+ """Mean across batches of tuples via iterator."""
+ total = 0.0
+ for col0, col1 in series_iter:
+ total += (col0.mean() or 0) + (col1.mean() or 0)
+ return total
+
+ _eval_type = PythonEvalType.SQL_GROUPED_AGG_PANDAS_ITER_UDF
+ _udfs = {
+ "sum_udf": _grouped_agg_pandas_iter_sum,
+ "mean_multi_udf": _grouped_agg_pandas_iter_mean_multi,
+ }
+ params = [list(_GroupedAggArrowBenchMixin._scenario_configs), list(_udfs)]
+ param_names = ["scenario", "udf"]
+
+
+class GroupedAggPandasIterUDFTimeBench(_GroupedAggPandasIterBenchMixin,
_TimeBenchBase):
+ pass
+
+
+class GroupedAggPandasIterUDFPeakmemBench(_GroupedAggPandasIterBenchMixin,
_PeakmemBenchBase):
+ pass
+
+
# -- SQL_GROUPED_MAP_ARROW_UDF ------------------------------------------------
# UDF receives ``pa.Table``, returns ``pa.Table``.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
