You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by vi...@apache.org on 2026/07/11 05:40:33 UTC
(spark) branch branch-4.2 updated: [SPARK-58024][PYTHON] Convert Arrow struct and map columns to Python rows in bulk
This is an automated email from the ASF dual-hosted git repository.
viirya pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 44daf0131f7b [SPARK-58024][PYTHON] Convert Arrow struct and map columns to Python rows in bulk
44daf0131f7b is described below
commit 44daf0131f7bd8be5e4c6864e19d73b0dd078dca
Author: Liang-Chi Hsieh <vi...@gmail.com>
AuthorDate: Fri Jul 10 22:40:05 2026 -0700
[SPARK-58024][PYTHON] Convert Arrow struct and map columns to Python rows in bulk
### What changes were proposed in this pull request?
Follow-up of SPARK-58019 (#57099); **only the last commit is new — this PR is stacked on #57099 and will be rebased once it merges.** (Originally stacked on SPARK-58023 (#57104), which has been closed in favor of the upstream apache/arrow#50326 fix; the numbers below are measured without it.)
Extend `ArrowTableToRowsConversion._to_pylist` with bulk paths for struct and map columns:
- **Struct** columns convert each child field in bulk (recursively reusing the bulk list paths), then zip the field values into one dict per row, masked by the validity bitmap. Duplicate field names fall back to `to_pylist`, so they raise the same `ValueError` that `StructScalar.as_py` raises.
- **Map** columns share the list offsets layout: the flattened keys and items children are each converted in bulk, and every row becomes a list of `(key, value)` tuples, matching `MapScalar.as_py` exactly.
### Why are the changes needed?
Struct and map columns still convert one Scalar per row; per-row Scalar/wrapper allocation dominates (apache/arrow#50326), and maps are the worst case since every row also wraps its keys/items in per-row Arrays. ASV microbenchmark (`bench_arrow.ArrowStructMapColumnToRowsBenchmark`, 1M rows, 10% nulls, PyArrow 24.0.0):
| case | `to_pylist()` | this PR | speedup |
|---|---|---|---|
| `struct<int64,string>` | 970 ms | 474 ms | 2.0x |
| `map<string,int64>` (2 entries/row) | 2.16 s | 799 ms | 2.7x |
Peak memory unchanged.
### Does this PR introduce _any_ user-facing change?
No. Only performance; conversion results are identical (covered by exact-type tests).
### How was this patch tested?
Extended `ArrowColumnToPylistTests` with struct (incl. nested struct/list children, empty struct, all-null), map (incl. list values), struct-of-map, list-of-struct, sliced and chunked views — all compared against `column.to_pylist()` with exact element-type assertions — plus dedicated tests that duplicate struct field names still raise `ValueError` and that empty-struct rows are distinct dict objects. New ASV benchmark class `ArrowStructMapColumnToRowsBenchmark`.
### Was this patch authored or co-authored using generative AI tooling?
Yes. This pull request and its description were written by Isaac (Claude Code).
Closes #57105 from viirya/arrow-to-pylist-maps-structs.
Authored-by: Liang-Chi Hsieh <vi...@gmail.com>
Signed-off-by: Liang-Chi Hsieh <vi...@gmail.com>
(cherry picked from commit 9c6e57bf7e0c6f5ab4870d28611449d36ab7c99c)
Signed-off-by: Liang-Chi Hsieh <vi...@gmail.com>
---
python/benchmarks/bench_arrow.py | 43 ++++++++++++++++++
python/pyspark/sql/conversion.py | 68 ++++++++++++++++++++++++++---
python/pyspark/sql/tests/test_conversion.py | 46 +++++++++++++++++++
3 files changed, 152 insertions(+), 5 deletions(-)
diff --git a/python/benchmarks/bench_arrow.py b/python/benchmarks/bench_arrow.py
index b59a43d42072..4ed1c91d579c 100644
--- a/python/benchmarks/bench_arrow.py
+++ b/python/benchmarks/bench_arrow.py
@@ -170,3 +170,46 @@ class ArrowListColumnToRowsBenchmark:
def peakmem_array_of_structs_to_rows(self, n_rows, method):
self.convert(self.array_of_structs)
+
+
+class ArrowStructMapColumnToRowsBenchmark:
+ """
+ Benchmark for converting Arrow struct and map columns to Python rows.
+
+ ``baseline`` measures plain ``column.to_pylist()``; ``bulk`` measures
+ ``ArrowTableToRowsConversion._to_pylist`` with the struct/map bulk paths.
+ """
+
+ params = [
+ [100000, 1000000],
+ ["baseline", "bulk"],
+ ]
+ param_names = ["n_rows", "method"]
+
+ def setup(self, n_rows, method):
+ from pyspark.sql.conversion import ArrowTableToRowsConversion
+
+ self.structs = pa.array(
+ [{"a": i, "b": f"s{i}"} if i % 10 != 0 else None for i in range(n_rows)],
+ type=pa.struct([("a", pa.int64()), ("b", pa.string())]),
+ )
+ self.maps = pa.array(
+ [
+ [(f"k{i % 3}", i), (f"q{i % 5}", i + 1)] if i % 10 != 0 else None
+ for i in range(n_rows)
+ ],
+ type=pa.map_(pa.string(), pa.int64()),
+ )
+ if method == "bulk":
+ self.convert = ArrowTableToRowsConversion._to_pylist
+ else:
+ self.convert = lambda column: column.to_pylist()
+
+ def time_structs_to_rows(self, n_rows, method):
+ self.convert(self.structs)
+
+ def time_maps_to_rows(self, n_rows, method):
+ self.convert(self.maps)
+
+ def peakmem_structs_to_rows(self, n_rows, method):
+ self.convert(self.structs)
diff --git a/python/pyspark/sql/conversion.py b/python/pyspark/sql/conversion.py
index e22915b5a065..0550ff4bb56f 100644
--- a/python/pyspark/sql/conversion.py
+++ b/python/pyspark/sql/conversion.py
@@ -947,8 +947,11 @@ class ArrowTableToRowsConversion:
@staticmethod
def _to_pylist(column: Union["pa.Array", "pa.ChunkedArray"]) -> List[Any]:
"""
- Equivalent to ``column.to_pylist()``, but converts (nested) list columns in bulk
- instead of one scalar at a time.
+ Equivalent to ``column.to_pylist()``, but converts (nested) list, struct and map
+ columns in bulk instead of one scalar at a time. Structs become dicts (with
+ a fallback to ``to_pylist`` for duplicate field names, which raise ``ValueError``
+ there) and maps become lists of ``(key, value)`` tuples, matching
+ ``StructScalar.as_py`` and ``MapScalar.as_py`` exactly.
Internal helper for the worker and ``convert`` call sites; do not use
externally.
@@ -979,9 +982,44 @@ class ArrowTableToRowsConversion:
result.extend(ArrowTableToRowsConversion._to_pylist(chunk))
return result
- if (pa.types.is_list(column.type) or pa.types.is_large_list(column.type)) and len(
- column
- ) > 0:
+ if len(column) == 0:
+ return []
+
+ if pa.types.is_map(column.type):
+ # Maps have the same offsets layout as lists; each row becomes a
+ # list of (key, value) tuples, matching MapScalar.as_py.
+ n = len(column)
+ offsets = column.offsets.to_numpy(zero_copy_only=True).tolist()
+ start = offsets[0]
+ length = offsets[-1] - start
+ keys = ArrowTableToRowsConversion._to_pylist(column.keys.slice(start, length))
+ items = ArrowTableToRowsConversion._to_pylist(column.items.slice(start, length))
+ if column.null_count == 0:
+ return [
+ list(
+ zip(
+ keys[offsets[i] - start : offsets[i + 1] - start],
+ items[offsets[i] - start : offsets[i + 1] - start],
+ )
+ )
+ for i in range(n)
+ ]
+ valid = column.is_valid().to_numpy(zero_copy_only=False).tolist()
+ return [
+ (
+ list(
+ zip(
+ keys[offsets[i] - start : offsets[i + 1] - start],
+ items[offsets[i] - start : offsets[i + 1] - start],
+ )
+ )
+ if valid[i]
+ else None
+ )
+ for i in range(n)
+ ]
+
+ elif pa.types.is_list(column.type) or pa.types.is_large_list(column.type):
n = len(column)
# List offset buffers never carry a validity bitmap, so this conversion is
# always zero-copy; zero_copy_only=True asserts that invariant and would
@@ -999,6 +1037,26 @@ class ArrowTableToRowsConversion:
for i in range(n)
]
+ elif pa.types.is_struct(column.type):
+ n = len(column)
+ names = [column.type.field(i).name for i in range(column.type.num_fields)]
+ if len(set(names)) != len(names):
+ # StructScalar.as_py raises ValueError on duplicate field names;
+ # let the generic path surface the same error.
+ return column.to_pylist()
+ fields = [
+ ArrowTableToRowsConversion._to_pylist(column.field(i))
+ for i in range(column.type.num_fields)
+ ]
+ if column.null_count == 0:
+ if not names:
+ return [{} for _ in range(n)]
+ return [dict(zip(names, row)) for row in zip(*fields)]
+ valid = column.is_valid().to_numpy(zero_copy_only=False).tolist()
+ if not names:
+ return [{} if m else None for m in valid]
+ return [dict(zip(names, row)) if m else None for row, m in zip(zip(*fields), valid)]
+
return column.to_pylist()
@staticmethod
diff --git a/python/pyspark/sql/tests/test_conversion.py b/python/pyspark/sql/tests/test_conversion.py
index c5ce79b9d188..872a0c1e248f 100644
--- a/python/pyspark/sql/tests/test_conversion.py
+++ b/python/pyspark/sql/tests/test_conversion.py
@@ -15,6 +15,7 @@
# limitations under the License.
#
import datetime
+import decimal
import unittest
import unittest.mock
from zoneinfo import ZoneInfo
@@ -843,6 +844,37 @@ class ArrowColumnToPylistTests(unittest.TestCase):
pa.array([], type=pa.list_(pa.int32())),
pa.array([None, None], type=pa.list_(pa.string())),
pa.array([[1, 2], None], type=pa.list_(pa.int64(), 2)),
+ # non-list leaves keep as_py semantics (native to_pylist)
+ pa.array([b"", None, b"\x00\xff"], type=pa.binary()),
+ pa.array([datetime.date(2020, 1, 2), None], type=pa.date32()),
+ pa.array([decimal.Decimal("1.23"), None], type=pa.decimal128(10, 2)),
+ pa.array([[b"x", None], None, [b""]], type=pa.list_(pa.binary())),
+ pa.array([[True, None], [False]], type=pa.list_(pa.bool_())),
+ # struct and map bulk paths
+ pa.array(
+ [{"a": 1, "b": "x"}, None, {"a": None, "b": None}],
+ type=pa.struct([("a", pa.int64()), ("b", pa.string())]),
+ ),
+ pa.array(
+ [{"s": {"a": 1}, "l": [1, None]}, None],
+ type=pa.struct(
+ [("s", pa.struct([("a", pa.int32())])), ("l", pa.list_(pa.int64()))]
+ ),
+ ),
+ pa.array([{}, None, {}], type=pa.struct([])),
+ pa.array([None] * 4, type=pa.struct([("a", pa.int32())])),
+ pa.array(
+ [[("k1", [1, None]), ("k2", None)], None, []],
+ type=pa.map_(pa.string(), pa.list_(pa.int32())),
+ ),
+ pa.array(
+ [{"m": [("k", 1)]}, None],
+ type=pa.struct([("m", pa.map_(pa.string(), pa.int64()))]),
+ ),
+ pa.array(
+ [[{"a": 1}, None], None],
+ type=pa.list_(pa.struct([("a", pa.int64())])),
+ ),
]
for column in columns:
views = [column, column.slice(1), column.slice(0, max(len(column) - 1, 0))]
@@ -866,6 +898,20 @@ class ArrowColumnToPylistTests(unittest.TestCase):
self.assertEqual(result, [[1, None, 3]])
self.assertEqual([type(v) for v in result[0]], [int, type(None), int])
+ def test_struct_duplicate_field_names_still_raises(self):
+ import pyarrow as pa
+
+ dup = pa.StructArray.from_arrays([pa.array([1, 2]), pa.array(["a", "b"])], names=["x", "x"])
+ with self.assertRaises(ValueError):
+ ArrowTableToRowsConversion._to_pylist(dup)
+
+ def test_struct_rows_are_distinct_dicts(self):
+ import pyarrow as pa
+
+ result = ArrowTableToRowsConversion._to_pylist(pa.array([{}, {}], type=pa.struct([])))
+ self.assertEqual(result, [{}, {}])
+ self.assertIsNot(result[0], result[1])
+
def test_convert_table_with_list_columns(self):
import pyarrow as pa
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org