This is an automated email from the ASF dual-hosted git repository.
Yicong-Huang pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 647c36f61b7a [SPARK-58978][PYTHON] Simplify Arrow collect empty-batch
handling
647c36f61b7a is described below
commit 647c36f61b7a7ae5340d418a2b31700c9a275430
Author: Yicong Huang <[email protected]>
AuthorDate: Tue Aug 25 22:26:12 2026 +0000
[SPARK-58978][PYTHON] Simplify Arrow collect empty-batch handling
### What changes were proposed in this pull request?
This PR simplifies the Arrow collect path in `PandasConversionMixin`
(`python/pyspark/sql/pandas/conversion.py`) by consolidating how empty
(zero-record) results are handled.
- `toArrow()` now handles the empty-result case by building the table from
`schema.empty_table()` (the `schema` it already computes), mirroring what
`toPandas()` already does with `arrow_schema.empty_table()`. Both call sites
now share the same shape: `pa.Table.from_batches(batches)` when there are
records, `schema.empty_table()` otherwise.
- `_collect_as_arrow()` drops its `empty_list_if_zero_records` parameter
and the dead `else` branch that fabricated a single empty `RecordBatch` from a
separately-built schema. The method now always returns the collected list
(empty when there are no records) and ends in a single `return batches`.
- The `prefers_large_var_types` parameter of `_collect_as_arrow()` is
removed as well. It was introduced in SPARK-54300/SPARK-54396 solely to build
the schema for that empty-batch branch; with the branch gone it is dead, so the
two call sites no longer pass it (each still computes its own schema locally).
### Why are the changes needed?
`toArrow()` already computes the full Arrow schema (with
`error_on_duplicated_field_names_in_struct=True`, `timezone="UTC"`, and
large-var-types) at the top of the method, so it can build the empty table
directly, exactly as `toPandas()` already does. The old
`empty_list_if_zero_records=False` path instead had `_collect_as_arrow()`
rebuild a separate schema (without
`error_on_duplicated_field_names_in_struct=True`) only to fabricate a single
empty batch. That flag-less schema was neve [...]
### Does this PR introduce _any_ user-facing change?
No. `toArrow()` and `toPandas()` return the same empty table with the same
schema as before; only the internal code path changes.
### How was this patch tested?
Existing Arrow tests in `python/pyspark/sql/tests/arrow/test_arrow.py`
cover the empty-result paths (`test_toArrow_empty_rows`,
`test_toArrow_empty_columns`, `test_toPandas_empty_*`) and the direct
`_collect_as_arrow` self-destruct call. Extended
`test_toArrow_duplicate_field_names` with a regression guard asserting that an
empty result (`df.limit(0).toArrow()`) still rejects duplicated struct field
names. This locks in existing behavior (the check already ran before the
empty/non-emp [...]
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #58260 from Yicong-Huang/simplify-collect-empty.
Authored-by: Yicong Huang <[email protected]>
Signed-off-by: Yicong-Huang <[email protected]>
(cherry picked from commit d45be0f202e31d99242e2b9ee83b090c1ad8625a)
Signed-off-by: Yicong-Huang <[email protected]>
---
python/pyspark/sql/pandas/conversion.py | 38 ++++++++--------------------
python/pyspark/sql/tests/arrow/test_arrow.py | 6 +++++
2 files changed, 16 insertions(+), 28 deletions(-)
diff --git a/python/pyspark/sql/pandas/conversion.py
b/python/pyspark/sql/pandas/conversion.py
index e5d89f90ea3c..3bdaa67b91e6 100644
--- a/python/pyspark/sql/pandas/conversion.py
+++ b/python/pyspark/sql/pandas/conversion.py
@@ -394,7 +394,6 @@ class PandasConversionMixin:
batches = self._collect_as_arrow(
split_batches=arrowPySparkSelfDestructEnabled ==
"true",
- prefers_large_var_types=prefers_large_var_types,
)
if len(batches) > 0:
@@ -493,12 +492,12 @@ class PandasConversionMixin:
import pyarrow as pa
self_destruct = arrowPySparkSelfDestructEnabled == "true"
- batches = self._collect_as_arrow(
- split_batches=self_destruct,
- empty_list_if_zero_records=False,
- prefers_large_var_types=prefers_large_var_types,
- )
- table = pa.Table.from_batches(batches).cast(schema)
+ batches = self._collect_as_arrow(split_batches=self_destruct)
+ if batches:
+ table = pa.Table.from_batches(batches).cast(schema)
+ else:
+ # empty dataset
+ table = schema.empty_table()
# Ensure only the table has a reference to the batches, so that
# self_destruct (if enabled) is effective
del batches
@@ -507,21 +506,15 @@ class PandasConversionMixin:
def _collect_as_arrow(
self,
split_batches: bool = False,
- empty_list_if_zero_records: bool = True,
- prefers_large_var_types: bool = False,
) -> List["pa.RecordBatch"]:
"""
- Returns all records as a list of Arrow RecordBatches. PyArrow must be
installed
- and available on driver and worker Python environments.
- This is an experimental feature.
+ Returns all records as a list of Arrow RecordBatches, which is empty
if the result has
+ 0 records. PyArrow must be installed and available on driver and
worker Python
+ environments. This is an experimental feature.
:param split_batches: split batches such that each column is in its
own allocation, so
that the selfDestruct optimization is effective; default False.
- :param empty_list_if_zero_records: If True (the default), returns an
empty list if the
- result has 0 records. Otherwise, returns a list of length 1
containing an empty
- Arrow RecordBatch which includes the schema.
-
.. note:: Experimental.
"""
from pyspark.sql.dataframe import DataFrame
@@ -562,18 +555,7 @@ class PandasConversionMixin:
# Join serving thread and raise any exceptions from
collectAsArrowToPython
jsocket_auth_server.getResult()
- if len(batches) or empty_list_if_zero_records:
- return batches
- else:
- import pyarrow as pa
-
- from pyspark.sql.pandas.types import to_arrow_schema
-
- schema = to_arrow_schema(
- self.schema, timezone="UTC",
prefers_large_types=prefers_large_var_types
- )
- empty_arrays = [pa.array([], type=field.type) for field in schema]
- return [pa.RecordBatch.from_arrays(empty_arrays, schema=schema)]
+ return batches
class SparkConversionMixin:
diff --git a/python/pyspark/sql/tests/arrow/test_arrow.py
b/python/pyspark/sql/tests/arrow/test_arrow.py
index 7637efc2218e..df4239651d34 100644
--- a/python/pyspark/sql/tests/arrow/test_arrow.py
+++ b/python/pyspark/sql/tests/arrow/test_arrow.py
@@ -1427,6 +1427,12 @@ class ArrowTestsMixin:
):
df.toArrow()
+ # An empty result must reject duplicated struct field names just like
a non-empty one.
+ with self.assertRaisesRegex(
+ UnsupportedOperationException,
"DUPLICATED_FIELD_NAME_IN_ARROW_STRUCT"
+ ):
+ df.limit(0).toArrow()
+
def test_createDataFrame_pandas_duplicate_field_names(self):
for arrow_enabled in [True, False]:
with self.subTest(arrow_enabled=arrow_enabled):
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
