From 51771f59ae2c5f7b4452d2f357b072e0c01ec716 Mon Sep 17 00:00:00 2001
From: Liang-Chi Hsieh <viirya@gmail.com>
Date: Tue, 7 Jul 2026 16:43:41 -0700
Subject: [PATCH 1/9] [SPARK-58019][PYTHON] Convert Arrow list columns to
 Python rows in bulk

PyArrow's Array.to_pylist() materializes one Scalar per element; for
list-typed columns each row additionally allocates a C++ scalar, a
Python Scalar wrapper and a Python Array wrapper before converting
elements one by one. This makes Arrow-optimized Python UDF inputs and
Spark Connect collect() several times slower on array columns than
necessary (apache/arrow#50326; a fix is proposed upstream in
apache/arrow#50327 but will only be available in a future PyArrow
release).

Add ArrowTableToRowsConversion._to_pylist, which converts the flattened
child values of a list column in a single pass and slices the resulting
Python list per row using the offsets and the validity bitmap, and use
it in the Arrow-to-rows conversion paths (Spark Connect collect, Arrow
batch UDF inputs, Arrow UDTF inputs). Leaf values are still converted by
Arrow's own to_pylist, so results are exactly identical - None stays
None and values inside numeric lists stay Python ints, unlike a pandas
round trip which coerces them to floats/NaN. NumPy is only used for the
offsets and validity bitmap, never for the values.

ASV microbenchmark (python/benchmarks/bench_arrow.py, 1M rows):
list<string> 769ms -> 507ms (1.5x); list<list<int32>> with nulls
1.86s -> 537ms (3.5x). Peak memory unchanged.

The helper can be removed once the minimum supported PyArrow version
includes the upstream fix.

Co-authored-by: Isaac
---
 python/benchmarks/bench_arrow.py            | 40 +++++++++++
 python/pyspark/sql/conversion.py            | 66 ++++++++++++++++--
 python/pyspark/sql/tests/test_conversion.py | 77 +++++++++++++++++++++
 python/pyspark/worker.py                    | 32 +++++----
 4 files changed, 196 insertions(+), 19 deletions(-)

diff --git a/python/benchmarks/bench_arrow.py b/python/benchmarks/bench_arrow.py
index 29a95dbcc98b5..81189707016b6 100644
--- a/python/benchmarks/bench_arrow.py
+++ b/python/benchmarks/bench_arrow.py
@@ -114,3 +114,43 @@ def time_long_with_nulls_to_pandas_ext(self, n_rows, method):
 
     def peakmem_long_with_nulls_to_pandas_ext(self, n_rows, method):
         self.run_long_with_nulls_to_pandas_ext(n_rows, method)
+
+
+class ArrowListColumnToRowsBenchmark:
+    """
+    Benchmark for converting Arrow list-typed columns to Python rows, the hot
+    path of Arrow-optimized Python UDF inputs and Spark Connect collect().
+
+    ``baseline`` measures plain ``column.to_pylist()``; ``bulk`` measures
+    ``ArrowTableToRowsConversion._to_pylist`` (see apache/arrow#50326).
+    """
+
+    params = [
+        [100000, 1000000],
+        ["baseline", "bulk"],
+    ]
+    param_names = ["n_rows", "method"]
+
+    def setup(self, n_rows, method):
+        from pyspark.sql.conversion import ArrowTableToRowsConversion
+
+        self.list_of_strings = pa.array(
+            [[f"s{i}", f"t{i}"] for i in range(n_rows)], type=pa.list_(pa.string())
+        )
+        self.nested_ints_with_nulls = pa.array(
+            [[[i, i + 1], None, [i + 2]] if i % 10 != 0 else None for i in range(n_rows)],
+            type=pa.list_(pa.list_(pa.int32())),
+        )
+        if method == "bulk":
+            self.convert = ArrowTableToRowsConversion._to_pylist
+        else:
+            self.convert = lambda column: column.to_pylist()
+
+    def time_list_of_strings_to_rows(self, n_rows, method):
+        self.convert(self.list_of_strings)
+
+    def time_nested_ints_with_nulls_to_rows(self, n_rows, method):
+        self.convert(self.nested_ints_with_nulls)
+
+    def peakmem_list_of_strings_to_rows(self, n_rows, method):
+        self.convert(self.list_of_strings)
diff --git a/python/pyspark/sql/conversion.py b/python/pyspark/sql/conversion.py
index bfa0d4a559a51..e3342adce2012 100644
--- a/python/pyspark/sql/conversion.py
+++ b/python/pyspark/sql/conversion.py
@@ -980,6 +980,60 @@ class ArrowTableToRowsConversion:
     Conversion from Arrow Table to Rows.
     """
 
+    @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.
+
+        ``Array.to_pylist()`` materializes one Scalar per element; for list types each row
+        additionally allocates a C++ scalar, a Python Scalar wrapper and a Python Array
+        wrapper for the row's values before converting elements one by one, which is
+        several times slower than converting the flattened child values in a single pass
+        and slicing the resulting Python list per row (see apache/arrow#50326). The values
+        themselves are still converted by Arrow's own ``to_pylist``, so results are exactly
+        identical: ``None`` stays ``None`` and values inside numeric lists stay Python ints,
+        unlike a pandas round trip which would coerce them to floats/NaN. NumPy is used
+        only for the offsets (non-null integers) and the validity bitmap (booleans), so no
+        value coercion can occur.
+
+        This can be removed once the minimum supported PyArrow version includes the fix
+        for apache/arrow#50326.
+        """
+        import pyarrow as pa
+        import pyarrow.types as pa_types
+
+        try:
+            import numpy  # noqa: F401
+        except ImportError:
+            return column.to_pylist()
+
+        if isinstance(column, pa.ChunkedArray):
+            result: List[Any] = []
+            for chunk in column.chunks:
+                result.extend(ArrowTableToRowsConversion._to_pylist(chunk))
+            return result
+
+        column_type = column.type
+        if (pa_types.is_list(column_type) or pa_types.is_large_list(column_type)) and len(
+            column
+        ) > 0:
+            n = len(column)
+            offsets = column.offsets.to_numpy(zero_copy_only=True).tolist()
+            start = offsets[0]
+            flat = ArrowTableToRowsConversion._to_pylist(
+                column.values.slice(start, offsets[-1] - start)
+            )
+            if column.null_count == 0:
+                return [flat[offsets[i] - start : offsets[i + 1] - start] for i in range(n)]
+            valid = column.is_valid().to_numpy(zero_copy_only=False).tolist()
+            return [
+                flat[offsets[i] - start : offsets[i + 1] - start] if valid[i] else None
+                for i in range(n)
+            ]
+
+        return column.to_pylist()
+
     @staticmethod
     def _need_converter(dataType: DataType) -> bool:
         if isinstance(dataType, NullType):
