You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "Spenserrrr (via GitHub)" <gi...@apache.org> on 2026/09/11 21:46:40 UTC
[PR] [SPARK-54966][PYTHON] Factor out legacy pandas-to-Arrow column conversion [spark]
Spenserrrr opened a new pull request, #58751:
URL: https://github.com/apache/spark/pull/58751
### What changes were proposed in this pull request?
Separate pandas-to-Arrow batch handling into `from_pandas()` and route each Series through `convert()` to the extracted `convert_legacy()` implementation. Rename `is_legacy` to `use_legacy_error_handling` and update callers.
### Why are the changes needed?
Separate batch handling from column conversion to prepare for additional pandas conversion strategies.
### Does this PR introduce _any_ user-facing change?
No; this is an internal refactor.
### How was this patch tested?
Existing conversion tests with pandas 2.3/PyArrow 24 and pandas 3/PyArrow 25, plus baseline-parity and error-handling probes.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex (GPT-6)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54966][PYTHON] Factor out legacy pandas-to-Arrow column conversion [spark]
Posted by "gaogaotiantian (via GitHub)" <gi...@apache.org>.
gaogaotiantian closed pull request #58751: [SPARK-54966][PYTHON] Factor out legacy pandas-to-Arrow column conversion
URL: https://github.com/apache/spark/pull/58751
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54966][PYTHON] Factor out legacy pandas-to-Arrow column conversion [spark]
Posted by "Spenserrrr (via GitHub)" <gi...@apache.org>.
Spenserrrr commented on PR #58751:
URL: https://github.com/apache/spark/pull/58751#issuecomment-5670779494
Thanks, @gaogaotiantian! from_pandas() handles batches, and I’ve renamed the internal helpers to _convert_column() and _convert_column_legacy() to make their scope explicit. The dispatcher currently calls the legacy implementation, and the planned NumPy and PyArrow strategies will be added later once this PR is merged. I’ll revise the read leg to follow the same naming convention in a new PR.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54966][PYTHON] Factor out legacy pandas-to-Arrow column conversion [spark]
Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang commented on code in PR #58751:
URL: https://github.com/apache/spark/pull/58751#discussion_r4031070937
##########
python/pyspark/sql/conversion.py:
##########
@@ -410,105 +401,161 @@ def convert_column(
assign_cols_by_name=assign_cols_by_name,
int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
- is_legacy=is_legacy,
+ use_legacy_error_handling=use_legacy_error_handling,
)
# Wrap the nested RecordBatch as a single StructArray column
- return ArrowBatchTransformer.wrap_struct(nested_batch).column(0)
+ converted.append(ArrowBatchTransformer.wrap_struct(nested_batch).column(0))
+ else:
+ assert isinstance(col, pd.Series)
+ converted.append(
+ cls._convert_column(
+ col,
+ field,
+ timezone=timezone,
+ safecheck=safecheck,
+ arrow_cast=arrow_cast,
+ prefers_large_types=prefers_large_types,
+ int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
+ ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
+ use_legacy_error_handling=use_legacy_error_handling,
+ )
+ )
- series = col
- field_name = field.name
- ret_type = field.dataType
+ # pa.Array.from_pandas returns a pa.ChunkedArray for a chunked arrow-backed Series
+ # (e.g. a pyarrow-backed extension dtype), which pa.RecordBatch.from_arrays rejects.
+ arrays = [a.combine_chunks() if isinstance(a, pa.ChunkedArray) else a for a in converted]
+ return pa.RecordBatch.from_arrays(arrays, schema.names)
- if isinstance(series.dtype, pd.CategoricalDtype):
- series = series.astype(series.dtype.categories.dtype)
+ @classmethod
+ def _convert_column(
+ cls,
+ series: "pd.Series",
+ field: StructField,
+ *,
+ timezone: Optional[str] = None,
+ safecheck: bool = True,
+ arrow_cast: bool = False,
+ prefers_large_types: bool = False,
+ int_to_decimal_coercion_enabled: bool = False,
+ ignore_unexpected_complex_type_values: bool = False,
+ use_legacy_error_handling: bool = False,
+ ) -> Union["pa.Array", "pa.ChunkedArray"]:
+ """Dispatch a pandas column to its conversion strategy."""
+ # Future NumPy and PyArrow strategies will be selected here; use legacy for now.
+ return cls._convert_column_legacy(
+ series,
+ field,
+ timezone=timezone,
+ safecheck=safecheck,
+ arrow_cast=arrow_cast,
+ prefers_large_types=prefers_large_types,
+ int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
+ ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
+ use_legacy_error_handling=use_legacy_error_handling,
+ )
- arrow_type = to_arrow_type(
- ret_type, timezone=timezone, prefers_large_types=prefers_large_types
- )
- series = _create_converter_from_pandas(
- ret_type,
- timezone=timezone,
- error_on_duplicated_field_names=False,
- int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
- ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
- )(series)
+ @classmethod
+ def _convert_column_legacy(
+ cls,
+ series: "pd.Series",
+ field: StructField,
+ *,
+ timezone: Optional[str] = None,
+ safecheck: bool = True,
+ arrow_cast: bool = False,
+ prefers_large_types: bool = False,
+ int_to_decimal_coercion_enabled: bool = False,
+ ignore_unexpected_complex_type_values: bool = False,
+ use_legacy_error_handling: bool = False,
+ ) -> Union["pa.Array", "pa.ChunkedArray"]:
+ """Convert a pandas column using the legacy conversion strategy."""
+ import pandas as pd
+ import pyarrow as pa
+
+ from pyspark.errors import PySparkTypeError, PySparkValueError
+ from pyspark.sql.pandas.types import _create_converter_from_pandas, to_arrow_type
+
+ field_name = field.name
Review Comment:
Optional, non-blocking (follow-up): the old `convert_column` docstring explained *why* it uses `field.name` instead of `series.name` here — avoiding a `Series.rename()` copy, ~20% on the hot path. That rationale was dropped in the refactor while `field.name` is still used. Consider re-adding a short comment so the perf reason isn't lost for future maintainers.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54966][PYTHON] Factor out legacy pandas-to-Arrow column conversion [spark]
Posted by "gaogaotiantian (via GitHub)" <gi...@apache.org>.
gaogaotiantian commented on PR #58751:
URL: https://github.com/apache/spark/pull/58751#issuecomment-5704724923
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/8e1f4f755337339a81602fccd644df87fca37cf5
- merged into branch-4.x https://github.com/apache/spark/commit/6093a61f84da1fc7b2c61ed68142013fa717d776
*Posted by `merge_spark_pr.py`*
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54966][PYTHON] Factor out legacy pandas-to-Arrow column conversion [spark]
Posted by "gaogaotiantian (via GitHub)" <gi...@apache.org>.
gaogaotiantian commented on code in PR #58751:
URL: https://github.com/apache/spark/pull/58751#discussion_r4010613916
##########
python/pyspark/sql/conversion.py:
##########
@@ -410,105 +402,168 @@ def convert_column(
assign_cols_by_name=assign_cols_by_name,
int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
- is_legacy=is_legacy,
+ use_legacy_error_handling=use_legacy_error_handling,
)
# Wrap the nested RecordBatch as a single StructArray column
- return ArrowBatchTransformer.wrap_struct(nested_batch).column(0)
+ converted.append(ArrowBatchTransformer.wrap_struct(nested_batch).column(0))
+ elif isinstance(col, pd.Series):
+ converted.append(
+ cls._convert_column(
+ col,
+ field,
+ timezone=timezone,
+ safecheck=safecheck,
+ arrow_cast=arrow_cast,
+ prefers_large_types=prefers_large_types,
+ int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
+ ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
+ use_legacy_error_handling=use_legacy_error_handling,
+ )
+ )
+ else:
+ raise PySparkTypeError(
+ errorClass="UDF_RETURN_TYPE",
Review Comment:
I don't think `UDF_RETURN_TYPE` is a proper `errorClass` here. First of all, it's not return type. Also this code could be called outside of UDFs. Maybe pick another error class or even use error message directly. A bad error class is worse than no error class.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54966][PYTHON] Factor out legacy pandas-to-Arrow column conversion [spark]
Posted by "gaogaotiantian (via GitHub)" <gi...@apache.org>.
gaogaotiantian commented on code in PR #58751:
URL: https://github.com/apache/spark/pull/58751#discussion_r4009755526
##########
python/pyspark/sql/conversion.py:
##########
@@ -410,105 +401,159 @@ def convert_column(
assign_cols_by_name=assign_cols_by_name,
int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
- is_legacy=is_legacy,
+ use_legacy_error_handling=use_legacy_error_handling,
)
# Wrap the nested RecordBatch as a single StructArray column
- return ArrowBatchTransformer.wrap_struct(nested_batch).column(0)
+ converted.append(ArrowBatchTransformer.wrap_struct(nested_batch).column(0))
+ else:
Review Comment:
Let's either add an `assert` here for the type of `col`, or do `elif` check for the type and raise an exception in `else`. Depending on whether we believe it's an internal bug or a user input error.
##########
python/pyspark/sql/conversion.py:
##########
@@ -410,105 +401,159 @@ def convert_column(
assign_cols_by_name=assign_cols_by_name,
int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
- is_legacy=is_legacy,
+ use_legacy_error_handling=use_legacy_error_handling,
)
# Wrap the nested RecordBatch as a single StructArray column
- return ArrowBatchTransformer.wrap_struct(nested_batch).column(0)
+ converted.append(ArrowBatchTransformer.wrap_struct(nested_batch).column(0))
+ else:
+ converted.append(
+ cls._convert_column(
+ col,
+ field,
+ timezone=timezone,
+ safecheck=safecheck,
+ arrow_cast=arrow_cast,
+ prefers_large_types=prefers_large_types,
+ int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
+ ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
+ use_legacy_error_handling=use_legacy_error_handling,
+ )
+ )
- series = col
- field_name = field.name
- ret_type = field.dataType
+ # pa.Array.from_pandas returns a pa.ChunkedArray for a chunked arrow-backed Series
+ # (e.g. a pyarrow-backed extension dtype), which pa.RecordBatch.from_arrays rejects.
+ arrays = [a.combine_chunks() if isinstance(a, pa.ChunkedArray) else a for a in converted]
+ return pa.RecordBatch.from_arrays(arrays, schema.names)
- if isinstance(series.dtype, pd.CategoricalDtype):
- series = series.astype(series.dtype.categories.dtype)
+ @classmethod
+ def _convert_column(
+ cls,
+ series: "pd.Series",
+ field: StructField,
+ *,
+ timezone: Optional[str] = None,
+ safecheck: bool = True,
+ arrow_cast: bool = False,
+ prefers_large_types: bool = False,
+ int_to_decimal_coercion_enabled: bool = False,
+ ignore_unexpected_complex_type_values: bool = False,
+ use_legacy_error_handling: bool = False,
+ ) -> Union["pa.Array", "pa.ChunkedArray"]:
+ """Dispatch a pandas column to its conversion strategy."""
Review Comment:
Let's add some comment here about why this just calls into legacy. We have future plans to change it right?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54966][PYTHON] Factor out legacy pandas-to-Arrow column conversion [spark]
Posted by "Spenserrrr (via GitHub)" <gi...@apache.org>.
Spenserrrr commented on PR #58751:
URL: https://github.com/apache/spark/pull/58751#issuecomment-5641155800
Hi @Yicong-Huang @zhengruifeng! This PR separates pandas-to-Arrow batch processing from column conversion. Right now, from_pandas() handles batch orchestration, and convert() selects the per-column strategy. This prepares the column layer for subsequent convert_numpy and convert_pyarrow implementations. Could you take a look when you have time? Thanks!
cc @gaogaotiantian
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54966][PYTHON] Factor out legacy pandas-to-Arrow column conversion [spark]
Posted by "Spenserrrr (via GitHub)" <gi...@apache.org>.
Spenserrrr commented on code in PR #58751:
URL: https://github.com/apache/spark/pull/58751#discussion_r4011617138
##########
python/pyspark/sql/conversion.py:
##########
@@ -410,105 +402,168 @@ def convert_column(
assign_cols_by_name=assign_cols_by_name,
int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
- is_legacy=is_legacy,
+ use_legacy_error_handling=use_legacy_error_handling,
)
# Wrap the nested RecordBatch as a single StructArray column
- return ArrowBatchTransformer.wrap_struct(nested_batch).column(0)
+ converted.append(ArrowBatchTransformer.wrap_struct(nested_batch).column(0))
+ elif isinstance(col, pd.Series):
+ converted.append(
+ cls._convert_column(
+ col,
+ field,
+ timezone=timezone,
+ safecheck=safecheck,
+ arrow_cast=arrow_cast,
+ prefers_large_types=prefers_large_types,
+ int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
+ ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
+ use_legacy_error_handling=use_legacy_error_handling,
+ )
+ )
+ else:
+ raise PySparkTypeError(
+ errorClass="UDF_RETURN_TYPE",
Review Comment:
Agreed. I switched this check to an assertion for now and I will create a new PR for the input-validation and error-class cleanup.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54966][PYTHON] Factor out legacy pandas-to-Arrow column conversion [spark]
Posted by "Spenserrrr (via GitHub)" <gi...@apache.org>.
Spenserrrr commented on code in PR #58751:
URL: https://github.com/apache/spark/pull/58751#discussion_r4010745746
##########
python/pyspark/sql/conversion.py:
##########
@@ -410,105 +402,168 @@ def convert_column(
assign_cols_by_name=assign_cols_by_name,
int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
- is_legacy=is_legacy,
+ use_legacy_error_handling=use_legacy_error_handling,
)
# Wrap the nested RecordBatch as a single StructArray column
- return ArrowBatchTransformer.wrap_struct(nested_batch).column(0)
+ converted.append(ArrowBatchTransformer.wrap_struct(nested_batch).column(0))
+ elif isinstance(col, pd.Series):
+ converted.append(
+ cls._convert_column(
+ col,
+ field,
+ timezone=timezone,
+ safecheck=safecheck,
+ arrow_cast=arrow_cast,
+ prefers_large_types=prefers_large_types,
+ int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
+ ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
+ use_legacy_error_handling=use_legacy_error_handling,
+ )
+ )
+ else:
+ raise PySparkTypeError(
+ errorClass="UDF_RETURN_TYPE",
Review Comment:
Thanks for the suggestion! I switched to the existing generic NOT_EXPECTED_TYPE error class, since this validates converter inputs and shouldn’t assume a UDF-specific call context.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54966][PYTHON] Factor out legacy pandas-to-Arrow column conversion [spark]
Posted by "Spenserrrr (via GitHub)" <gi...@apache.org>.
Spenserrrr commented on code in PR #58751:
URL: https://github.com/apache/spark/pull/58751#discussion_r4010236982
##########
python/pyspark/sql/conversion.py:
##########
@@ -410,105 +401,159 @@ def convert_column(
assign_cols_by_name=assign_cols_by_name,
int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
- is_legacy=is_legacy,
+ use_legacy_error_handling=use_legacy_error_handling,
)
# Wrap the nested RecordBatch as a single StructArray column
- return ArrowBatchTransformer.wrap_struct(nested_batch).column(0)
+ converted.append(ArrowBatchTransformer.wrap_struct(nested_batch).column(0))
+ else:
+ converted.append(
+ cls._convert_column(
+ col,
+ field,
+ timezone=timezone,
+ safecheck=safecheck,
+ arrow_cast=arrow_cast,
+ prefers_large_types=prefers_large_types,
+ int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
+ ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
+ use_legacy_error_handling=use_legacy_error_handling,
+ )
+ )
- series = col
- field_name = field.name
- ret_type = field.dataType
+ # pa.Array.from_pandas returns a pa.ChunkedArray for a chunked arrow-backed Series
+ # (e.g. a pyarrow-backed extension dtype), which pa.RecordBatch.from_arrays rejects.
+ arrays = [a.combine_chunks() if isinstance(a, pa.ChunkedArray) else a for a in converted]
+ return pa.RecordBatch.from_arrays(arrays, schema.names)
- if isinstance(series.dtype, pd.CategoricalDtype):
- series = series.astype(series.dtype.categories.dtype)
+ @classmethod
+ def _convert_column(
+ cls,
+ series: "pd.Series",
+ field: StructField,
+ *,
+ timezone: Optional[str] = None,
+ safecheck: bool = True,
+ arrow_cast: bool = False,
+ prefers_large_types: bool = False,
+ int_to_decimal_coercion_enabled: bool = False,
+ ignore_unexpected_complex_type_values: bool = False,
+ use_legacy_error_handling: bool = False,
+ ) -> Union["pa.Array", "pa.ChunkedArray"]:
+ """Dispatch a pandas column to its conversion strategy."""
Review Comment:
Added a comment clarifying why this currently delegates only to the legacy implementation. Planned follow-ups will migrate applicable types to _convert_column_numpy, following the read leg, and add _convert_column_pyarrow for Arrow-backed pandas Series.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54966][PYTHON] Factor out legacy pandas-to-Arrow column conversion [spark]
Posted by "Spenserrrr (via GitHub)" <gi...@apache.org>.
Spenserrrr commented on code in PR #58751:
URL: https://github.com/apache/spark/pull/58751#discussion_r4010234767
##########
python/pyspark/sql/conversion.py:
##########
@@ -410,105 +401,159 @@ def convert_column(
assign_cols_by_name=assign_cols_by_name,
int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
- is_legacy=is_legacy,
+ use_legacy_error_handling=use_legacy_error_handling,
)
# Wrap the nested RecordBatch as a single StructArray column
- return ArrowBatchTransformer.wrap_struct(nested_batch).column(0)
+ converted.append(ArrowBatchTransformer.wrap_struct(nested_batch).column(0))
+ else:
Review Comment:
Added an explicit type check with PySparkTypeError (UDF_RETURN_TYPE). Some scalar pandas UDF paths currently check the result’s length without requiring a Series, so an invalid return such as a list can reach here. I chose an exception rather than an assertion because this can be invalid user input, not just an internal programming error.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54966][PYTHON] Factor out legacy pandas-to-Arrow column conversion [spark]
Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang commented on code in PR #58751:
URL: https://github.com/apache/spark/pull/58751#discussion_r4010967174
##########
python/pyspark/sql/conversion.py:
##########
@@ -410,105 +402,168 @@ def convert_column(
assign_cols_by_name=assign_cols_by_name,
int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
- is_legacy=is_legacy,
+ use_legacy_error_handling=use_legacy_error_handling,
)
# Wrap the nested RecordBatch as a single StructArray column
- return ArrowBatchTransformer.wrap_struct(nested_batch).column(0)
+ converted.append(ArrowBatchTransformer.wrap_struct(nested_batch).column(0))
+ elif isinstance(col, pd.Series):
+ converted.append(
+ cls._convert_column(
+ col,
+ field,
+ timezone=timezone,
+ safecheck=safecheck,
+ arrow_cast=arrow_cast,
+ prefers_large_types=prefers_large_types,
+ int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
+ ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
+ use_legacy_error_handling=use_legacy_error_handling,
+ )
+ )
+ else:
+ raise PySparkTypeError(
+ errorClass="UDF_RETURN_TYPE",
Review Comment:
This is kind of a historical issue in PySpark, many error classes are not picked/designed carefully. I also wanted to revise the error class, that to be said, I’d suggest doing that separately in a new PR if possible, to maintain no behavioral change for this PR.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54966][PYTHON] Factor out legacy pandas-to-Arrow column conversion [spark]
Posted by "Spenserrrr (via GitHub)" <gi...@apache.org>.
Spenserrrr commented on code in PR #58751:
URL: https://github.com/apache/spark/pull/58751#discussion_r4031107934
##########
python/pyspark/sql/conversion.py:
##########
@@ -410,105 +401,161 @@ def convert_column(
assign_cols_by_name=assign_cols_by_name,
int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
- is_legacy=is_legacy,
+ use_legacy_error_handling=use_legacy_error_handling,
)
# Wrap the nested RecordBatch as a single StructArray column
- return ArrowBatchTransformer.wrap_struct(nested_batch).column(0)
+ converted.append(ArrowBatchTransformer.wrap_struct(nested_batch).column(0))
+ else:
+ assert isinstance(col, pd.Series)
+ converted.append(
+ cls._convert_column(
+ col,
+ field,
+ timezone=timezone,
+ safecheck=safecheck,
+ arrow_cast=arrow_cast,
+ prefers_large_types=prefers_large_types,
+ int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
+ ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
+ use_legacy_error_handling=use_legacy_error_handling,
+ )
+ )
- series = col
- field_name = field.name
- ret_type = field.dataType
+ # pa.Array.from_pandas returns a pa.ChunkedArray for a chunked arrow-backed Series
+ # (e.g. a pyarrow-backed extension dtype), which pa.RecordBatch.from_arrays rejects.
+ arrays = [a.combine_chunks() if isinstance(a, pa.ChunkedArray) else a for a in converted]
+ return pa.RecordBatch.from_arrays(arrays, schema.names)
- if isinstance(series.dtype, pd.CategoricalDtype):
- series = series.astype(series.dtype.categories.dtype)
+ @classmethod
+ def _convert_column(
+ cls,
+ series: "pd.Series",
+ field: StructField,
+ *,
+ timezone: Optional[str] = None,
+ safecheck: bool = True,
+ arrow_cast: bool = False,
+ prefers_large_types: bool = False,
+ int_to_decimal_coercion_enabled: bool = False,
+ ignore_unexpected_complex_type_values: bool = False,
+ use_legacy_error_handling: bool = False,
+ ) -> Union["pa.Array", "pa.ChunkedArray"]:
+ """Dispatch a pandas column to its conversion strategy."""
+ # Future NumPy and PyArrow strategies will be selected here; use legacy for now.
+ return cls._convert_column_legacy(
+ series,
+ field,
+ timezone=timezone,
+ safecheck=safecheck,
+ arrow_cast=arrow_cast,
+ prefers_large_types=prefers_large_types,
+ int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
+ ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
+ use_legacy_error_handling=use_legacy_error_handling,
+ )
- arrow_type = to_arrow_type(
- ret_type, timezone=timezone, prefers_large_types=prefers_large_types
- )
- series = _create_converter_from_pandas(
- ret_type,
- timezone=timezone,
- error_on_duplicated_field_names=False,
- int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
- ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
- )(series)
+ @classmethod
+ def _convert_column_legacy(
+ cls,
+ series: "pd.Series",
+ field: StructField,
+ *,
+ timezone: Optional[str] = None,
+ safecheck: bool = True,
+ arrow_cast: bool = False,
+ prefers_large_types: bool = False,
+ int_to_decimal_coercion_enabled: bool = False,
+ ignore_unexpected_complex_type_values: bool = False,
+ use_legacy_error_handling: bool = False,
+ ) -> Union["pa.Array", "pa.ChunkedArray"]:
+ """Convert a pandas column using the legacy conversion strategy."""
+ import pandas as pd
+ import pyarrow as pa
+
+ from pyspark.errors import PySparkTypeError, PySparkValueError
+ from pyspark.sql.pandas.types import _create_converter_from_pandas, to_arrow_type
+
+ field_name = field.name
Review Comment:
Got it. I will add this back in later PRs. Thanks for pointing this out!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org