[WIP] POC for serializer changes

python/pyspark/errors/error-conditions.json

@@ -380,6 +380,11 @@
       "<arg_name> index out of range, got '<index>'."
     ]
   },
+  "INVALID_ARROW_BATCH_ZIP": {
+    "message": [
+      "Cannot zip Arrow batches/arrays: <reason>."
+    ]
+  },
   "INVALID_ARROW_UDTF_RETURN_TYPE": {
     "message": [
       "The return type of the arrow-optimized Python UDTF should be of type 'pandas.DataFrame', but the '<func>' method returned a value of type <return_type> with value: <value>."

python/pyspark/sql/connect/session.py

@@ -69,7 +69,7 @@
 from pyspark.sql.connect.readwriter import DataFrameReader
 from pyspark.sql.connect.streaming.readwriter import DataStreamReader
 from pyspark.sql.connect.streaming.query import StreamingQueryManager
-from pyspark.sql.pandas.serializers import ArrowStreamPandasSerializer
+from pyspark.sql.conversion import PandasBatchTransformer
 from pyspark.sql.pandas.types import (
     to_arrow_schema,
     to_arrow_type,
@@ -630,15 +630,15 @@ def createDataFrame(
 
             safecheck = configs["spark.sql.execution.pandas.convertToArrowArraySafely"]
 
-            ser = ArrowStreamPandasSerializer(cast(str, timezone), safecheck == "true", False)
-
             _table = pa.Table.from_batches(
                 [
-                    ser._create_batch(
+                    PandasBatchTransformer.to_arrow(
                         [
                             (c, at, st)
                             for (_, c), at, st in zip(data.items(), arrow_types, spark_types)
-                        ]
+                        ],
+                        timezone=cast(str, timezone),
+                        safecheck=safecheck == "true",
                     )
                 ]
             )