@@ -1069,9 +1123,9 @@ def convert_struct(value: Any) -> Any:
                 dataType.elementType, none_on_identity=True, binary_as_bytes=binary_as_bytes
             )
 
-            assert element_conv is not None, (
-                f"_need_converter() returned True for ArrayType of {dataType.elementType}"
-            )
+            assert (
+                element_conv is not None
+            ), f"_need_converter() returned True for ArrayType of {dataType.elementType}"
 
             def convert_array(value: Any) -> Any:
                 if value is None:
@@ -1306,7 +1360,11 @@ def convert(
             ]
 
             columnar_data = [
-                [conv(v) for v in column.to_pylist()] if conv is not None else column.to_pylist()
+                (
+                    [conv(v) for v in ArrowTableToRowsConversion._to_pylist(column)]
+                    if conv is not None
+                    else ArrowTableToRowsConversion._to_pylist(column)
+                )
                 for column, conv in zip(table.columns, field_converters)
             ]
 
diff --git a/python/pyspark/sql/tests/test_conversion.py b/python/pyspark/sql/tests/test_conversion.py
index dd5c7f44d2818..0944e49664cb8 100644
--- a/python/pyspark/sql/tests/test_conversion.py
+++ b/python/pyspark/sql/tests/test_conversion.py
@@ -844,6 +844,83 @@ def test_geometry_convert_numpy(self):
         self.assertEqual(len(result), 0)
 
 
+class ArrowColumnToPylistTests(unittest.TestCase):
+    """
+    ArrowTableToRowsConversion._to_pylist must return exactly what
+    column.to_pylist() returns, including exact element types.
+    """
+
+    def _assert_identical_types(self, actual, expected):
+        self.assertIs(type(actual), type(expected))
+        if isinstance(actual, (list, tuple)):
+            self.assertEqual(len(actual), len(expected))
+            for a, e in zip(actual, expected):
+                self._assert_identical_types(a, e)
+
+    def test_matches_to_pylist(self):
+        import pyarrow as pa
+
+        columns = [
+            pa.array([[1, None, 3], None, [], [4]], type=pa.list_(pa.int32())),
+            pa.array([["a", None], None, [], ["bcd", ""]], type=pa.list_(pa.string())),
+            pa.array([["a", None], None, ["b"]], type=pa.large_list(pa.string())),
+            pa.array([[[1], None, [2, None]], None], type=pa.list_(pa.list_(pa.int32()))),
+            pa.array(
+                [[{"a": 1, "b": "x"}, None], None],
+                type=pa.list_(pa.struct([("a", pa.int32()), ("b", pa.string())])),
+            ),
+            pa.array([[("k1", 1), ("k2", None)], None, []], type=pa.map_(pa.string(), pa.int32())),
+            pa.array([[1.5, None], [float("nan")]], type=pa.list_(pa.float64())),
+            pa.array([1, None, 3], type=pa.int64()),
+            pa.array(["x", None], type=pa.string()),
+            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)),
+        ]
+        for column in columns:
+            views = [column, column.slice(1), column.slice(0, max(len(column) - 1, 0))]
+            views.append(pa.chunked_array([column, column.slice(1)], type=column.type))
+            for view in views:
+                with self.subTest(type=str(column.type), length=len(view)):
+                    actual = ArrowTableToRowsConversion._to_pylist(view)
+                    expected = view.to_pylist()
+                    # NaN != NaN; compare via repr for the float case
+                    self.assertEqual(repr(actual), repr(expected))
+                    self._assert_identical_types(actual, expected)
+
+    def test_int_list_with_nulls_stays_int(self):
+        # The exact case that makes a pandas round trip unusable: ints must not
+        # become floats/NaN when the list contains nulls.
+        import pyarrow as pa
+
+        result = ArrowTableToRowsConversion._to_pylist(
+            pa.array([[1, None, 3]], type=pa.list_(pa.int32()))
+        )
+        self.assertEqual(result, [[1, None, 3]])
+        self.assertEqual([type(v) for v in result[0]], [int, type(None), int])
+
+    def test_convert_table_with_list_columns(self):
+        import pyarrow as pa
+
+        schema = (
+            StructType()
+            .add("arr", ArrayType(IntegerType()))
+            .add("nested", ArrayType(ArrayType(StringType())))
+        )
+        tbl = pa.table(
+            {
+                "arr": pa.array([[1, None], None, []], type=pa.list_(pa.int32())),
+                "nested": pa.array(
+                    [[["a"], None], [[]], None], type=pa.list_(pa.list_(pa.string()))
+                ),
+            }
+        )
+        actual = ArrowTableToRowsConversion.convert(tbl, schema)
+        self.assertEqual(actual[0], Row(arr=[1, None], nested=[["a"], None]))
+        self.assertEqual(actual[1], Row(arr=None, nested=[[]]))
+        self.assertEqual(actual[2], Row(arr=[], nested=None))
+
+
 if __name__ == "__main__":
     from pyspark.testing import main
 
