[SPARK-53696][PYTHON][CONNECT][SQL]Default to bytes for BinaryType in PySpark

docs/sql-ref-datatypes.md

@@ -131,7 +131,7 @@ from pyspark.sql.types import *
 |**StringType**|str|StringType()|
 |**CharType(length)**|str|CharType(length)|
 |**VarcharType(length)**|str|VarcharType(length)|
-|**BinaryType**|bytearray|BinaryType()|
+|**BinaryType**|bytes|BinaryType()|
 |**BooleanType**|bool|BooleanType()|
 |**TimestampType**|datetime.datetime|TimestampType()|
 |**TimestampNTZType**|datetime.datetime|TimestampNTZType()|

python/docs/source/tutorial/sql/type_conversions.rst

@@ -67,6 +67,8 @@ are listed below:
     * - spark.sql.execution.pandas.inferPandasDictAsMap
       - When enabled, Pandas dictionaries are inferred as MapType. Otherwise, they are inferred as StructType.
       - False
+    * - spark.sql.execution.pyspark.binaryAsBytes
+      - Introduced in Spark 4.1.0. When enabled, BinaryType is mapped consistently to Python bytes; when disabled, matches the PySpark default behavior before 4.1.0.
 
 All Conversions
 ---------------
@@ -105,7 +107,7 @@ All Conversions
       - string
       - StringType()
     * - **BinaryType**
-      - bytearray
+      - bytes
       - BinaryType()
     * - **BooleanType**
       - bool

python/pyspark/sql/connect/dataframe.py

@@ -1823,7 +1823,16 @@ def collect(self) -> List[Row]:
 
         assert schema is not None and isinstance(schema, StructType)
 
-        return ArrowTableToRowsConversion.convert(table, schema)
+        return ArrowTableToRowsConversion.convert(
+            table, schema, binary_as_bytes=self._get_binary_as_bytes()
+        )
+
+    def _get_binary_as_bytes(self) -> bool:
+        """Get the binary_as_bytes configuration value from Spark session."""
+        return (
+            self._session.conf.get("spark.sql.execution.pyspark.binaryAsBytes", "true").lower()
+            == "true"
+        )
 
     def _to_table(self) -> Tuple["pa.Table", Optional[StructType]]:
         query = self._plan.to_proto(self._session.client)
@@ -2075,7 +2084,9 @@ def toLocalIterator(self, prefetchPartitions: bool = False) -> Iterator[Row]:
                 table = schema_or_table
                 if schema is None:
                     schema = from_arrow_schema(table.schema, prefer_timestamp_ntz=True)
