viirya commented on code in PR #57099:
URL: https://github.com/apache/spark/pull/57099#discussion_r3561992123
##########
python/pyspark/sql/conversion.py:
##########
@@ -980,6 +1017,64 @@ 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
+
+ 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):
+ result: List[Any] = []
+ for chunk in column.chunks:
+ result.extend(ArrowTableToRowsConversion._to_pylist(chunk))
+ return result
+
+ column_type = column.type
Review Comment:
It is actually needed: `ArrowTableToRowsConversion.convert` passes
`table.columns`, and `pa.Table.columns` are `ChunkedArray`s (the Spark Connect
collect path). Without this branch a chunked list column falls into the list
path and fails with `AttributeError: 'pyarrow.lib.ChunkedArray' object has no
attribute 'offsets'`. The worker.py call sites pass `RecordBatch` columns
(plain `Array`s), so only the `convert` path exercises it; the chunked variants
in `ArrowColumnToPylistTests` cover it.
##########
python/pyspark/sql/conversion.py:
##########
@@ -980,6 +1017,64 @@ 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
Review Comment:
Done in 8e24c2fa2ef — dropped the alias and use `pa.types.is_list` /
`pa.types.is_large_list` directly. Agreed that `pa_types` reads confusingly
close to the `pa_type` variables in this file.
##########
python/pyspark/sql/conversion.py:
##########
@@ -980,6 +1017,64 @@ 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
+
+ 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):
+ result: List[Any] = []
Review Comment:
You're right — mypy infers `list[Any]` from the `extend` in the same scope.
Removed in 8e24c2fa2ef and verified with the pinned mypy 1.19.1.
##########
python/pyspark/sql/conversion.py:
##########
@@ -506,6 +506,43 @@ 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
Review Comment:
Renamed the cached flag to `has_numpy` in 8e24c2fa2ef to align with #57163.
I didn't cache an `np` module global though: unlike
`stateful_processor_api_client.py`, this file never references the numpy module
directly — numpy is only needed because `pyarrow.Array.to_numpy()` requires it
— so `np` would have no user here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