diff --git a/python/pyspark/worker.py b/python/pyspark/worker.py
index 9e629138d5370..cc23eace59bd9 100644
--- a/python/pyspark/worker.py
+++ b/python/pyspark/worker.py
@@ -1755,9 +1755,9 @@ def func(split_index: int, data: Iterator[pa.RecordBatch]) -> Iterator[pa.Record
                     # then call eval once per input row.
                     pylist = [
                         (
-                            [conv(v) for v in column.to_pylist()]
+                            [conv(v) for v in ArrowTableToRowsConversion._to_pylist(column)]
                             if conv is not None
-                            else column.to_pylist()
+                            else ArrowTableToRowsConversion._to_pylist(column)
                         )
                         for column, conv in zip(batch.columns, converters)
                     ]
@@ -2616,9 +2616,9 @@ def grouped_func(
         assert num_udfs == 1, "One GROUPED_MAP_ARROW_ITER UDF expected here."
         grouped_udf, arg_offsets, return_type, num_udf_args = udfs[0]
         parsed_offsets = extract_key_value_indexes(arg_offsets)
-        assert len(parsed_offsets) == 1, (
-            "Expected one pair of offsets for GROUPED_MAP_ARROW_ITER UDF."
-        )
+        assert (
+            len(parsed_offsets) == 1
+        ), "Expected one pair of offsets for GROUPED_MAP_ARROW_ITER UDF."
 
         arrow_return_type = to_arrow_type(
             return_type, timezone="UTC", prefers_large_types=runner_conf.use_large_var_types
@@ -2752,9 +2752,9 @@ def grouped_func(
         assert num_udfs == 1, "One GROUPED_MAP_PANDAS_ITER UDF expected here."
         grouped_udf, arg_offsets, return_type, num_udf_args = udfs[0]
         parsed_offsets = extract_key_value_indexes(arg_offsets)
-        assert len(parsed_offsets) == 1, (
-            "Expected one pair of offsets for GROUPED_MAP_PANDAS_ITER UDF."
-        )
+        assert (
+            len(parsed_offsets) == 1
+        ), "Expected one pair of offsets for GROUPED_MAP_PANDAS_ITER UDF."
 
         key_offsets = parsed_offsets[0][0]
         value_offsets = parsed_offsets[0][1]
@@ -3067,7 +3067,11 @@ def func(split_index: int, data: Iterator[pa.RecordBatch]) -> Iterator[pa.Record
 
                 # --- Input: Arrow -> Python columns ---
                 columns = [
-                    [conv(v) for v in col.to_pylist()] if conv is not None else col.to_pylist()
+                    (
+                        [conv(v) for v in ArrowTableToRowsConversion._to_pylist(col)]
+                        if conv is not None
+                        else ArrowTableToRowsConversion._to_pylist(col)
+                    )
                     for col, conv in zip(input_batch.itercolumns(), arrow_to_py_converters)
                 ]
                 if not columns:
@@ -3119,9 +3123,7 @@ def func(split_index: int, data: Iterator[pa.RecordBatch]) -> Iterator[pa.Record
             coerce = (
                 str
                 if isinstance(udf_return_type, StringType)
-                else bytes
-                if isinstance(udf_return_type, BinaryType)
-                else None
+                else bytes if isinstance(udf_return_type, BinaryType) else None
             )
             udf_infos.append(
                 (
@@ -3363,9 +3365,9 @@ def process_results():
         # See TransformWithStateInPandasExec for how arg_offsets are used to
         # distinguish between grouping attributes and data attributes
         parsed_offsets = extract_key_value_indexes(arg_offsets)
-        assert len(parsed_offsets) == 1, (
-            "Expected one pair of offsets for TRANSFORM_WITH_STATE_PANDAS UDF."
-        )
+        assert (
+            len(parsed_offsets) == 1
+        ), "Expected one pair of offsets for TRANSFORM_WITH_STATE_PANDAS UDF."
 
         key_offsets = parsed_offsets[0][0]
         value_offsets = parsed_offsets[0][1]

From eebbc4fd05957ca316041ff3682a2473dce5a62b Mon Sep 17 00:00:00 2001
From: Liang-Chi Hsieh <viirya@gmail.com>
Date: Tue, 7 Jul 2026 18:11:05 -0700
Subject: [PATCH 2/9] [SPARK-58019][PYTHON] Document the zero-copy offsets
 invariant

Co-authored-by: Isaac
---
 python/pyspark/sql/conversion.py | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/python/pyspark/sql/conversion.py b/python/pyspark/sql/conversion.py
index e3342adce2012..94ced9f9c58ae 100644
--- a/python/pyspark/sql/conversion.py
+++ b/python/pyspark/sql/conversion.py
@@ -1019,6 +1019,9 @@ def _to_pylist(column: Union["pa.Array", "pa.ChunkedArray"]) -> List[Any]:
             column
         ) > 0:
             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
+            # fail loudly if a future Arrow list variant ever violated it.
             offsets = column.offsets.to_numpy(zero_copy_only=True).tolist()
             start = offsets[0]
             flat = ArrowTableToRowsConversion._to_pylist(

From 691885a81adcba6b6ef806da0c45448b5289c1fc Mon Sep 17 00:00:00 2001
From: Liang-Chi Hsieh <viirya@gmail.com>
Date: Tue, 7 Jul 2026 21:00:12 -0700
Subject: [PATCH 3/9] [SPARK-58019][PYTHON] Restore ruff formatting

The files were formatted with black 26.3.1 (dev/requirements.txt), but
the CI linter checks with ruff format, which disagrees on hunks
unrelated to this change. Reformat with ruff 0.14.8 to match CI.

Co-authored-by: Isaac
---
 python/pyspark/sql/conversion.py |  6 +++---
 python/pyspark/worker.py         | 22 ++++++++++++----------
 2 files changed, 15 insertions(+), 13 deletions(-)

diff --git a/python/pyspark/sql/conversion.py b/python/pyspark/sql/conversion.py
index 94ced9f9c58ae..2e4c5bd93500b 100644
--- a/python/pyspark/sql/conversion.py
+++ b/python/pyspark/sql/conversion.py
@@ -1126,9 +1126,9 @@ def convert_struct(value: Any) -> Any:
                 dataType.elementType, none_on_identity=True, binary_as_bytes=binary_as_bytes
             )
 
-            assert (
-                element_conv is not None
-            ), f"_need_converter() returned True for ArrayType of {dataType.elementType}"
+            assert element_conv is not None, (
+                f"_need_converter() returned True for ArrayType of {dataType.elementType}"
+            )
 
             def convert_array(value: Any) -> Any:
                 if value is None:
diff --git a/python/pyspark/worker.py b/python/pyspark/worker.py
index cc23eace59bd9..83ceeed1bd102 100644
--- a/python/pyspark/worker.py
+++ b/python/pyspark/worker.py
@@ -2616,9 +2616,9 @@ def grouped_func(
         assert num_udfs == 1, "One GROUPED_MAP_ARROW_ITER UDF expected here."
         grouped_udf, arg_offsets, return_type, num_udf_args = udfs[0]
         parsed_offsets = extract_key_value_indexes(arg_offsets)
-        assert (
-            len(parsed_offsets) == 1
-        ), "Expected one pair of offsets for GROUPED_MAP_ARROW_ITER UDF."
+        assert len(parsed_offsets) == 1, (
+            "Expected one pair of offsets for GROUPED_MAP_ARROW_ITER UDF."
+        )
 
         arrow_return_type = to_arrow_type(
             return_type, timezone="UTC", prefers_large_types=runner_conf.use_large_var_types
@@ -2752,9 +2752,9 @@ def grouped_func(
         assert num_udfs == 1, "One GROUPED_MAP_PANDAS_ITER UDF expected here."
         grouped_udf, arg_offsets, return_type, num_udf_args = udfs[0]
         parsed_offsets = extract_key_value_indexes(arg_offsets)
-        assert (
-            len(parsed_offsets) == 1
-        ), "Expected one pair of offsets for GROUPED_MAP_PANDAS_ITER UDF."
+        assert len(parsed_offsets) == 1, (
+            "Expected one pair of offsets for GROUPED_MAP_PANDAS_ITER UDF."
+        )
 
         key_offsets = parsed_offsets[0][0]
         value_offsets = parsed_offsets[0][1]
@@ -3123,7 +3123,9 @@ def func(split_index: int, data: Iterator[pa.RecordBatch]) -> Iterator[pa.Record
             coerce = (
                 str
                 if isinstance(udf_return_type, StringType)
-                else bytes if isinstance(udf_return_type, BinaryType) else None
+                else bytes
+                if isinstance(udf_return_type, BinaryType)
+                else None
             )
             udf_infos.append(
                 (
@@ -3365,9 +3367,9 @@ def process_results():
         # See TransformWithStateInPandasExec for how arg_offsets are used to
         # distinguish between grouping attributes and data attributes
         parsed_offsets = extract_key_value_indexes(arg_offsets)
-        assert (
-            len(parsed_offsets) == 1
-        ), "Expected one pair of offsets for TRANSFORM_WITH_STATE_PANDAS UDF."
+        assert len(parsed_offsets) == 1, (
+            "Expected one pair of offsets for TRANSFORM_WITH_STATE_PANDAS UDF."
+        )
 
         key_offsets = parsed_offsets[0][0]
         value_offsets = parsed_offsets[0][1]

From 784974ceeceb1605e6586c222b0cbb86c6084fb9 Mon Sep 17 00:00:00 2001
From: Liang-Chi Hsieh <viirya@gmail.com>
Date: Tue, 7 Jul 2026 23:34:30 -0700
Subject: [PATCH 4/9] [SPARK-58019][PYTHON] Skip Arrow to_pylist tests without
 PyArrow

Co-authored-by: Isaac
---
 python/pyspark/sql/tests/test_conversion.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/python/pyspark/sql/tests/test_conversion.py b/python/pyspark/sql/tests/test_conversion.py
index 0944e49664cb8..b559d3ac3081e 100644
--- a/python/pyspark/sql/tests/test_conversion.py
+++ b/python/pyspark/sql/tests/test_conversion.py
@@ -844,6 +844,7 @@ def test_geometry_convert_numpy(self):
         self.assertEqual(len(result), 0)
 
 
+@unittest.skipIf(not have_pyarrow, pyarrow_requirement_message)
 class ArrowColumnToPylistTests(unittest.TestCase):
     """
     ArrowTableToRowsConversion._to_pylist must return exactly what

From 1fd52c8e3517278c229ac0120c137d99f0c9feb8 Mon Sep 17 00:00:00 2001
From: Liang-Chi Hsieh <viirya@gmail.com>
Date: Wed, 8 Jul 2026 13:06:22 -0700
Subject: [PATCH 5/9] [SPARK-58019][PYTHON] Cache NumPy availability; add
 peakmem and array<struct> benchmarks

Address review: cache the NumPy availability check in a module-level
helper instead of re-running try/import on every (recursive)
invocation, and extend the ASV benchmark with an array<struct> case
plus peakmem variants for the nested cases (1M rows: nested list
862M -> 862M, array<struct> 885M -> 921M, ~4% peak increase from the
transient flattened pointer list; leaf objects are shared).

Co-authored-by: Isaac
---
 python/benchmarks/bench_arrow.py | 16 ++++++++++++++++
 python/pyspark/sql/conversion.py | 19 ++++++++++++++++---
 2 files changed, 32 insertions(+), 3 deletions(-)

diff --git a/python/benchmarks/bench_arrow.py b/python/benchmarks/bench_arrow.py
index 81189707016b6..b59a43d420725 100644
--- a/python/benchmarks/bench_arrow.py
+++ b/python/benchmarks/bench_arrow.py
@@ -141,6 +141,13 @@ def setup(self, n_rows, method):
             [[[i, i + 1], None, [i + 2]] if i % 10 != 0 else None for i in range(n_rows)],
             type=pa.list_(pa.list_(pa.int32())),
         )
+        self.array_of_structs = pa.array(
+            [
+                [{"i": i, "s": f"a{i}"}, {"i": i + 1, "s": f"b{i}"}] if i % 10 != 0 else None
+                for i in range(n_rows)
+            ],
+            type=pa.list_(pa.struct([("i", pa.int32()), ("s", pa.string())])),
+        )
         if method == "bulk":
             self.convert = ArrowTableToRowsConversion._to_pylist
         else:
@@ -152,5 +159,14 @@ def time_list_of_strings_to_rows(self, n_rows, method):
     def time_nested_ints_with_nulls_to_rows(self, n_rows, method):
         self.convert(self.nested_ints_with_nulls)
 
+    def time_array_of_structs_to_rows(self, n_rows, method):
+        self.convert(self.array_of_structs)
+
     def peakmem_list_of_strings_to_rows(self, n_rows, method):
         self.convert(self.list_of_strings)
+
+    def peakmem_nested_ints_with_nulls_to_rows(self, n_rows, method):
+        self.convert(self.nested_ints_with_nulls)
+
+    def peakmem_array_of_structs_to_rows(self, n_rows, method):
+        self.convert(self.array_of_structs)
diff --git a/python/pyspark/sql/conversion.py b/python/pyspark/sql/conversion.py
index 2e4c5bd93500b..bd545e3ab2f61 100644
--- a/python/pyspark/sql/conversion.py
+++ b/python/pyspark/sql/conversion.py
@@ -506,6 +506,21 @@ def convert_column(
         return pa.RecordBatch.from_arrays(arrays, schema.names)
 
 
+_numpy_available: Optional[bool] = None
+
+
+def _is_numpy_available() -> bool:
+    global _numpy_available
+    if _numpy_available is None:
+        try:
+            import numpy  # noqa: F401
+
+            _numpy_available = True
+        except ImportError:
+            _numpy_available = False
+    return _numpy_available
+
+
 class LocalDataToArrowConversion:
     """
     Conversion from local data (except pandas DataFrame and numpy ndarray) to Arrow.
@@ -1003,9 +1018,7 @@ def _to_pylist(column: Union["pa.Array", "pa.ChunkedArray"]) -> List[Any]:
         import pyarrow as pa
         import pyarrow.types as pa_types
 
-        try:
-            import numpy  # noqa: F401
-        except ImportError:
+        if not _is_numpy_available():
             return column.to_pylist()
 
         if isinstance(column, pa.ChunkedArray):

From 2e405eb6ffbb4dbc3e27f7b5a9b7451157b4fe84 Mon Sep 17 00:00:00 2001
From: Liang-Chi Hsieh <viirya@gmail.com>
Date: Thu, 9 Jul 2026 12:14:44 -0700
Subject: [PATCH 6/9] [SPARK-58019][PYTHON] Use native to_pylist when PyArrow
 converts without per-element Scalars

Spark 4.2 ships with this code frozen, so gate the pure-Python bulk
conversion on the installed PyArrow version: releases containing the
apache/arrow#50326 fix (planned for 25.0.1) convert natively without
per-element Scalars and keep improving (apache/arrow#50448), so
_to_pylist short-circuits to column.to_pylist() there and only uses
the bulk paths on older PyArrow. The version constant can be bumped
in a patch release if the fix ships elsewhere. Tests force the gate
off so the bulk paths stay covered on any PyArrow, plus a gate
pass-through test.

Co-authored-by: Isaac
---
 python/pyspark/sql/conversion.py            | 27 ++++++++++++++++++++-
 python/pyspark/sql/tests/test_conversion.py | 20 +++++++++++++++
 2 files changed, 46 insertions(+), 1 deletion(-)

diff --git a/python/pyspark/sql/conversion.py b/python/pyspark/sql/conversion.py
index bd545e3ab2f61..08065373fb075 100644
--- a/python/pyspark/sql/conversion.py
+++ b/python/pyspark/sql/conversion.py
@@ -506,6 +506,28 @@ def convert_column(
         return pa.RecordBatch.from_arrays(arrays, schema.names)
 
 
+# The pure-Python bulk conversion in ArrowTableToRowsConversion._to_pylist is
+# a workaround for PyArrow materializing one Scalar per element (see
+# apache/arrow#50326). PyArrow releases containing the fix convert natively
+# without per-element Scalars, in which case the native conversion is used
+# directly. Bump this constant if the fix ships in a different release.
+_MIN_PYARROW_NATIVE_TO_PYLIST_VERSION = "25.0.1"
+
+_pyarrow_native_to_pylist_is_fast: Optional[bool] = None
+
+
+def _has_fast_native_to_pylist() -> bool:
+    global _pyarrow_native_to_pylist_is_fast
+    if _pyarrow_native_to_pylist_is_fast is None:
+        import pyarrow as pa
+        from pyspark.loose_version import LooseVersion
+
+        _pyarrow_native_to_pylist_is_fast = LooseVersion(pa.__version__) >= LooseVersion(
+            _MIN_PYARROW_NATIVE_TO_PYLIST_VERSION
+        )
+    return _pyarrow_native_to_pylist_is_fast
+
+
 _numpy_available: Optional[bool] = None
 
 
@@ -1018,7 +1040,10 @@ def _to_pylist(column: Union["pa.Array", "pa.ChunkedArray"]) -> List[Any]:
         import pyarrow as pa
         import pyarrow.types as pa_types
 
-        if not _is_numpy_available():
+        if _has_fast_native_to_pylist() or not _is_numpy_available():
+            # Recent PyArrow converts without per-element Scalars natively
+            # (apache/arrow#50326); without NumPy the bulk paths below are
+            # unavailable. Either way, use the native conversion.
             return column.to_pylist()
 
         if isinstance(column, pa.ChunkedArray):
diff --git a/python/pyspark/sql/tests/test_conversion.py b/python/pyspark/sql/tests/test_conversion.py
index b559d3ac3081e..12780a1bd4fa7 100644
--- a/python/pyspark/sql/tests/test_conversion.py
+++ b/python/pyspark/sql/tests/test_conversion.py
@@ -851,6 +851,26 @@ class ArrowColumnToPylistTests(unittest.TestCase):
     column.to_pylist() returns, including exact element types.
     """
 
+    def setUp(self):
+        # Force the bulk paths so they stay covered regardless of the
+        # installed PyArrow version (with a fast native PyArrow the method
+        # short-circuits to column.to_pylist()).
+        import pyspark.sql.conversion as conversion_mod
+
+        self._conversion_mod = conversion_mod
+        self._saved_gate = conversion_mod._pyarrow_native_to_pylist_is_fast
+        conversion_mod._pyarrow_native_to_pylist_is_fast = False
+
+    def tearDown(self):
+        self._conversion_mod._pyarrow_native_to_pylist_is_fast = self._saved_gate
+
+    def test_native_to_pylist_gate(self):
+        import pyarrow as pa
+
+        column = pa.array([[1, None], None], type=pa.list_(pa.int32()))
+        self._conversion_mod._pyarrow_native_to_pylist_is_fast = True
+        self.assertEqual(ArrowTableToRowsConversion._to_pylist(column), [[1, None], None])
+
     def _assert_identical_types(self, actual, expected):
         self.assertIs(type(actual), type(expected))
         if isinstance(actual, (list, tuple)):

From 8e24c2fa2ef5b62d765d763234c384abfe51993f Mon Sep 17 00:00:00 2001
From: Liang-Chi Hsieh <viirya@gmail.com>
Date: Fri, 10 Jul 2026 13:46:41 -0700
Subject: [PATCH 7/9] [SPARK-58019][PYTHON] Address review: use pa.types
 directly, drop redundant annotation, align has_numpy naming

Co-authored-by: Isaac
---
 python/pyspark/sql/conversion.py | 21 ++++++++++++---------
 1 file changed, 12 insertions(+), 9 deletions(-)

diff --git a/python/pyspark/sql/conversion.py b/python/pyspark/sql/conversion.py
index 08065373fb075..5bf898ee14d7f 100644
--- a/python/pyspark/sql/conversion.py
+++ b/python/pyspark/sql/conversion.py
@@ -528,19 +528,23 @@ def _has_fast_native_to_pylist() -> bool:
     return _pyarrow_native_to_pylist_is_fast
 
 
-_numpy_available: Optional[bool] = None
+# None means not yet checked; True/False after the first _is_numpy_available()
+# call. NumPy is only needed indirectly here (pyarrow's Array.to_numpy in
+# _to_pylist), so unlike stateful_processor_api_client.py no np module
+# reference is cached.
+has_numpy: Optional[bool] = None
 
 
 def _is_numpy_available() -> bool:
-    global _numpy_available
-    if _numpy_available is None:
+    global has_numpy
+    if has_numpy is None:
         try:
             import numpy  # noqa: F401
 
-            _numpy_available = True
+            has_numpy = True
         except ImportError:
-            _numpy_available = False
-    return _numpy_available
+            has_numpy = False
+    return has_numpy
 
 
 class LocalDataToArrowConversion:
@@ -1038,7 +1042,6 @@ def _to_pylist(column: Union["pa.Array", "pa.ChunkedArray"]) -> List[Any]:
         for apache/arrow#50326.
         """
         import pyarrow as pa
-        import pyarrow.types as pa_types
 
         if _has_fast_native_to_pylist() or not _is_numpy_available():
             # Recent PyArrow converts without per-element Scalars natively
@@ -1047,13 +1050,13 @@ def _to_pylist(column: Union["pa.Array", "pa.ChunkedArray"]) -> List[Any]:
             return column.to_pylist()
 
         if isinstance(column, pa.ChunkedArray):
-            result: List[Any] = []
+            result = []
             for chunk in column.chunks:
                 result.extend(ArrowTableToRowsConversion._to_pylist(chunk))
             return result
 
         column_type = column.type
-        if (pa_types.is_list(column_type) or pa_types.is_large_list(column_type)) and len(
+        if (pa.types.is_list(column_type) or pa.types.is_large_list(column_type)) and len(
             column
         ) > 0:
             n = len(column)

From 5ce205e5d3a03e0de433ea816ed37849671979b7 Mon Sep 17 00:00:00 2001
From: Liang-Chi Hsieh <viirya@gmail.com>
Date: Fri, 10 Jul 2026 14:10:59 -0700
Subject: [PATCH 8/9] [SPARK-58019][PYTHON] Address review: fold version/numpy
 checks into cached _should_manual_bulk staticmethod

Co-authored-by: Isaac
---
 python/pyspark/sql/conversion.py            | 82 +++++++++------------
 python/pyspark/sql/tests/test_conversion.py | 20 ++---
 2 files changed, 46 insertions(+), 56 deletions(-)

diff --git a/python/pyspark/sql/conversion.py b/python/pyspark/sql/conversion.py
index 5bf898ee14d7f..0bfdb14612689 100644
--- a/python/pyspark/sql/conversion.py
+++ b/python/pyspark/sql/conversion.py
@@ -18,6 +18,7 @@
 import array
 import datetime
 import decimal
+import functools
 from typing import TYPE_CHECKING, Any, Callable, List, Optional, Sequence, Union, overload
 
 import pyspark
@@ -506,47 +507,6 @@ def convert_column(
         return pa.RecordBatch.from_arrays(arrays, schema.names)
 
 
-# The pure-Python bulk conversion in ArrowTableToRowsConversion._to_pylist is
-# a workaround for PyArrow materializing one Scalar per element (see
-# apache/arrow#50326). PyArrow releases containing the fix convert natively
-# without per-element Scalars, in which case the native conversion is used
-# directly. Bump this constant if the fix ships in a different release.
-_MIN_PYARROW_NATIVE_TO_PYLIST_VERSION = "25.0.1"
-
-_pyarrow_native_to_pylist_is_fast: Optional[bool] = None
-
-
-def _has_fast_native_to_pylist() -> bool:
-    global _pyarrow_native_to_pylist_is_fast
-    if _pyarrow_native_to_pylist_is_fast is None:
-        import pyarrow as pa
-        from pyspark.loose_version import LooseVersion
-
-        _pyarrow_native_to_pylist_is_fast = LooseVersion(pa.__version__) >= LooseVersion(
-            _MIN_PYARROW_NATIVE_TO_PYLIST_VERSION
-        )
-    return _pyarrow_native_to_pylist_is_fast
-
-
-# None means not yet checked; True/False after the first _is_numpy_available()
-# call. NumPy is only needed indirectly here (pyarrow's Array.to_numpy in
-# _to_pylist), so unlike stateful_processor_api_client.py no np module
-# reference is cached.
-has_numpy: Optional[bool] = None
-
-
-def _is_numpy_available() -> bool:
-    global has_numpy
-    if has_numpy is None:
-        try:
-            import numpy  # noqa: F401
-
-            has_numpy = True
-        except ImportError:
-            has_numpy = False
-    return has_numpy
-
-
 class LocalDataToArrowConversion:
     """
     Conversion from local data (except pandas DataFrame and numpy ndarray) to Arrow.
@@ -1021,12 +981,42 @@ class ArrowTableToRowsConversion:
     Conversion from Arrow Table to Rows.
     """
 
+    @staticmethod
+    @functools.cache
+    def _should_manual_bulk() -> bool:
+        """
+        Whether ``_to_pylist`` should convert nested columns manually in bulk.
+
+        Internal helper for ``_to_pylist`` only; do not use externally. Returns True
+        when the installed PyArrow still materializes one Scalar per element in
+        ``to_pylist`` (apache/arrow#50326, fix expected in PyArrow 25.0.1 — adjust the
+        version below if it ships in a different release) and NumPy (used for the
+        offsets and validity buffers) is available.
+
+        This method and the manual bulk paths in ``_to_pylist`` should be removed once
+        the minimum supported PyArrow version contains the fix.
+        """
+        import pyarrow as pa
+        from pyspark.loose_version import LooseVersion
+
+        if LooseVersion(pa.__version__) >= LooseVersion("25.0.1"):
+            # Native to_pylist converts without per-element Scalars.
+            return False
+        try:
+            import numpy  # noqa: F401
+        except ImportError:
+            return False
+        return True
+
     @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.
 
+        Internal helper for the worker and ``convert`` call sites; do not use
+        externally.
+
         ``Array.to_pylist()`` materializes one Scalar per element; for list types each row
         additionally allocates a C++ scalar, a Python Scalar wrapper and a Python Array
         wrapper for the row's values before converting elements one by one, which is
@@ -1038,15 +1028,13 @@ def _to_pylist(column: Union["pa.Array", "pa.ChunkedArray"]) -> List[Any]:
         only for the offsets (non-null integers) and the validity bitmap (booleans), so no
         value coercion can occur.
 
-        This can be removed once the minimum supported PyArrow version includes the fix
-        for apache/arrow#50326.
+        This method should be removed (its call sites reverting to plain
+        ``column.to_pylist()``) once the minimum supported PyArrow version includes the
+        fix for apache/arrow#50326.
         """
         import pyarrow as pa
 
-        if _has_fast_native_to_pylist() or not _is_numpy_available():
-            # Recent PyArrow converts without per-element Scalars natively
-            # (apache/arrow#50326); without NumPy the bulk paths below are
-            # unavailable. Either way, use the native conversion.
+        if not ArrowTableToRowsConversion._should_manual_bulk():
             return column.to_pylist()
 
         if isinstance(column, pa.ChunkedArray):
diff --git a/python/pyspark/sql/tests/test_conversion.py b/python/pyspark/sql/tests/test_conversion.py
index 12780a1bd4fa7..7ce2dfe01411c 100644
--- a/python/pyspark/sql/tests/test_conversion.py
+++ b/python/pyspark/sql/tests/test_conversion.py
@@ -16,6 +16,7 @@
 #
 import datetime
 import unittest
+import unittest.mock
 from zoneinfo import ZoneInfo
 
 from pyspark.errors import PySparkRuntimeError, PySparkTypeError, PySparkValueError
@@ -852,24 +853,25 @@ class ArrowColumnToPylistTests(unittest.TestCase):
     """
 
     def setUp(self):
-        # Force the bulk paths so they stay covered regardless of the
+        # Force the manual bulk paths so they stay covered regardless of the
         # installed PyArrow version (with a fast native PyArrow the method
         # short-circuits to column.to_pylist()).
-        import pyspark.sql.conversion as conversion_mod
-
-        self._conversion_mod = conversion_mod
-        self._saved_gate = conversion_mod._pyarrow_native_to_pylist_is_fast
-        conversion_mod._pyarrow_native_to_pylist_is_fast = False
+        self._gate_patcher = unittest.mock.patch.object(
+            ArrowTableToRowsConversion, "_should_manual_bulk", lambda: True
+        )
+        self._gate_patcher.start()
 
     def tearDown(self):
-        self._conversion_mod._pyarrow_native_to_pylist_is_fast = self._saved_gate
+        self._gate_patcher.stop()
 
     def test_native_to_pylist_gate(self):
         import pyarrow as pa
 
         column = pa.array([[1, None], None], type=pa.list_(pa.int32()))
-        self._conversion_mod._pyarrow_native_to_pylist_is_fast = True
-        self.assertEqual(ArrowTableToRowsConversion._to_pylist(column), [[1, None], None])
+        with unittest.mock.patch.object(
+            ArrowTableToRowsConversion, "_should_manual_bulk", lambda: False
+        ):
+            self.assertEqual(ArrowTableToRowsConversion._to_pylist(column), [[1, None], None])
 
     def _assert_identical_types(self, actual, expected):
         self.assertIs(type(actual), type(expected))

From a96c82b83a953084ae16d3f4c8e5620623b8778f Mon Sep 17 00:00:00 2001
From: Liang-Chi Hsieh <viirya@gmail.com>
Date: Fri, 10 Jul 2026 14:17:53 -0700
Subject: [PATCH 9/9] [SPARK-58019][PYTHON] Address review: inline column.type

Co-authored-by: Isaac
---
 python/pyspark/sql/conversion.py | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/python/pyspark/sql/conversion.py b/python/pyspark/sql/conversion.py
index 0bfdb14612689..fddb42c63d8c0 100644
--- a/python/pyspark/sql/conversion.py
+++ b/python/pyspark/sql/conversion.py
@@ -1043,8 +1043,7 @@ def _to_pylist(column: Union["pa.Array", "pa.ChunkedArray"]) -> List[Any]:
                 result.extend(ArrowTableToRowsConversion._to_pylist(chunk))
             return result
 
-        column_type = column.type
-        if (pa.types.is_list(column_type) or pa.types.is_large_list(column_type)) and len(
+        if (pa.types.is_list(column.type) or pa.types.is_large_list(column.type)) and len(
             column
         ) > 0:
             n = len(column)