[SPARK-58191][PYTHON] Refactor SQL_TRANSFORM_WITH_STATE_PYTHON_ROW_UDF and SQL_TRANSFORM_WITH_STATE_PYTHON_ROW_INIT_STATE_UDF

[Yicong-Huang]

What changes were proposed in this pull request?

This PR refactors SQL_TRANSFORM_WITH_STATE_PYTHON_ROW_UDF and SQL_TRANSFORM_WITH_STATE_PYTHON_ROW_INIT_STATE_UDF so that the worker uses the plain ArrowStreamSerializer for pure Arrow stream I/O, moving the per-eval-type logic from TransformWithStateInPySparkRowSerializer / TransformWithStateInPySparkRowInitStateSerializer into read_udfs() in worker.py:

• Row variant: reading each input batch row by row into Row objects, regrouping by grouping key, invoking the UDF once per key with a lazy iterator of Rows, and converting the yielded (iterator of Row, Spark type) result back into a struct-wrapped Arrow RecordBatch.
• Row init-state variant: flattening the inputData / initState struct columns, regrouping rows by grouping key, re-chunking into Row lists bounded by arrow_max_records_per_batch, splitting each chunk into a data iterator and an init-state iterator, and converting results back to Arrow.

This mirrors the earlier refactor of the Pandas variants (SPARK-57400, SPARK-58128). Both variants are done together in one PR because they share a serializer class hierarchy (...RowInitStateSerializer extends ...RowSerializer) and are deleted together in the follow-up cleanup. The two serializer classes are kept for now and deleted in that cleanup.

Why are the changes needed?

Part of SPARK-55388. Keeping serializers as pure Arrow stream I/O and concentrating eval-type-specific logic in worker.py makes the per-eval-type data flow explicit.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Existing tests. No behavior change.

ASV comparison (bench_eval_type.TransformWithStateRowUDFTimeBench / TransformWithStateRowInitStateUDFTimeBench, -a repeat=3, each combo run in a fresh Python process to match production): before = upstream/master, after = this PR. Values are from one representative run per side, keeping ASV's native +- intervals; the conclusion is consistent across runs. All deltas are within run-to-run noise, showing no regression. (A handful of combos flagged larger deltas only when the before/after runs were executed concurrently; re-running those in isolation brought every one back within noise, confirming they were benchmark-ordering / scheduling artifacts that do not apply in production where each Spark task runs in a fresh Python worker.)

time_worker -- SQL_TRANSFORM_WITH_STATE_PYTHON_ROW_UDF

scenario        udf           before       after        diff 
--------------  ------------  -----------  -----------  -----
few_groups_lg   count_udf     3.97+-0.05s  4.01+-0.02s  +1.0%
few_groups_lg   identity_udf  4.95+-0.02s  4.83+-0.05s  -2.4%
few_groups_lg   rebuild_udf   5.65+-0.07s  5.63+-0.03s  -0.4%
few_groups_sm   count_udf     399+-3ms     421+-5ms     +5.6%
few_groups_sm   identity_udf  528+-17ms    503+-2ms     -4.7%
few_groups_sm   rebuild_udf   606+-30ms    600+-14ms    -1.1%
many_groups_lg  count_udf     1.65+-0.02s  1.65+-0.03s  +0.4%
many_groups_lg  identity_udf  2.18+-0.03s  2.10+-0.05s  -3.6%
many_groups_lg  rebuild_udf   2.45+-0.01s  2.45+-0.04s  -0.1%
many_groups_sm  count_udf     1.80+-0.02s  1.87+-0.02s  +4.0%
many_groups_sm  identity_udf  2.32+-0.03s  2.42+-0.05s  +4.3%
many_groups_sm  rebuild_udf   2.81+-0.05s  2.76+-0.02s  -1.6%
mixed_cols      count_udf     1.69+-0.03s  1.71+-0.04s  +1.1%
mixed_cols      identity_udf  2.11+-0.04s  2.05+-0.07s  -2.8%
mixed_cols      rebuild_udf   2.42+-0.04s  2.41+-0.05s  -0.5%
nested_struct   count_udf     2.51+-0.03s  2.58+-0.01s  +2.8%
nested_struct   identity_udf  3.34+-0.02s  3.31+-0.06s  -0.8%
nested_struct   rebuild_udf   3.74+-0.05s  3.74+-0.03s  -0.1%
wide_cols       count_udf     4.54+-0.11s  4.55+-0.04s  +0.2%
wide_cols       identity_udf  5.39+-0.03s  5.52+-0.16s  +2.3%
wide_cols       rebuild_udf   6.27+-0.13s  6.23+-0.04s  -0.7%

time_worker -- SQL_TRANSFORM_WITH_STATE_PYTHON_ROW_INIT_STATE_UDF

scenario        udf           before       after        diff 
--------------  ------------  -----------  -----------  -----
few_groups_lg   count_udf     3.65+-0.05s  3.75+-0.08s  +2.7%
few_groups_lg   identity_udf  4.65+-0.03s  4.70+-0.06s  +1.2%
few_groups_lg   rebuild_udf   5.25+-0.10s  5.21+-0.06s  -0.8%
few_groups_sm   count_udf     425+-9ms     407+-14ms    -4.1%
few_groups_sm   identity_udf  500+-13ms    529+-16ms    +5.8%
few_groups_sm   rebuild_udf   593+-44ms    607+-30ms    +2.4%
many_groups_lg  count_udf     2.10+-0.10s  2.17+-0.06s  +3.2%
many_groups_lg  identity_udf  2.60+-0.05s  2.62+-0.07s  +0.6%
many_groups_lg  rebuild_udf   2.83+-0.06s  2.82+-0.02s  -0.3%
many_groups_sm  count_udf     4.54+-0.06s  4.62+-0.08s  +1.6%
many_groups_sm  identity_udf  5.05+-0.10s  5.03+-0.13s  -0.4%
many_groups_sm  rebuild_udf   5.44+-0.13s  5.43+-0.04s  -0.2%
mixed_cols      count_udf     1.74+-0.05s  1.76+-0.04s  +1.4%
mixed_cols      identity_udf  2.16+-0.03s  2.16+-0.03s  -0.1%
mixed_cols      rebuild_udf   2.40+-0.05s  2.35+-0.02s  -1.8%
nested_struct   count_udf     2.79+-0.01s  2.76+-0.02s  -1.3%
nested_struct   identity_udf  3.63+-0.03s  3.60+-0.08s  -0.8%
nested_struct   rebuild_udf   3.94+-0.09s  3.88+-0.09s  -1.5%
wide_cols       count_udf     4.38+-0.05s  4.49+-0.03s  +2.5%
wide_cols       identity_udf  5.63+-0.10s  5.51+-0.09s  -2.2%
wide_cols       rebuild_udf   5.97+-0.00s  5.83+-0.12s  -2.4%