-                yield from ArrowTableToRowsConversion.convert(table, schema)
+                yield from ArrowTableToRowsConversion.convert(
+                    table, schema, binary_as_bytes=self._get_binary_as_bytes()
+                )
 
     def pandas_api(
         self, index_col: Optional[Union[str, List[str]]] = None
@@ -2161,8 +2172,12 @@ def foreach_func(row: Any) -> None:
 
     def foreachPartition(self, f: Callable[[Iterator[Row]], None]) -> None:
         schema = self._schema
+        binary_as_bytes = self._get_binary_as_bytes()
         field_converters = [
-            ArrowTableToRowsConversion._create_converter(f.dataType) for f in schema.fields
+            ArrowTableToRowsConversion._create_converter(
+                f.dataType, binary_as_bytes=binary_as_bytes
+            )
+            for f in schema.fields
         ]
 
         def foreach_partition_func(itr: Iterable[pa.RecordBatch]) -> Iterable[pa.RecordBatch]:

python/pyspark/sql/conversion.py

@@ -518,13 +518,13 @@ def _create_converter(dataType: DataType) -> Callable:
     @overload
     @staticmethod
     def _create_converter(
-        dataType: DataType, *, none_on_identity: bool = True
+        dataType: DataType, *, none_on_identity: bool = True, binary_as_bytes: bool = True
     ) -> Optional[Callable]:
         pass
 
     @staticmethod
     def _create_converter(
-        dataType: DataType, *, none_on_identity: bool = False
+        dataType: DataType, *, none_on_identity: bool = False, binary_as_bytes: bool = True
     ) -> Optional[Callable]:
         assert dataType is not None and isinstance(dataType, DataType)
 
@@ -542,7 +542,9 @@ def _create_converter(
             dedup_field_names = _dedup_names(field_names)
 
             field_convs = [
-                ArrowTableToRowsConversion._create_converter(f.dataType, none_on_identity=True)
+                ArrowTableToRowsConversion._create_converter(
+                    f.dataType, none_on_identity=True, binary_as_bytes=binary_as_bytes
+                )
                 for f in dataType.fields
             ]
 
@@ -564,7 +566,7 @@ def convert_struct(value: Any) -> Any:
 
         elif isinstance(dataType, ArrayType):
             element_conv = ArrowTableToRowsConversion._create_converter(
-                dataType.elementType, none_on_identity=True
+                dataType.elementType, none_on_identity=True, binary_as_bytes=binary_as_bytes
             )
 
             if element_conv is None:
@@ -589,10 +591,10 @@ def convert_array(value: Any) -> Any:
 
         elif isinstance(dataType, MapType):
             key_conv = ArrowTableToRowsConversion._create_converter(
-                dataType.keyType, none_on_identity=True
+                dataType.keyType, none_on_identity=True, binary_as_bytes=binary_as_bytes
             )
             value_conv = ArrowTableToRowsConversion._create_converter(
-                dataType.valueType, none_on_identity=True
+                dataType.valueType, none_on_identity=True, binary_as_bytes=binary_as_bytes
             )
 
             if key_conv is None:
@@ -646,7 +648,7 @@ def convert_binary(value: Any) -> Any:
                     return None
                 else:
                     assert isinstance(value, bytes)
-                    return bytearray(value)
+                    return value if binary_as_bytes else bytearray(value)
 
             return convert_binary
 
@@ -676,7 +678,7 @@ def convert_timestample_ntz(value: Any) -> Any:
             udt: UserDefinedType = dataType
 
             conv = ArrowTableToRowsConversion._create_converter(
-                udt.sqlType(), none_on_identity=True
+                udt.sqlType(), none_on_identity=True, binary_as_bytes=binary_as_bytes
             )
 
             if conv is None:
@@ -735,7 +737,11 @@ def convert(
 
     @staticmethod  # type: ignore[misc]
     def convert(
-        table: "pa.Table", schema: StructType, *, return_as_tuples: bool = False
+        table: "pa.Table",
+        schema: StructType,
+        *,
+        return_as_tuples: bool = False,
+        binary_as_bytes: bool = True,
     ) -> List[Union[Row, tuple]]:
         require_minimum_pyarrow_version()
         import pyarrow as pa
@@ -748,7 +754,9 @@ def convert(
 
         if len(fields) > 0:
             field_converters = [
-                ArrowTableToRowsConversion._create_converter(f.dataType, none_on_identity=True)
+                ArrowTableToRowsConversion._create_converter(
+                    f.dataType, none_on_identity=True, binary_as_bytes=binary_as_bytes
+                )
                 for f in schema.fields
             ]
 

python/pyspark/sql/pandas/serializers.py

@@ -854,14 +854,17 @@ class ArrowBatchUDFSerializer(ArrowStreamArrowUDFSerializer):
     int_to_decimal_coercion_enabled : bool
         If True, applies additional coercions in Python before converting to Arrow
         This has performance penalties.
+    binary_as_bytes : bool
+        If True, binary type will be deserialized as bytes, otherwise as bytearray.
     """
 
     def __init__(
         self,
         timezone,
         safecheck,
         input_types,
-        int_to_decimal_coercion_enabled=False,
+        int_to_decimal_coercion_enabled,
+        binary_as_bytes,
     ):
         super().__init__(
             timezone=timezone,
@@ -871,6 +874,7 @@ def __init__(
         )
         self._input_types = input_types
         self._int_to_decimal_coercion_enabled = int_to_decimal_coercion_enabled
+        self._binary_as_bytes = binary_as_bytes
 
     def load_stream(self, stream):
         """
@@ -887,7 +891,9 @@ def load_stream(self, stream):
             List of columns containing list of Python values.
         """
         converters = [
-            ArrowTableToRowsConversion._create_converter(dt, none_on_identity=True)
+            ArrowTableToRowsConversion._create_converter(
+                dt, none_on_identity=True, binary_as_bytes=self._binary_as_bytes
+            )
             for dt in self._input_types
         ]
 

python/pyspark/sql/tests/arrow/test_arrow_python_udf.py

@@ -19,9 +19,19 @@
 
 from pyspark.errors import AnalysisException, PythonException, PySparkNotImplementedError
 from pyspark.sql import Row
-from pyspark.sql.functions import udf
+from pyspark.sql.functions import udf, col
 from pyspark.sql.tests.test_udf import BaseUDFTestsMixin
-from pyspark.sql.types import DayTimeIntervalType, VarcharType, StructType, StructField, StringType
+from pyspark.sql.types import (
+    ArrayType,
+    BinaryType,
+    DayTimeIntervalType,
+    IntegerType,
+    MapType,
+    StringType,
+    StructField,
+    StructType,
+    VarcharType,
+)
 from pyspark.testing.sqlutils import (
     have_pandas,
     have_pyarrow,
@@ -422,6 +432,93 @@ def tearDownClass(cls):
         finally:
             super(ArrowPythonUDFNonLegacyTests, cls).tearDownClass()
 
+    def test_udf_binary_type(self):
+        def get_binary_type(x):
+            return type(x).__name__
+
+        binary_udf = udf(get_binary_type, returnType="string", useArrow=True)
+
+        df = self.spark.createDataFrame(
+            [Row(b=b"hello"), Row(b=b"world")], schema=StructType([StructField("b", BinaryType())])
+        )
+
+        with self.sql_conf({"spark.sql.execution.pyspark.binaryAsBytes": "true"}):
+            result = df.select(binary_udf(col("b")).alias("type_name")).collect()
+            self.assertEqual(result[0]["type_name"], "bytes")
+            self.assertEqual(result[1]["type_name"], "bytes")
+
+        with self.sql_conf({"spark.sql.execution.pyspark.binaryAsBytes": "false"}):
+            result = df.select(binary_udf(col("b")).alias("type_name")).collect()
+            self.assertEqual(result[0]["type_name"], "bytearray")
+            self.assertEqual(result[1]["type_name"], "bytearray")
+
+    def test_udf_array_binary_type(self):
+        def check_array_binary_types(arr):
+            return [type(x).__name__ for x in arr]
+
+        array_binary_udf = udf(check_array_binary_types, returnType="array<string>", useArrow=True)
+
+        df = self.spark.createDataFrame(
+            [Row(arr_b=[b"a", b"b"]), Row(arr_b=[b"c", b"d"])],
+            schema=StructType([StructField("arr_b", ArrayType(BinaryType()))]),
+        )
+
+        with self.sql_conf({"spark.sql.execution.pyspark.binaryAsBytes": "true"}):
+            result = df.select(array_binary_udf(col("arr_b")).alias("types")).collect()
+            self.assertEqual(result[0]["types"], ["bytes", "bytes"])
+            self.assertEqual(result[1]["types"], ["bytes", "bytes"])
+
+        with self.sql_conf({"spark.sql.execution.pyspark.binaryAsBytes": "false"}):
+            result = df.select(array_binary_udf(col("arr_b")).alias("types")).collect()
+            self.assertEqual(result[0]["types"], ["bytearray", "bytearray"])
+            self.assertEqual(result[1]["types"], ["bytearray", "bytearray"])
+
+    def test_udf_map_binary_type(self):
+        def check_map_binary_types(m):
+            return [type(v).__name__ for v in m.values()]
+
+        map_binary_udf = udf(check_map_binary_types, returnType="array<string>", useArrow=True)
+
+        df = self.spark.createDataFrame(
+            [Row(map_b={"k1": b"v1", "k2": b"v2"}), Row(map_b={"k3": b"v3"})],
+            schema=StructType([StructField("map_b", MapType(StringType(), BinaryType()))]),
+        )
+
+        with self.sql_conf({"spark.sql.execution.pyspark.binaryAsBytes": "true"}):
+            result = df.select(map_binary_udf(col("map_b")).alias("types")).collect()
+            self.assertEqual(set(result[0]["types"]), {"bytes"})
+            self.assertEqual(result[1]["types"], ["bytes"])
+
+        with self.sql_conf({"spark.sql.execution.pyspark.binaryAsBytes": "false"}):
+            result = df.select(map_binary_udf(col("map_b")).alias("types")).collect()
+            self.assertEqual(set(result[0]["types"]), {"bytearray"})
+            self.assertEqual(result[1]["types"], ["bytearray"])
+
+    def test_udf_struct_binary_type(self):
+        def check_struct_binary_type(s):
+            return type(s.b).__name__
+
+        struct_binary_udf = udf(check_struct_binary_type, returnType="string", useArrow=True)
+
+        struct_schema = StructType(
+            [StructField("i", IntegerType()), StructField("b", BinaryType())]
+        )
+
+        df = self.spark.createDataFrame(
+            [Row(struct_b=Row(i=1, b=b"data1")), Row(struct_b=Row(i=2, b=b"data2"))],
+            schema=StructType([StructField("struct_b", struct_schema)]),
+        )
+
+        with self.sql_conf({"spark.sql.execution.pyspark.binaryAsBytes": "true"}):
+            result = df.select(struct_binary_udf(col("struct_b")).alias("type_name")).collect()
+            self.assertEqual(result[0]["type_name"], "bytes")
+            self.assertEqual(result[1]["type_name"], "bytes")
+
+        with self.sql_conf({"spark.sql.execution.pyspark.binaryAsBytes": "false"}):
+            result = df.select(struct_binary_udf(col("struct_b")).alias("type_name")).collect()
+            self.assertEqual(result[0]["type_name"], "bytearray")
+            self.assertEqual(result[1]["type_name"], "bytearray")
+
 
 if __name__ == "__main__":
     from pyspark.sql.tests.arrow.test_arrow_python_udf import *  # noqa: F401

python/pyspark/sql/tests/connect/client/test_client.py

@@ -163,6 +163,20 @@ def Interrupt(self, req: proto.InterruptRequest, metadata):
             resp.session_id = self._session_id
             return resp
 
+        def Config(self, req: proto.ConfigRequest, metadata):
+            self.req = req
+            resp = proto.ConfigResponse()
+            resp.session_id = self._session_id
+            if req.operation.HasField("get"):
+                pair = resp.pairs.add()
+                pair.key = req.operation.get.keys[0]
+                pair.value = "true"  # Default value
+            elif req.operation.HasField("get_with_default"):
+                pair = resp.pairs.add()
+                pair.key = req.operation.get_with_default.pairs[0].key
+                pair.value = req.operation.get_with_default.pairs[0].value or "true"
+            return resp
+
     # The _cleanup_ml_cache invocation will hang in this test (no valid spark cluster)
     # and it blocks the test process exiting because it is registered as the atexit handler
     # in `SparkConnectClient` constructor. To bypass the issue, patch the method in the test.