viirya opened a new pull request, #57137: URL: https://github.com/apache/spark/pull/57137
### What changes were proposed in this pull request? Follow-up of SPARK-58019 (#57099), SPARK-58023 (#57104) and SPARK-58024 (#57105); **only the last commit is new — this PR is stacked on #57105 and will be rebased as its predecessors merge.** Even with the bulk `_to_pylist` conversions from the predecessor PRs, the Arrow Python UDF worker still runs **per-row Python converters** on both sides of the UDF. This PR fuses them into bulk operations: - **Input**: new `ArrowTableToRowsConversion._to_rows_column` is equivalent to `[conv(v) for v in _to_pylist(column)]` but builds map rows as dicts and struct rows as `Row`s directly from the flattened child columns, applying child converters per flattened column instead of per row. Anything not covered falls back to the per-row converter unchanged. - **Output**: when the element/field/value converters are identity, UDF results are assembled in bulk — arrays are passed to `pa.array` directly (skipping the per-row defensive copy), map results are passed as dicts directly (PyArrow accepts dicts for map types), and struct results (tuples/`Row`s) are transposed and assembled via `pa.StructArray.from_arrays` with a null mask. Any shape/validation mismatch falls back to the existing per-row path, preserving error behavior. ### Why are the changes needed? Worker-side profiling shows the per-row converters dominate the remaining performance gap between Arrow Python UDFs and pickled Python UDFs on nested types. End-to-end (`bench_python_udf`-style workload, 6.4M rows, macOS arm64, `local[4]`, on top of the three predecessor PRs): | case | pickled UDF | Arrow UDF before | Arrow UDF after | delta | |---|---|---|---|---| | `array<string>` | 1.10 s | 2.39 s | 1.77 s | 1.35x | | `map<string,int>` | 1.18 s | 3.65 s | 2.40 s | 1.53x | | `struct<i:int,s:string,d:double>` | 4.32 s | 6.03 s | 4.87 s | 1.24x | Microbenchmarks of the fused chains (400k rows, byte-identical outputs): input map→dict 3.0x, input struct→Row 1.8x, output array 7.5x, output map 4.2x, output struct 4.0x. The remaining input-side cost is PyArrow's per-element `to_pylist`, addressed upstream in apache/arrow#50326 / apache/arrow#50327. ### Does this PR introduce _any_ user-facing change? No. Only performance; conversion results are identical (covered by exact-type tests and an arrow-vs-pickle end-to-end comparison with injected nulls). ### How was this patch tested? New `ArrowToRowsColumnTests` compares `_to_rows_column` against the per-row converter chain with exact-type assertions across map (incl. nested values), struct (incl. nested map), array-of-struct/map, fallback types (timestamp/binary), sliced and chunked views, plus dedicated tests that struct rows are `Row` objects and map rows preserve Arrow entry order. Full `test_conversion.py` passes (43 tests, 200 subtests). End-to-end: arrow-vs-pickle `collect()` results verified identical for all nested benchmark cases with 10% injected nulls. ### Was this patch authored or co-authored using generative AI tooling? Yes. This pull request and its description were written by Isaac (Claude Code). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
