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 b1547b4b2cab [SPARK-58019][PYTHON][FOLLOWUP] Convert Arrow list
columns to Python rows in bulk at foreachPartition and data source write
b1547b4b2cab is described below
commit b1547b4b2cabad27fc57367eefd18013ac811287
Author: Yicong Huang <[email protected]>
AuthorDate: Fri Jul 17 04:16:51 2026 +0000
[SPARK-58019][PYTHON][FOLLOWUP] Convert Arrow list columns to Python rows
in bulk at foreachPartition and data source write
### What changes were proposed in this pull request?
Follow-up to #253 (SPARK-58019). That PR added
`ArrowTableToRowsConversion._to_pylist`, a drop-in, byte-identical replacement
for `Array.to_pylist()` that converts list/struct/map columns in bulk (flatten
the child values once, slice per row via the offset buffer and validity bitmap)
instead of materializing one Arrow Scalar per element -- measured 1.5x-3.5x
faster on array columns (apache/arrow#50326) -- and transparently falls back to
`to_pylist()` when the installed PyArrow is >= 2 [...]
Two analogous Arrow-to-rows call sites were left out of that change and
still used plain `to_pylist()`. This follow-up routes both through
`_to_pylist`: the Spark Connect `DataFrame.foreachPartition` path, and the
row-based Python data source write path (`batch_to_rows` in
`write_into_data_source.py`).
`plan_data_source_read.py` is intentionally left unchanged: it converts a
single `BinaryType` cell (`num_columns == 1, num_rows == 1`, a pickled
`InputPartition`), where the bulk path has zero benefit and binary falls
through to `to_pylist()` regardless.
### Why are the changes needed?
Both changed sites carry arbitrary user-defined schemas that realistically
include array/struct/map columns, so they now get the same speedup as the main
PR when eligible, and are otherwise identical to before (the conversion falls
back to `to_pylist()` for scalar-only columns or newer PyArrow, so it can never
regress correctness or performance).
### Does this PR introduce any user-facing change?
No. `_to_pylist` is byte-identical to `to_pylist()`.
### How was this patch tested?
No behavior change; the byte-identical contract of `_to_pylist` is covered
by the existing `test_matches_to_pylist` in
`python/pyspark/sql/tests/test_conversion.py`. Both call sites already have
integration coverage (`foreachPartition` and `DataSourceWriter` tests).
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #57316 from Yicong-Huang/SPARK-58019-followup.
Authored-by: Yicong Huang <[email protected]>
Signed-off-by: Yicong-Huang <[email protected]>
---
python/pyspark/sql/connect/dataframe.py | 4 +++-
python/pyspark/sql/worker/write_into_data_source.py | 4 +++-
2 files changed, 6 insertions(+), 2 deletions(-)
diff --git a/python/pyspark/sql/connect/dataframe.py
b/python/pyspark/sql/connect/dataframe.py
index d0ef3061e58a..64277d1b35b6 100644
--- a/python/pyspark/sql/connect/dataframe.py
+++ b/python/pyspark/sql/connect/dataframe.py
@@ -2334,7 +2334,9 @@ class DataFrame(ParentDataFrame):
def foreach_partition_func(itr: Iterable[pa.RecordBatch]) ->
Iterable[pa.RecordBatch]:
def flatten() -> Iterator[Row]:
for table in itr:
- columnar_data = [column.to_pylist() for column in
table.columns]
+ columnar_data = [
+ ArrowTableToRowsConversion._to_pylist(column) for
column in table.columns
+ ]
for i in range(0, table.num_rows):
values = [
field_converters[j](columnar_data[j][i]) # type:
ignore[misc]
diff --git a/python/pyspark/sql/worker/write_into_data_source.py
b/python/pyspark/sql/worker/write_into_data_source.py
index 1c39eae6f212..ca97090f004f 100644
--- a/python/pyspark/sql/worker/write_into_data_source.py
+++ b/python/pyspark/sql/worker/write_into_data_source.py
@@ -187,7 +187,9 @@ def _main(infile: IO, outfile: IO) -> None:
def data_source_write_func(iterator: Iterator[pa.RecordBatch]) ->
Iterator[pa.RecordBatch]:
def batch_to_rows() -> Iterator[Row]:
for batch in iterator:
- columns = [column.to_pylist() for column in batch.columns]
+ columns = [
+ ArrowTableToRowsConversion._to_pylist(column) for column
in batch.columns
+ ]
for row in range(0, batch.num_rows):
values = [
converters[col](columns[col][row]) for col in
range(batch.num_columns)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
