Uh oh!
There was an error while loading. Please reload this page.
[SPARK-58019][PYTHON][FOLLOWUP] Convert Arrow list columns to Python rows in bulk at foreachPartition and data source write - #57316
Closed
Yicong-Huang wants to merge 1 commit into
Closed
Conversation
…tition and data source write
gaogaotiantian
approved these changes
Jul 16, 2026
Yicong-Huang added a commit
that referenced
this pull request
Jul 17, 2026
…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 >= 25.0.1, NumPy is absent, or the column is not list/map/struct. 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 <17627829+Yicong-Huang@users.noreply.github.com> Signed-off-by: Yicong-Huang <17627829+Yicong-Huang@users.noreply.github.com> (cherry picked from commit b1547b4) Signed-off-by: Yicong-Huang <17627829+Yicong-Huang@users.noreply.github.com>
Yicong-Huang added a commit
that referenced
this pull request
Jul 17, 2026
…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 >= 25.0.1, NumPy is absent, or the column is not list/map/struct. 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 <17627829+Yicong-Huang@users.noreply.github.com> Signed-off-by: Yicong-Huang <17627829+Yicong-Huang@users.noreply.github.com> (cherry picked from commit b1547b4) Signed-off-by: Yicong-Huang <17627829+Yicong-Huang@users.noreply.github.com>
Yicong-Huang
commented
Jul 17, 2026
ContributorAuthor
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for freeto join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
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 forArray.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 toto_pylist()when the installed PyArrow is >= 25.0.1, NumPy is absent, or the column is not list/map/struct.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 ConnectDataFrame.foreachPartitionpath, and the row-based Python data source write path (batch_to_rowsinwrite_into_data_source.py).plan_data_source_read.pyis intentionally left unchanged: it converts a singleBinaryTypecell (num_columns == 1, num_rows == 1, a pickledInputPartition), where the bulk path has zero benefit and binary falls through toto_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_pylistis byte-identical toto_pylist().How was this patch tested?
No behavior change; the byte-identical contract of
_to_pylistis covered by the existingtest_matches_to_pylistinpython/pyspark/sql/tests/test_conversion.py. Both call sites already have integration coverage (foreachPartitionandDataSourceWritertests).Was this patch authored or co-authored using generative AI tooling?
No