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 <[email protected]>
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 <[email protected]>
Signed-off-by: Liang-Chi Hsieh <[email protected]>
(cherry picked from commit 9c6e57bf7e0c6f5ab4870d28611449d36ab7c99c)
Signed-off-by: Liang-Chi Hsieh <[email protected]>
---
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: [email protected]
For additional commands, e-mail: [email protected]
