[SPARK-58023][PYTHON] Fast paths for converting Arrow string, binary, numeric and boolean columns to Python rows

Extend ArrowTableToRowsConversion._to_pylist with leaf fast paths:

* string/large_string/binary/large_binary columns use Arrow's
  object-dtype NumPy conversion, which produces exactly str/bytes and
  None - no other type can come out of it.
* integer/float32/float64/boolean columns without nulls convert via a
  zero-copy (except bit-packed booleans) NumPy view and ndarray.tolist,
  which materializes exact Python ints/floats/bools. With nulls, the
  values are filled with a placeholder first (pc.fill_null) and nulls
  are restored to None from the validity bitmap, so ints are always
  materialized from the original int buffer, never through a float
  representation.

Types whose as_py returns non-primitive objects (dates, timestamps,
decimals, ...) keep using to_pylist. Since list columns convert their
flattened child values through _to_pylist, list-typed columns get the
leaf speedup on top of the bulk offsets slicing.

ASV microbenchmark (bench_arrow.ArrowLeafColumnToRowsBenchmark, 1M
rows): string with nulls 196ms -> 20ms (9.7x); int64 with nulls
99ms -> 28ms (3.6x); float64 without nulls 100ms -> 9ms (11x).
End-to-end list<string> conversion (ArrowListColumnToRowsBenchmark, 1M rows) improves from 507ms to 118ms on
top of SPARK-58019.

Co-authored-by: Isaac

Author: viirya

python/benchmarks/bench_arrow.py

@@ -154,3 +154,46 @@ def time_nested_ints_with_nulls_to_rows(self, n_rows, method):
 
     def peakmem_list_of_strings_to_rows(self, n_rows, method):
         self.convert(self.list_of_strings)
+
+
+class ArrowLeafColumnToRowsBenchmark:
+    """
+    Benchmark for converting flat (leaf) Arrow columns to Python rows.
+
+    ``baseline`` measures plain ``column.to_pylist()``; ``bulk`` measures
+    ``ArrowTableToRowsConversion._to_pylist`` with the string/binary/numeric
+    fast paths.
+    """
+
+    params = [
+        [100000, 1000000],
+        ["baseline", "bulk"],
+    ]
+    param_names = ["n_rows", "method"]
+
+    def setup(self, n_rows, method):
+        from pyspark.sql.conversion import ArrowTableToRowsConversion
+
+        self.strings = pa.array(
+            [f"s{i}" if i % 10 != 0 else None for i in range(n_rows)], type=pa.string()
+        )
+        self.longs_with_nulls = pa.array(
+            [i if i % 10 != 0 else None for i in range(n_rows)], type=pa.int64()
+        )
+        self.doubles = pa.array([float(i) for i in range(n_rows)], type=pa.float64())
+        if method == "bulk":
+            self.convert = ArrowTableToRowsConversion._to_pylist
+        else:
+            self.convert = lambda column: column.to_pylist()
+
+    def time_strings_with_nulls_to_rows(self, n_rows, method):
+        self.convert(self.strings)
+
+    def time_longs_with_nulls_to_rows(self, n_rows, method):
+        self.convert(self.longs_with_nulls)
+
+    def time_doubles_to_rows(self, n_rows, method):
+        self.convert(self.doubles)
+
+    def peakmem_strings_with_nulls_to_rows(self, n_rows, method):
+        self.convert(self.strings)

python/pyspark/sql/conversion.py

@@ -984,18 +984,23 @@ class ArrowTableToRowsConversion:
     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.
+        instead of one scalar at a time, with fast paths for string, binary, integral,
+        floating point and boolean leaves.
 
         ``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.
