From c0ea8c58d23d34d4012dbcc76227c0185a2b4623 Mon Sep 17 00:00:00 2001 From: Spenser Sun Date: Fri, 11 Sep 2026 21:30:50 +0000 Subject: [PATCH] [SPARK-54966][PYTHON] Factor out legacy pandas-to-Arrow column conversion Separate batch handling in from_pandas from single-Series dispatch in convert and the existing implementation in convert_legacy. Clarify the legacy error-handling flag and update the worker callers and tests. --- python/pyspark/sql/conversion.py | 257 ++++++++++++-------- python/pyspark/sql/tests/test_conversion.py | 66 ++--- python/pyspark/worker.py | 32 +-- 3 files changed, 201 insertions(+), 154 deletions(-) diff --git a/python/pyspark/sql/conversion.py b/python/pyspark/sql/conversion.py index fc03acf026436..d42f374916bde 100644 --- a/python/pyspark/sql/conversion.py +++ b/python/pyspark/sql/conversion.py @@ -309,7 +309,7 @@ class PandasToArrowConversion: """ @classmethod - def convert( + def from_pandas( cls, data: Union["pd.DataFrame", Sequence[Union["pd.Series", "pd.DataFrame"]]], schema: StructType, @@ -321,7 +321,7 @@ def convert( assign_cols_by_name: bool = False, int_to_decimal_coercion_enabled: bool = False, ignore_unexpected_complex_type_values: bool = False, - is_legacy: bool = False, + use_legacy_error_handling: bool = False, ) -> "pa.RecordBatch": """ Convert a pandas DataFrame or list of Series/DataFrames to an Arrow RecordBatch. @@ -348,13 +348,11 @@ def convert( Whether to enable int to decimal coercion (default False) ignore_unexpected_complex_type_values : bool Whether to ignore unexpected complex type values in converter (default False) - is_legacy : bool - Whether to use the legacy pandas-to-Arrow conversion path. The legacy - path uses broader Arrow exception handling (ArrowException) to allow - more implicit type coercions (e.g., int->boolean, dict->struct via - ArrowTypeError). The non-legacy path only catches ArrowInvalid for - the cast fallback, so type mismatches like string->decimal raise - immediately. (default False) + use_legacy_error_handling : bool + Whether to use legacy error handling and error messages. Legacy handling + catches ArrowException (including ArrowTypeError) for the cast fallback. + The new error handling only catches ArrowInvalid for the fallback; + ArrowTypeError is raised without retrying the conversion. (default False) Returns ------- @@ -363,8 +361,7 @@ def convert( 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 + from pyspark.sql.pandas.types import to_arrow_type # Handle empty schema (0 columns) # Use dummy column + select([]) to preserve row count (PyArrow limitation workaround) @@ -390,17 +387,11 @@ def convert( else: columns = list(data) - def convert_column( - col: Union["pd.Series", "pd.DataFrame"], field: StructField - ) -> "pa.Array": - """Convert a single column (Series or DataFrame) to an Arrow Array. - - Uses field.name for error messages instead of series.name to avoid - copying the Series via rename() - a ~20% overhead on the hot path. - """ + converted: List[Union["pa.Array", "pa.ChunkedArray"]] = [] + for col, field in zip(columns, schema.fields): if isinstance(col, pd.DataFrame): assert isinstance(field.dataType, StructType) - nested_batch = cls.convert( + nested_batch = cls.from_pandas( col, field.dataType, timezone=timezone, @@ -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( + 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( + 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 Series to an Arrow Array or ChunkedArray.""" + return cls.convert_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_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 Series to an Arrow Array or ChunkedArray.""" + 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 + ret_type = field.dataType + + if isinstance(series.dtype, pd.CategoricalDtype): + series = series.astype(series.dtype.categories.dtype) - mask = None if hasattr(series.array, "__arrow_array__") else series.isnull() + 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) - if is_legacy: - # Legacy pandas conversion path: broad ArrowException catch so - # that both ArrowInvalid AND ArrowTypeError (e.g. dict->struct) - # trigger the cast fallback. + mask = None if hasattr(series.array, "__arrow_array__") else series.isnull() + + if use_legacy_error_handling: + # Legacy error handling: both ArrowInvalid and ArrowTypeError can + # trigger the cast fallback when arrow_cast is enabled. + try: try: - try: - return pa.Array.from_pandas( - series, mask=mask, type=arrow_type, safe=safecheck + return pa.Array.from_pandas(series, mask=mask, type=arrow_type, safe=safecheck) + except pa.lib.ArrowException: # broad: includes ArrowTypeError + if arrow_cast: + return pa.Array.from_pandas(series, mask=mask).cast( + target_type=arrow_type, safe=safecheck ) - except pa.lib.ArrowException: # broad: includes ArrowTypeError - if arrow_cast: - return pa.Array.from_pandas(series, mask=mask).cast( - target_type=arrow_type, safe=safecheck - ) - raise - except pa.lib.ArrowException as e: - error_msg = ( - "Exception thrown when converting pandas.Series (%s) " - "with name '%s' to Arrow Array (%s)." - % (series.dtype, field_name, arrow_type) + raise + except pa.lib.ArrowException as e: + error_msg = ( + "Exception thrown when converting pandas.Series (%s) " + "with name '%s' to Arrow Array (%s)." % (series.dtype, field_name, arrow_type) + ) + if isinstance(e, TypeError): + raise PySparkTypeError(error_msg) from e + if safecheck: + error_msg += ( + " It can be caused by overflows or other " + "unsafe conversions warned by Arrow. Arrow safe " + "type check can be disabled by using SQL config " + "`spark.sql.execution.pandas." + "convertToArrowArraySafely`." ) - if isinstance(e, TypeError): - raise PySparkTypeError(error_msg) from e - if safecheck: - error_msg += ( - " It can be caused by overflows or other " - "unsafe conversions warned by Arrow. Arrow safe " - "type check can be disabled by using SQL config " - "`spark.sql.execution.pandas." - "convertToArrowArraySafely`." - ) - raise PySparkValueError(error_msg) from e - else: - # Non-legacy path: only ArrowInvalid triggers the cast fallback. - # ArrowTypeError (e.g. string->decimal) must NOT be silently cast. + raise PySparkValueError(error_msg) from e + else: + # Non-legacy path: only ArrowInvalid triggers the cast fallback. + # ArrowTypeError must NOT be silently cast. + try: try: - try: - return pa.Array.from_pandas( - series, mask=mask, type=arrow_type, safe=safecheck + return pa.Array.from_pandas(series, mask=mask, type=arrow_type, safe=safecheck) + except pa.lib.ArrowInvalid: # narrow: skip ArrowTypeError + if arrow_cast: + return pa.Array.from_pandas(series, mask=mask).cast( + target_type=arrow_type, safe=safecheck ) - except pa.lib.ArrowInvalid: # narrow: skip ArrowTypeError - if arrow_cast: - return pa.Array.from_pandas(series, mask=mask).cast( - target_type=arrow_type, safe=safecheck - ) - raise - except TypeError as e: - raise PySparkTypeError( - f"Cannot convert the output value of the column " - f"'{field_name}' with type '{series.dtype}' to the " - f"specified return type of the column: '{arrow_type}'." - f" Please check if the data types match and try again." - ) from e - except ValueError as e: - error_msg = ( - f"Failed to convert the value of the column " - f"'{field_name}' with type '{series.dtype}' to Arrow " - f"type '{arrow_type}'." + raise + except TypeError as e: + raise PySparkTypeError( + f"Cannot convert the output value of the column " + f"'{field_name}' with type '{series.dtype}' to the " + f"specified return type of the column: '{arrow_type}'." + f" Please check if the data types match and try again." + ) from e + except ValueError as e: + error_msg = ( + f"Failed to convert the value of the column " + f"'{field_name}' with type '{series.dtype}' to Arrow " + f"type '{arrow_type}'." + ) + if safecheck: + error_msg += ( + " It can be caused by overflows or other unsafe " + "conversions warned by Arrow. Arrow safe type " + "check can be disabled by using SQL config " + "`spark.sql.execution.pandas." + "convertToArrowArraySafely`." ) - if safecheck: - error_msg += ( - " It can be caused by overflows or other unsafe " - "conversions warned by Arrow. Arrow safe type " - "check can be disabled by using SQL config " - "`spark.sql.execution.pandas." - "convertToArrowArraySafely`." - ) - raise PySparkValueError(error_msg) from e - - converted = [convert_column(col, field) for col, field in zip(columns, schema.fields)] - # 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) + raise PySparkValueError(error_msg) from e class LocalDataToArrowConversion: diff --git a/python/pyspark/sql/tests/test_conversion.py b/python/pyspark/sql/tests/test_conversion.py index f660ffa0e7d32..feb11ae6f56c4 100644 --- a/python/pyspark/sql/tests/test_conversion.py +++ b/python/pyspark/sql/tests/test_conversion.py @@ -270,7 +270,7 @@ def test_enforce_schema_table_input(self): @unittest.skipIf(not have_pyarrow, pyarrow_requirement_message) @unittest.skipIf(not have_pandas, pandas_requirement_message) class PandasToArrowConversionTests(unittest.TestCase): - def test_convert(self): + def test_from_pandas(self): """Test basic DataFrame/Series to Arrow RecordBatch conversion.""" import pandas as pd import pyarrow as pa @@ -278,7 +278,7 @@ def test_convert(self): # Basic DataFrame conversion df = pd.DataFrame({"a": [1, 2, 3], "b": [1.0, 2.0, 3.0]}) schema = StructType([StructField("a", IntegerType()), StructField("b", DoubleType())]) - result = PandasToArrowConversion.convert(df, schema) + result = PandasToArrowConversion.from_pandas(df, schema) self.assertIsInstance(result, pa.RecordBatch) self.assertEqual(result.num_rows, 3) self.assertEqual(result.num_columns, 2) @@ -286,26 +286,26 @@ def test_convert(self): # List of Series input series_list = [pd.Series([1, 2, 3]), pd.Series([1.0, 2.0, 3.0])] - result = PandasToArrowConversion.convert(series_list, schema) + result = PandasToArrowConversion.from_pandas(series_list, schema) self.assertEqual(result.num_rows, 3) # With nulls df = pd.DataFrame({"a": [1, None, 3], "b": [1.0, 2.0, None]}) - result = PandasToArrowConversion.convert(df, schema) + result = PandasToArrowConversion.from_pandas(df, schema) self.assertEqual(result.column(0).to_pylist(), [1, None, 3]) # Empty DataFrame (0 rows) df = pd.DataFrame({"a": pd.Series([], dtype=int), "b": pd.Series([], dtype=float)}) - result = PandasToArrowConversion.convert(df, schema) + result = PandasToArrowConversion.from_pandas(df, schema) self.assertEqual(result.num_rows, 0) # Empty schema (0 columns) should preserve row count df = pd.DataFrame({"a": [1, 2, 3], "b": [4.0, 5.0, 6.0]}) - result = PandasToArrowConversion.convert(df, StructType([])) + result = PandasToArrowConversion.from_pandas(df, StructType([])) self.assertEqual(result.num_columns, 0) self.assertEqual(result.num_rows, 3) - def test_convert_assign_cols_by_name(self): + def test_from_pandas_assign_cols_by_name(self): """Test assign_cols_by_name reorders columns to match schema.""" import pandas as pd @@ -314,18 +314,18 @@ def test_convert_assign_cols_by_name(self): schema = StructType([StructField("a", IntegerType()), StructField("b", StringType())]) # With assign_cols_by_name=True - reorders columns to match schema field names - result = PandasToArrowConversion.convert(df, schema, assign_cols_by_name=True) + result = PandasToArrowConversion.from_pandas(df, schema, assign_cols_by_name=True) self.assertEqual(result.column(0).to_pylist(), [1, 2, 3]) # a self.assertEqual(result.column(1).to_pylist(), ["x", "y", "z"]) # b # Without assign_cols_by_name - uses positional order (b first, a second) df = pd.DataFrame({"b": [10, 20, 30], "a": [1.0, 2.0, 3.0]}) schema = StructType([StructField("x", IntegerType()), StructField("y", DoubleType())]) - result = PandasToArrowConversion.convert(df, schema, assign_cols_by_name=False) + result = PandasToArrowConversion.from_pandas(df, schema, assign_cols_by_name=False) self.assertEqual(result.column(0).to_pylist(), [10, 20, 30]) # positional: b -> x self.assertEqual(result.column(1).to_pylist(), [1.0, 2.0, 3.0]) # positional: a -> y - def test_convert_timezone(self): + def test_from_pandas_timezone(self): """Test timezone handling for timestamp conversion.""" import pandas as pd @@ -334,11 +334,11 @@ def test_convert_timezone(self): schema = StructType([StructField("ts", TimestampType())]) # Convert with timezone - result = PandasToArrowConversion.convert(df, schema, timezone="UTC") + result = PandasToArrowConversion.from_pandas(df, schema, timezone="UTC") self.assertEqual(result.num_rows, 2) self.assertEqual(result.num_columns, 1) - def test_convert_arrow_cast(self): + def test_from_pandas_arrow_cast(self): """Test arrow_cast allows type coercion on mismatch.""" import pandas as pd @@ -347,10 +347,10 @@ def test_convert_arrow_cast(self): schema = StructType([StructField("a", LongType())]) # With arrow_cast=True, should allow the conversion - result = PandasToArrowConversion.convert(df, schema, arrow_cast=True) + result = PandasToArrowConversion.from_pandas(df, schema, arrow_cast=True) self.assertEqual(result.column(0).to_pylist(), [1, 2, 3]) - def test_convert_decimal(self): + def test_from_pandas_decimal(self): """Test int to decimal coercion.""" from decimal import Decimal @@ -361,13 +361,15 @@ def test_convert_decimal(self): schema = StructType([StructField("a", DecimalType(10, 2))]) # With int_to_decimal_coercion_enabled=True - result = PandasToArrowConversion.convert(df, schema, int_to_decimal_coercion_enabled=True) + result = PandasToArrowConversion.from_pandas( + df, schema, int_to_decimal_coercion_enabled=True + ) self.assertEqual(result.num_rows, 3) # Values should be converted to decimal values = result.column(0).to_pylist() self.assertEqual(values, [Decimal("1.00"), Decimal("2.00"), Decimal("3.00")]) - def test_convert_struct(self): + def test_from_pandas_struct(self): """Test struct type conversion via nested DataFrame columns.""" import pandas as pd import pyarrow as pa @@ -383,7 +385,7 @@ def test_convert_struct(self): ) # List input: second element is a DataFrame (struct column) data = [pd.Series([1, 2]), pd.DataFrame({"x": [10, 20], "y": [1.1, 2.2]})] - result = PandasToArrowConversion.convert(data, schema) + result = PandasToArrowConversion.from_pandas(data, schema) self.assertEqual(result.num_rows, 2) self.assertEqual(result.num_columns, 2) # Struct column should be a StructArray @@ -394,10 +396,10 @@ def test_convert_struct(self): pd.Series([], dtype=int), pd.DataFrame({"x": pd.Series([], dtype=int), "y": pd.Series([], dtype=float)}), ] - result = PandasToArrowConversion.convert(data, schema) + result = PandasToArrowConversion.from_pandas(data, schema) self.assertEqual(result.num_rows, 0) - def test_convert_error_messages(self): + def test_from_pandas_error_messages(self): """Test error messages include series name from schema field.""" import pandas as pd @@ -406,12 +408,12 @@ def test_convert_error_messages(self): # Type mismatch: string data for integer column data = [pd.Series(["not_int", "bad"]), pd.Series(["a", "b"])] with self.assertRaises((PySparkValueError, PySparkTypeError)) as ctx: - PandasToArrowConversion.convert(data, schema) + PandasToArrowConversion.from_pandas(data, schema) # Error message should use the new format and reference the schema field name self.assertIn("age", str(ctx.exception)) - def test_convert_is_legacy(self): - """Test is_legacy=True uses the legacy error format.""" + def test_from_pandas_use_legacy_error_handling(self): + """Test use_legacy_error_handling=True uses the legacy error format.""" import pandas as pd schema = StructType([StructField("val", DoubleType())]) @@ -419,7 +421,7 @@ def test_convert_is_legacy(self): # ValueError path (string -> double) with self.assertRaises(PySparkValueError) as ctx: - PandasToArrowConversion.convert(data, schema, is_legacy=True) + PandasToArrowConversion.from_pandas(data, schema, use_legacy_error_handling=True) self.assertIn("Exception thrown when converting pandas.Series", str(ctx.exception)) self.assertIn("val", str(ctx.exception)) @@ -431,16 +433,16 @@ def test_convert_is_legacy(self): ) data = [pd.Series([0, 1])] with self.assertRaises(PySparkTypeError) as ctx: - PandasToArrowConversion.convert( + PandasToArrowConversion.from_pandas( data, struct_schema, - is_legacy=True, + use_legacy_error_handling=True, ignore_unexpected_complex_type_values=True, ) self.assertIn("Exception thrown when converting pandas.Series", str(ctx.exception)) self.assertIn("x", str(ctx.exception)) - def test_convert_prefers_large_types(self): + def test_from_pandas_prefers_large_types(self): """Test prefers_large_types produces large Arrow types.""" import pandas as pd import pyarrow as pa @@ -448,22 +450,22 @@ def test_convert_prefers_large_types(self): df = pd.DataFrame({"s": ["hello", "world"]}) schema = StructType([StructField("s", StringType())]) - result = PandasToArrowConversion.convert(df, schema, prefers_large_types=True) + result = PandasToArrowConversion.from_pandas(df, schema, prefers_large_types=True) self.assertEqual(result.column(0).type, pa.large_string()) - result = PandasToArrowConversion.convert(df, schema, prefers_large_types=False) + result = PandasToArrowConversion.from_pandas(df, schema, prefers_large_types=False) self.assertEqual(result.column(0).type, pa.string()) - def test_convert_categorical(self): + def test_from_pandas_categorical(self): """Test CategoricalDtype series is correctly converted.""" import pandas as pd cat_series = pd.Series(pd.Categorical(["a", "b", "a", "c"])) schema = StructType([StructField("cat", StringType())]) - result = PandasToArrowConversion.convert([cat_series], schema) + result = PandasToArrowConversion.from_pandas([cat_series], schema) self.assertEqual(result.column(0).to_pylist(), ["a", "b", "a", "c"]) - def test_convert_chunked_array_backed(self): + def test_from_pandas_chunked_array_backed(self): """Test a chunked arrow-backed series is converted to a single Array.""" import pandas as pd import pyarrow as pa @@ -474,7 +476,7 @@ def test_convert_chunked_array_backed(self): series = pd.Series(chunked, dtype="string[pyarrow]") schema = StructType([StructField("s", StringType())]) - result = PandasToArrowConversion.convert([series], schema, arrow_cast=True) + result = PandasToArrowConversion.from_pandas([series], schema, arrow_cast=True) self.assertIsInstance(result.column(0), pa.Array) self.assertEqual(result.column(0).to_pylist(), ["a", "b", "c", "d", "e"]) diff --git a/python/pyspark/worker.py b/python/pyspark/worker.py index 878b7bb63a126..3d8a947afeb10 100644 --- a/python/pyspark/worker.py +++ b/python/pyspark/worker.py @@ -1453,7 +1453,7 @@ def check_return_value(res: Any, method_name: str) -> Iterator: def convert_df_to_arrow(result: "pd.DataFrame") -> "pa.RecordBatch": # Convert the output pandas DataFrame into a single "_0" struct column, # applying the legacy pandas-to-Arrow coercions. - return PandasToArrowConversion.convert( + return PandasToArrowConversion.from_pandas( [result], output_schema, timezone=runner_conf.timezone, @@ -1462,7 +1462,7 @@ def convert_df_to_arrow(result: "pd.DataFrame") -> "pa.RecordBatch": assign_cols_by_name=False, int_to_decimal_coercion_enabled=runner_conf.int_to_decimal_coercion_enabled, ignore_unexpected_complex_type_values=True, - is_legacy=True, + use_legacy_error_handling=True, ) def evaluate_rows( @@ -2026,7 +2026,7 @@ def _elementwise_result_to_arrow(result, return_type, arrow_element_type, is_pan import pyarrow as pa if is_pandas: - batch = PandasToArrowConversion.convert( + batch = PandasToArrowConversion.from_pandas( [result], StructType([StructField("_0", return_type)]), timezone=runner_conf.timezone, @@ -2542,7 +2542,7 @@ def grouped_func( for udf_func, args_offsets, kwargs_offsets, _ in udfs ] result_series = [pd.Series([r]) for r in results] - yield PandasToArrowConversion.convert( + yield PandasToArrowConversion.from_pandas( result_series, output_schema, timezone=runner_conf.timezone, @@ -2587,7 +2587,7 @@ def grouped_func( # Drain remaining batches to maintain stream position for _ in series_iter: pass - yield PandasToArrowConversion.convert( + yield PandasToArrowConversion.from_pandas( [pd.Series([result])], output_schema, timezone=runner_conf.timezone, @@ -2835,7 +2835,7 @@ def grouped_func( messageParameters={"window_bound_type": bound_type}, ) - yield PandasToArrowConversion.convert( + yield PandasToArrowConversion.from_pandas( result_series, output_schema, timezone=runner_conf.timezone, @@ -3027,7 +3027,7 @@ def grouped_func( truncate_return_schema=False, ) - yield PandasToArrowConversion.convert( + yield PandasToArrowConversion.from_pandas( [result], output_schema, timezone=runner_conf.timezone, @@ -3099,7 +3099,7 @@ def dataframe_iter(): runner_conf.assign_cols_by_name, truncate_return_schema=False, ) - yield PandasToArrowConversion.convert( + yield PandasToArrowConversion.from_pandas( [df], output_schema, timezone=runner_conf.timezone, @@ -3231,7 +3231,7 @@ def dataframe_iter(): verify_pandas_result( df, return_type, assign_cols_by_name=True, truncate_return_schema=True ) - yield PandasToArrowConversion.convert( + yield PandasToArrowConversion.from_pandas( [df], output_schema, timezone=runner_conf.timezone, @@ -3298,7 +3298,7 @@ def cogrouped_func( truncate_return_schema=False, ) - yield PandasToArrowConversion.convert( + yield PandasToArrowConversion.from_pandas( [result], output_schema, timezone=runner_conf.timezone, @@ -3480,7 +3480,7 @@ def func(split_index: int, data: Iterator[pa.RecordBatch]) -> Iterator[pa.Record result_series.append(pd.Series(results)) # --- Output: pandas -> Arrow --- - yield PandasToArrowConversion.convert( + yield PandasToArrowConversion.from_pandas( result_series, return_schema, timezone=runner_conf.timezone, @@ -3969,7 +3969,7 @@ def func(split_index: int, data: Iterator[pa.RecordBatch]) -> Iterator[pa.Record results.append(result) # --- Output: pandas -> Arrow --- - yield PandasToArrowConversion.convert( + yield PandasToArrowConversion.from_pandas( results, return_schema, timezone=runner_conf.timezone, @@ -4027,7 +4027,7 @@ def process_results(): verify_pandas_result( result, return_type, assign_cols_by_name=True, truncate_return_schema=True ) - yield PandasToArrowConversion.convert( + yield PandasToArrowConversion.from_pandas( [result], return_schema, timezone=runner_conf.timezone, @@ -4158,7 +4158,7 @@ def convert_results(result_iter): "Invalid return type. Please make sure that the UDF returns a " "pandas.DataFrame when the specified return type is StructType." ) - yield PandasToArrowConversion.convert( + yield PandasToArrowConversion.from_pandas( [result], output_schema, timezone=runner_conf.timezone, @@ -4363,7 +4363,7 @@ def convert_results( "Invalid return type. Please make sure that the UDF returns a " "pandas.DataFrame when the specified return type is StructType." ) - yield PandasToArrowConversion.convert( + yield PandasToArrowConversion.from_pandas( [result], output_schema, timezone=runner_conf.timezone, @@ -4671,7 +4671,7 @@ def construct_record_batch( StructField("_2", result_state_df_type), ] ) - return PandasToArrowConversion.convert( + return PandasToArrowConversion.from_pandas( data, schema, timezone=runner_conf.timezone,