+        and slicing the resulting Python list per row (see apache/arrow#50326). Results
+        are exactly identical to ``to_pylist``: ``None`` stays ``None`` and values inside
+        numeric lists stay Python ints, unlike a pandas round trip which would coerce
+        them to floats/NaN. In particular the leaf fast paths cannot coerce: string and
+        binary columns use Arrow's object-dtype conversion, which only produces ``str`` /
+        ``bytes`` / ``None``; nullable numeric and boolean columns are converted from the
+        original values (nulls filled with a placeholder and restored to ``None`` from the
+        validity bitmap afterwards), so ints are materialized from the int buffer, never
+        via a float representation. Types whose ``as_py`` returns non-primitive objects
+        (dates, timestamps, decimals, ...) keep using ``to_pylist``.
 
         This can be removed once the minimum supported PyArrow version includes the fix
         for apache/arrow#50326.
@@ -1014,10 +1019,38 @@ def _to_pylist(column: Union["pa.Array", "pa.ChunkedArray"]) -> List[Any]:
                 result.extend(ArrowTableToRowsConversion._to_pylist(chunk))
             return result
 
+        if len(column) == 0:
+            return []
+
         column_type = column.type
-        if (pa_types.is_list(column_type) or pa_types.is_large_list(column_type)) and len(
-            column
-        ) > 0:
+
+        if (
+            pa_types.is_string(column_type)
+            or pa_types.is_large_string(column_type)
+            or pa_types.is_binary(column_type)
+            or pa_types.is_large_binary(column_type)
+        ):
+            # The object-dtype conversion produces exactly str/bytes and None.
+            return column.to_numpy(zero_copy_only=False).tolist()
+
+        if (
+            pa_types.is_integer(column_type)
+            or pa_types.is_float32(column_type)
+            or pa_types.is_float64(column_type)
+            or pa_types.is_boolean(column_type)
+        ):
+            # Booleans are bit-packed, so their conversion to NumPy is never zero-copy.
+            zero_copy = not pa_types.is_boolean(column_type)
+            if column.null_count == 0:
+                return column.to_numpy(zero_copy_only=zero_copy).tolist()
+            import pyarrow.compute as pc
+
+            valid = column.is_valid().to_numpy(zero_copy_only=False).tolist()
+            fill_value = False if pa_types.is_boolean(column_type) else 0
+            values = pc.fill_null(column, fill_value).to_numpy(zero_copy_only=zero_copy).tolist()
+            return [v if m else None for v, m in zip(values, valid)]
+
+        if 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

python/pyspark/sql/tests/test_conversion.py

@@ -15,6 +15,7 @@
 # limitations under the License.
 #
 import datetime
+import decimal
 import unittest
 from zoneinfo import ZoneInfo
 
@@ -876,6 +877,27 @@ def test_matches_to_pylist(self):
             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)),
+            # leaf fast paths: exact str/bytes/int/float/bool types, None for nulls
+            pa.array(["", None, "日本語", "\N{GRINNING FACE}", "x" * 40], type=pa.string()),
+            pa.array(["a", None, ""], type=pa.large_string()),
+            pa.array([b"", None, b"\x00\xff"], type=pa.binary()),
+            pa.array([b"a", None], type=pa.large_binary()),
+            pa.array([1, None, -(2**62), 3], type=pa.int64()),
+            pa.array([0, None, 2**63 + 7], type=pa.uint64()),
+            pa.array([-128, 127, None], type=pa.int8()),
+            pa.array([1.5, None, float("nan"), float("inf")], type=pa.float64()),
+            pa.array([1.5, None], type=pa.float32()),
+            pa.array([True, None, False], type=pa.bool_()),
+            pa.array([True, False] * 5, type=pa.bool_()),
+            pa.array(list(range(10)), type=pa.int32()),
+            # non-primitive leaves must keep as_py semantics (fallback path)
+            pa.array([datetime.date(2020, 1, 2), None], type=pa.date32()),
+            pa.array([datetime.datetime(2020, 1, 2, 3, 4, 5), None], type=pa.timestamp("us")),
+            pa.array([decimal.Decimal("1.23"), None], type=pa.decimal128(10, 2)),
+            # lists of fast-path leaves
+            pa.array([[b"x", None], None, [b""]], type=pa.list_(pa.binary())),
+            pa.array([[True, None], [False]], type=pa.list_(pa.bool_())),
+            pa.array([[1.5, None], None], type=pa.list_(pa.float32())),
         ]
         for column in columns:
             views = [column, column.slice(1), column.slice(0, max(len(column) - 1, 0))]