You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "xianzhe-databricks (via GitHub)" <gi...@apache.org> on 2025/09/26 10:46:29 UTC

[PR] [SPARK-53696][PYTHON]Default to bytes for BinaryType in PySpark UDF [spark]

xianzhe-databricks opened a new pull request, #52467:
URL: https://github.com/apache/spark/pull/52467

   <!--
   Thanks for sending a pull request!  Here are some tips for you:
     1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
     2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
     3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
     4. Be sure to keep the PR description updated to reflect all changes.
     5. Please write your PR title to summarize what this PR proposes.
     6. If possible, provide a concise example to reproduce the issue for a faster review.
     7. If you want to add a new configuration, please read the guideline first for naming configurations in
        'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
     8. If you want to add or modify an error type or message, please read the guideline first in
        'common/utils/src/main/resources/error/README.md'.
   -->
   
   ### What changes were proposed in this pull request?
   <!--
   Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue. 
   If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
     1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
     2. If you fix some SQL features, you can provide some references of other DBMSes.
     3. If there is design documentation, please add the link.
     4. If there is a discussion in the mailing list, please add the link.
   -->
   
   
   ### Why are the changes needed?
   <!--
   Please clarify why the changes are needed. For instance,
     1. If you propose a new API, clarify the use case for a new API.
     2. If you fix a bug, you can clarify why it is a bug.
   -->
   
   
   ### Does this PR introduce _any_ user-facing change?
   <!--
   Note that it means *any* user-facing change including all aspects such as new features, bug fixes, or other behavior changes. Documentation-only updates are not considered user-facing changes.
   
   If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
   If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
   If no, write 'No'.
   -->
   
   
   ### How was this patch tested?
   <!--
   If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
   If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
   If tests were not added, please describe why they were not added and/or why it was difficult to add.
   If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
   -->
   
   
   ### Was this patch authored or co-authored using generative AI tooling?
   <!--
   If generative AI tooling has been used in the process of authoring this patch, please include the
   phrase: 'Generated-by: ' followed by the name of the tool and its version.
   If no, write 'No'.
   Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
   -->
   


-- 
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-53696][PYTHON]Default to bytes for BinaryType in PySpark [spark]

Posted by "xianzhe-databricks (via GitHub)" <gi...@apache.org>.
xianzhe-databricks commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2391967264


##########
python/pyspark/sql/tests/connect/test_connect_collection.py:
##########
@@ -291,6 +324,125 @@ def test_collect_nested_type(self):
             ).collect(),
         )
 
+    def test_collect_binary_type(self):
+        """Test that df.collect() respects binary_as_bytes configuration for server-side data"""
+        # Use SQL to create data with binary type on the server side
+        # This ensures the data goes through Arrow conversion from server to client
+        query = """
+            SELECT * FROM VALUES
+            (CAST('hello' AS BINARY)),
+            (CAST('world' AS BINARY))
+            AS tab(b)
+        """
+
+        # Test with binary_as_bytes=True (default)
+        with self.both_conf({"spark.sql.execution.pyspark.binaryAsBytes": "true"}):
+            connect_rows = self.connect.sql(query).collect()
+            self.assertEqual(len(connect_rows), 2)
+            for row in connect_rows:
+                self.assertIsInstance(row.b, bytes)
+
+            spark_rows = self.spark.sql(query).collect()
+            self.assertEqual(len(spark_rows), 2)
+            for row in spark_rows:
+                self.assertIsInstance(row.b, bytes)
+
+        # Test with binary_as_bytes=False
+        with self.both_conf({"spark.sql.execution.pyspark.binaryAsBytes": "false"}):
+            connect_rows = self.connect.sql(query).collect()
+            self.assertEqual(len(connect_rows), 2)
+            for row in connect_rows:
+                self.assertIsInstance(row.b, bytearray)
+
+            spark_rows = self.spark.sql(query).collect()
+            self.assertEqual(len(spark_rows), 2)
+            for row in spark_rows:
+                self.assertIsInstance(row.b, bytearray)
+
+    def test_to_local_iterator_binary_type(self):
+        """Test that df.toLocalIterator() respects binary_as_bytes configuration"""
+        # Use server-side query that creates binary data
+        query = """
+            SELECT * FROM VALUES
+            (CAST('data1' AS BINARY)),
+            (CAST('data2' AS BINARY))
+            AS tab(b)
+        """
+
+        # Test with binary_as_bytes=True
+        with self.both_conf({"spark.sql.execution.pyspark.binaryAsBytes": "true"}):
+            connect_count = 0
+            for row in self.connect.sql(query).toLocalIterator():
+                self.assertIsInstance(row.b, bytes)
+                connect_count += 1
+            self.assertEqual(connect_count, 2)
+
+            spark_count = 0
+            for row in self.spark.sql(query).toLocalIterator():
+                self.assertIsInstance(row.b, bytes)
+                spark_count += 1
+            self.assertEqual(spark_count, 2)
+
+        # Test with binary_as_bytes=False
+        with self.both_conf({"spark.sql.execution.pyspark.binaryAsBytes": "false"}):
+            connect_count = 0
+            for row in self.connect.sql(query).toLocalIterator():
+                self.assertIsInstance(row.b, bytearray)
+                connect_count += 1
+            self.assertEqual(connect_count, 2)
+
+            spark_count = 0
+            for row in self.spark.sql(query).toLocalIterator():
+                self.assertIsInstance(row.b, bytearray)
+                spark_count += 1
+            self.assertEqual(spark_count, 2)
+
+    def test_foreach_partition_binary_type(self):
+        """Test that df.foreachPartition() respects binary_as_bytes configuration
+
+        Since foreachPartition() runs on executors and cannot return data to the driver,
+        we test by ensuring the function doesn't throw exceptions when it expects the correct types.
+        """
+        # Use server-side query that creates binary data
+        query = """
+            SELECT * FROM VALUES
+            (CAST('partition1' AS BINARY)),
+            (CAST('partition2' AS BINARY))
+            AS tab(b)
+        """
+
+        # Test with binary_as_bytes=True - should get bytes objects
+        with self.both_conf({"spark.sql.execution.pyspark.binaryAsBytes": "true"}):
+
+            def assert_bytes_type(iterator):
+                count = 0
+                for row in iterator:
+                    # This will raise an exception if the type is not bytes
+                    assert isinstance(row.b, bytes), f"Expected bytes, got {type(row.b).__name__}"
+                    count += 1
+                # Ensure we actually processed rows
+                assert count > 0, "No rows were processed"
+
+            self.connect.sql(query).foreachPartition(assert_bytes_type)
+            self.spark.sql(query).foreachPartition(assert_bytes_type)
+
+        # Test with binary_as_bytes=False - should get bytearray objects
+        with self.both_conf({"spark.sql.execution.pyspark.binaryAsBytes": "false"}):
+
+            def assert_bytearray_type(iterator):
+                count = 0
+                for row in iterator:
+                    # This will raise an exception if the type is not bytearray
+                    assert isinstance(
+                        row.b, bytearray
+                    ), f"Expected bytearray, got {type(row.b).__name__}"
+                    count += 1
+                # Ensure we actually processed rows
+                assert count > 0, "No rows were processed"
+
+            self.connect.sql(query).foreachPartition(assert_bytearray_type)
+            # self.spark.sql(query).foreachPartition(assert_bytearray_type)

Review Comment:
   need to investigate why this line fails: why is the SQL conf not propagated in this particular case?



-- 
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-53696][PYTHON]Default to bytes for BinaryType in PySpark [spark]

Posted by "xianzhe-databricks (via GitHub)" <gi...@apache.org>.
xianzhe-databricks commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2398636023


##########
python/pyspark/sql/connect/dataframe.py:
##########
@@ -1823,7 +1823,14 @@ 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."""
+        conf_value = self._session.conf.get("spark.sql.execution.pyspark.binaryAsBytes", "true")
+        return (conf_value or "true").lower() == "true"

Review Comment:
   The method `self._session.conf.get` returns only an `Optional[str]`. I think the conf is retrieved from Spark Connect Server. ([code pointer](https://github.com/apache/spark/blob/master/python/pyspark/sql/connect/conf.py#L68))
   
   This way, mypy does not complain about type incompatibility.



-- 
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-53696][PYTHON][CONNECT][SQL]Default to bytes for BinaryType in PySpark [spark]

Posted by "ueshin (via GitHub)" <gi...@apache.org>.
ueshin commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2408559407


##########
python/pyspark/sql/tests/connect/test_connect_collection.py:
##########
@@ -25,6 +26,38 @@
 
 
 class SparkConnectCollectionTests(ReusedMixedTestCase, PandasOnSparkTestUtils):
+    def connect_conf(self, conf_dict):

Review Comment:
   nit: how about moving this and `both_conf` to `ReusedMixedTestCase` to be reusable?



-- 
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-53696][PYTHON][CONNECT][SQL] Default to bytes for BinaryType in PySpark [spark]

Posted by "xianzhe-databricks (via GitHub)" <gi...@apache.org>.
xianzhe-databricks commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2435055643


##########
python/pyspark/sql/tests/test_udtf.py:
##########
@@ -3044,6 +3044,19 @@ def eval(self, v1, v2, v3, v4):
         for idx, field in enumerate(result_df.schema.fields):
             self.assertEqual(field.dataType, expected_output_types[idx])
 
+    def test_udtf_binary_type(self):

Review Comment:
   what test did you mean? The same test case but applied for arrow udtf? See another thread, as the class for arrow udtf inherits the class for regular udtf, it is already tested



-- 
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-53696][PYTHON]Default to bytes for BinaryType in PySpark arrow UDF [spark]

Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2386406508


##########
python/pyspark/sql/tests/arrow/test_arrow_binary_as_bytes_udf.py:
##########
@@ -0,0 +1,145 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import unittest
+
+from pyspark.sql import Row
+from pyspark.sql.functions import udf, col
+from pyspark.sql.types import (
+    ArrayType,
+    BinaryType,
+    IntegerType,
+    MapType,
+    StringType,
+    StructType,
+    StructField,
+)
+from pyspark.testing.sqlutils import (
+    have_pandas,
+    have_pyarrow,
+    pandas_requirement_message,
+    pyarrow_requirement_message,
+    ReusedSQLTestCase,
+)
+from pyspark.util import PythonEvalType
+
+
+@unittest.skipIf(
+    not have_pandas or not have_pyarrow, pandas_requirement_message or pyarrow_requirement_message
+)
+class BinaryAsBytesUDFTests(ReusedSQLTestCase):
+    def test_arrow_batched_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.arrow.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.arrow.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_arrow_batched_udf_array_binary_type(self):
+        """Test SQL_ARROW_BATCHED_UDF with array of binary"""
+        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.arrow.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.arrow.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_arrow_batched_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.arrow.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.arrow.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_arrow_batched_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.arrow.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.arrow.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_binary_as_bytes_udf import *  # noqa: F401
+
+    try:
+        import xmlrunner  # type: ignore
+
+        testRunner = xmlrunner.XMLTestRunner(output="target/test-reports", verbosity=2)
+    except ImportError:
+        testRunner = None
+    unittest.main(testRunner=testRunner, verbosity=2)

Review Comment:
   ```suggestion
       unittest.main(testRunner=testRunner, verbosity=2)
    
   ```



-- 
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-53696][PYTHON]Default to bytes for BinaryType in PySpark arrow UDF [spark]

Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2387518563


##########
python/pyspark/sql/tests/arrow/test_arrow_binary_as_bytes_udf.py:
##########
@@ -0,0 +1,165 @@
+#

Review Comment:
   This package should be added in `dev/sparktestsupport/modules.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-53696][PYTHON]Default to bytes for BinaryType in PySpark UDF [spark]

Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2386406508


##########
python/pyspark/sql/tests/arrow/test_arrow_binary_as_bytes_udf.py:
##########
@@ -0,0 +1,145 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import unittest
+
+from pyspark.sql import Row
+from pyspark.sql.functions import udf, col
+from pyspark.sql.types import (
+    ArrayType,
+    BinaryType,
+    IntegerType,
+    MapType,
+    StringType,
+    StructType,
+    StructField,
+)
+from pyspark.testing.sqlutils import (
+    have_pandas,
+    have_pyarrow,
+    pandas_requirement_message,
+    pyarrow_requirement_message,
+    ReusedSQLTestCase,
+)
+from pyspark.util import PythonEvalType
+
+
+@unittest.skipIf(
+    not have_pandas or not have_pyarrow, pandas_requirement_message or pyarrow_requirement_message
+)
+class BinaryAsBytesUDFTests(ReusedSQLTestCase):
+    def test_arrow_batched_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.arrow.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.arrow.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_arrow_batched_udf_array_binary_type(self):
+        """Test SQL_ARROW_BATCHED_UDF with array of binary"""
+        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.arrow.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.arrow.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_arrow_batched_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.arrow.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.arrow.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_arrow_batched_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.arrow.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.arrow.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_binary_as_bytes_udf import *  # noqa: F401
+
+    try:
+        import xmlrunner  # type: ignore
+
+        testRunner = xmlrunner.XMLTestRunner(output="target/test-reports", verbosity=2)
+    except ImportError:
+        testRunner = None
+    unittest.main(testRunner=testRunner, verbosity=2)

Review Comment:
   ```suggestion
       unittest.main(testRunner=testRunner, verbosity=2)
    
   ```



-- 
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-53696][PYTHON][CONNECT][SQL] Default to bytes for BinaryType in PySpark [spark]

Posted by "xianzhe-databricks (via GitHub)" <gi...@apache.org>.
xianzhe-databricks commented on PR #52467:
URL: https://github.com/apache/spark/pull/52467#issuecomment-3422692908

   CI actually passed on retry.


-- 
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-53696][PYTHON]Default to bytes for BinaryType in PySpark [spark]

Posted by "xianzhe-databricks (via GitHub)" <gi...@apache.org>.
xianzhe-databricks commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2390280924


##########
python/pyspark/sql/tests/arrow/test_arrow_python_udtf.py:
##########
@@ -0,0 +1,57 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import unittest
+
+from pyspark.sql.types import BinaryType
+from pyspark.testing.sqlutils import (
+    have_pyarrow,
+    pyarrow_requirement_message,
+    ReusedSQLTestCase,
+)
+
+
+@unittest.skipIf(not have_pyarrow, pyarrow_requirement_message)
+class ArrowPythonUDTFTests(ReusedSQLTestCase):

Review Comment:
   Ah I see. will move the files there



-- 
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-53696][PYTHON][CONNECT][SQL]Default to bytes for BinaryType in PySpark [spark]

Posted by "xianzhe-databricks (via GitHub)" <gi...@apache.org>.
xianzhe-databricks commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2411738919


##########
python/pyspark/sql/connect/dataframe.py:
##########
@@ -1823,7 +1823,14 @@ 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."""
+        conf_value = self._session.conf.get("spark.sql.execution.pyspark.binaryAsBytes", "true")
+        return (conf_value or "true").lower() == "true"

Review Comment:
   addressed



-- 
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-53696][PYTHON][CONNECT][SQL] Default to bytes for BinaryType in PySpark [spark]

Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon closed pull request #52467: [SPARK-53696][PYTHON][CONNECT][SQL] Default to bytes for BinaryType in PySpark
URL: https://github.com/apache/spark/pull/52467


-- 
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-53696][PYTHON]Default to bytes for BinaryType in PySpark [spark]

Posted by "xianzhe-databricks (via GitHub)" <gi...@apache.org>.
xianzhe-databricks commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2398453183


##########
python/pyspark/sql/tests/test_udf.py:
##########
@@ -1446,6 +1447,95 @@ def my_udf():
 
                 self.spark.range(1).select(my_udf().alias("result")).show()
 
+    def test_udf_binary_type(self):
+        def get_binary_type(x):
+            return type(x).__name__
+
+        binary_udf = udf(get_binary_type, returnType="string")
+
+        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")

Review Comment:
   Ah make huge sense. will do



-- 
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-53696][PYTHON][CONNECT][SQL] Default to bytes for BinaryType in PySpark [spark]

Posted by "xianzhe-databricks (via GitHub)" <gi...@apache.org>.
xianzhe-databricks commented on PR #52467:
URL: https://github.com/apache/spark/pull/52467#issuecomment-3407020000

   I did another pass and fixed the linter error. The PR is good for merge from my point of view. @HyukjinKwon @ueshin @allisonwang-db thank you!


-- 
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-53696][PYTHON][CONNECT][SQL]Default to bytes for BinaryType in PySpark [spark]

Posted by "xianzhe-databricks (via GitHub)" <gi...@apache.org>.
xianzhe-databricks commented on PR #52467:
URL: https://github.com/apache/spark/pull/52467#issuecomment-3378567126

   @ueshin @Yicong-Huang do you have other concerns? Thanks a lot!
   I'll fix the linter once we are fine with the change on business code.


-- 
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-53696][PYTHON][CONNECT][SQL]Default to bytes for BinaryType in PySpark [spark]

Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2408849437


##########
python/pyspark/sql/conversion.py:
##########
@@ -735,7 +744,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,

Review Comment:
   maybe this should be controlled by flag?



##########
python/pyspark/sql/tests/connect/arrow/test_parity_arrow_python_udf.py:
##########
@@ -46,6 +48,14 @@ def tearDownClass(cls):
         finally:
             super().tearDownClass()
 
+    @unittest.skip("Duplicate test as it is already tested in ArrowPythonUDFLegacyTests.")
+    def test_udf_binary_type(self):
+        super().test_udf_binary_type(self)
+
+    @unittest.skip("Duplicate test as it is already tested in ArrowPythonUDFLegacyTests.")
+    def test_udf_binary_type_in_nested_structures(self):
+        super().test_udf_binary_type_in_nested_structures(self)
+

Review Comment:
   why do we add tests then skip them?



##########
python/pyspark/sql/connect/dataframe.py:
##########
@@ -1823,7 +1823,14 @@ 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."""
+        conf_value = self._session.conf.get("spark.sql.execution.pyspark.binaryAsBytes", "true")
+        return (conf_value or "true").lower() == "true"

Review Comment:
   for optional, it can only be "None" or a string right? maybe we can use
   
   ```
   con_value is not None and con_value.lower() == "true"
   ```
   to bypass it?



##########
python/docs/source/tutorial/sql/type_conversions.rst:
##########
@@ -67,6 +67,9 @@ 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.

Review Comment:
   maybe we can explain briefly what is the default behavior before 4.1.0?



-- 
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-53696][PYTHON]Default to bytes for BinaryType in PySpark [spark]

Posted by "ueshin (via GitHub)" <gi...@apache.org>.
ueshin commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2395897959


##########
python/pyspark/sql/tests/test_udf.py:
##########
@@ -1446,6 +1447,95 @@ def my_udf():
 
                 self.spark.range(1).select(my_udf().alias("result")).show()
 
+    def test_udf_binary_type(self):
+        def get_binary_type(x):
+            return type(x).__name__
+
+        binary_udf = udf(get_binary_type, returnType="string")
+
+        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")

Review Comment:
   Shall we use `assertDataFrameEqual`? Same for the following checks?



##########
python/pyspark/sql/tests/test_udtf.py:
##########
@@ -3063,6 +3078,22 @@ def tearDownClass(cls):
     not have_pandas or not have_pyarrow, pandas_requirement_message or pyarrow_requirement_message
 )
 class LegacyUDTFArrowTestsMixin(BaseUDTFTestsMixin):
+    def test_udtf_binary_type(self):
+        @udtf(returnType="type_name: string")
+        class BinaryTypeUDF:

Review Comment:
   ditto.



##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/EvaluatePython.scala:
##########
@@ -49,39 +56,52 @@ object EvaluatePython {
   /**
    * Helper for converting from Catalyst type to java type suitable for Pickle.
    */
-  def toJava(obj: Any, dataType: DataType): Any = (obj, dataType) match {
-    case (null, _) => null
-
-    case (row: InternalRow, struct: StructType) =>
-      val values = new Array[Any](row.numFields)
-      var i = 0
-      while (i < row.numFields) {
-        values(i) = toJava(row.get(i, struct.fields(i).dataType), struct.fields(i).dataType)
-        i += 1
-      }
-      new GenericRowWithSchema(values, struct)
+  def toJava(
+      obj: Any,
+      dataType: DataType,
+      binaryAsBytes: Boolean = SQLConf.get.pysparkBinaryAsBytes): Any = {

Review Comment:
   I guess we shouldn't use the default value for `binaryAsBytes` to see if we updated all places where it uses `toJava` to avoid unexpected behavior change?



##########
python/pyspark/sql/connect/dataframe.py:
##########
@@ -1823,7 +1823,14 @@ 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."""
+        conf_value = self._session.conf.get("spark.sql.execution.pyspark.binaryAsBytes", "true")
+        return (conf_value or "true").lower() == "true"

Review Comment:
   Which case is `(conf_value or "true")` for?



##########
python/pyspark/sql/tests/test_udtf.py:
##########
@@ -3044,6 +3044,21 @@ def eval(self, v1, v2, v3, v4):
         for idx, field in enumerate(result_df.schema.fields):
             self.assertEqual(field.dataType, expected_output_types[idx])
 
+    def test_udtf_binary_type(self):
+        @udtf(returnType="type_name: string")
+        class BinaryTypeUDF:

Review Comment:
   nit:
   ```suggestion
           class BinaryTypeUDTF:
   ```



-- 
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-53696][PYTHON]Default to bytes for BinaryType in PySpark [spark]

Posted by "xianzhe-databricks (via GitHub)" <gi...@apache.org>.
xianzhe-databricks commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2399008737


##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/EvaluatePython.scala:
##########
@@ -49,39 +56,52 @@ object EvaluatePython {
   /**
    * Helper for converting from Catalyst type to java type suitable for Pickle.
    */
-  def toJava(obj: Any, dataType: DataType): Any = (obj, dataType) match {
-    case (null, _) => null
-
-    case (row: InternalRow, struct: StructType) =>
-      val values = new Array[Any](row.numFields)
-      var i = 0
-      while (i < row.numFields) {
-        values(i) = toJava(row.get(i, struct.fields(i).dataType), struct.fields(i).dataType)
-        i += 1
-      }
-      new GenericRowWithSchema(values, struct)
+  def toJava(
+      obj: Any,
+      dataType: DataType,
+      binaryAsBytes: Boolean = SQLConf.get.pysparkBinaryAsBytes): Any = {

Review Comment:
   done



-- 
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-53696][PYTHON]Default to bytes for BinaryType in PySpark arrow UDF [spark]

Posted by "allisonwang-db (via GitHub)" <gi...@apache.org>.
allisonwang-db commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2389381290


##########
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**|bytearray<br/>**Note:** When Arrow is enabled (`spark.sql.execution.arrow.pyspark.enabled=true`), BinaryType maps to `bytes` instead of `bytearray`.|BinaryType()|
 |**BooleanType**|bool|BooleanType()|

Review Comment:
   And this also requires this config `spark.sql.execution.arrow.pyspark.binaryAsBytes` to be enabled right? I think this can be confusing. If we enabled both configs `spark.sql.execution.arrow.pyspark.enabled` and `spark.sql.execution.arrow.pyspark.binaryAsBytes` by default in Spark 4.1, then can we directly change the documentation here to `bytes`? You can mention this is a behavior change in Spark 4.1.



-- 
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-53696][PYTHON][CONNECT][SQL]Default to bytes for BinaryType in PySpark [spark]

Posted by "xianzhe-databricks (via GitHub)" <gi...@apache.org>.
xianzhe-databricks commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2411748646


##########
python/pyspark/sql/conversion.py:
##########
@@ -735,7 +744,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,

Review Comment:
   it is already controlled by a flag, as `binary_as_bytes` is passed at caller's place as the value of the SQL conf `spark.sql.execution.pyspark.binaryAsBytes`.
   
   It is not possible, or against the style, to access the SQL conf in this conversion.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-53696][PYTHON]Default to bytes for BinaryType in PySpark arrow UDF [spark]

Posted by "ueshin (via GitHub)" <gi...@apache.org>.
ueshin commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2389406979


##########
python/pyspark/sql/tests/arrow/test_arrow_python_udtf.py:
##########
@@ -0,0 +1,57 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import unittest
+
+from pyspark.sql.types import BinaryType
+from pyspark.testing.sqlutils import (
+    have_pyarrow,
+    pyarrow_requirement_message,
+    ReusedSQLTestCase,
+)
+
+
+@unittest.skipIf(not have_pyarrow, pyarrow_requirement_message)
+class ArrowPythonUDTFTests(ReusedSQLTestCase):

Review Comment:
   Python UDTF tests should be in `test_udtf` that includes Arrow-optimized Python UDTF tests?



-- 
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-53696][PYTHON][CONNECT][SQL] Default to bytes for BinaryType in PySpark [spark]

Posted by "xianzhe-databricks (via GitHub)" <gi...@apache.org>.
xianzhe-databricks commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2435058016


##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3724,6 +3724,19 @@ object SQLConf {
       .booleanConf
       .createWithDefault(false)
 
+  val PYSPARK_BINARY_AS_BYTES =
+    buildConf("spark.sql.execution.pyspark.binaryAsBytes")
+      .doc("When true, BinaryType is mapped consistently to bytes in PySpark." +

Review Comment:
   exactly



-- 
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-53696][PYTHON][CONNECT][SQL] Default to bytes for BinaryType in PySpark [spark]

Posted by "xianzhe-databricks (via GitHub)" <gi...@apache.org>.
xianzhe-databricks commented on PR #52467:
URL: https://github.com/apache/spark/pull/52467#issuecomment-3422296521

   CI fails on something that seems unrelated/transient. Need to re-trigger CI


-- 
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-53696][PYTHON][CONNECT][SQL] Default to bytes for BinaryType in PySpark [spark]

Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #52467:
URL: https://github.com/apache/spark/pull/52467#issuecomment-3423994175

   Merged to master.


-- 
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-53696][PYTHON]Default to bytes for BinaryType in PySpark UDF [spark]

Posted by "xianzhe-databricks (via GitHub)" <gi...@apache.org>.
xianzhe-databricks commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2382690983


##########
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,

Review Comment:
   the default value for `int_to_decimal_coercion_enabled` is not used at all



-- 
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-53696][PYTHON]Default to bytes for BinaryType in PySpark [spark]

Posted by "xianzhe-databricks (via GitHub)" <gi...@apache.org>.
xianzhe-databricks commented on PR #52467:
URL: https://github.com/apache/spark/pull/52467#issuecomment-3352657354

   > Can we also add this change to the migration guide?
   
   where shall I add it? I found the pyspark migration guide is archived https://github.com/apache/spark/blob/master/docs/pyspark-migration-guide.md


-- 
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-53696][PYTHON][CONNECT][SQL]Default to bytes for BinaryType in PySpark [spark]

Posted by "xianzhe-databricks (via GitHub)" <gi...@apache.org>.
xianzhe-databricks commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2411792079


##########
python/docs/source/tutorial/sql/type_conversions.rst:
##########
@@ -67,6 +67,9 @@ 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.

Review Comment:
   I end up explaining more in the definition of `spark.sql.execution.pyspark.binaryAsBytes` in the SQL conf as it is really long...



-- 
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-53696][PYTHON][CONNECT][SQL]Default to bytes for BinaryType in PySpark [spark]

Posted by "xianzhe-databricks (via GitHub)" <gi...@apache.org>.
xianzhe-databricks commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2408669668


##########
python/pyspark/sql/tests/connect/test_connect_collection.py:
##########
@@ -25,6 +26,38 @@
 
 
 class SparkConnectCollectionTests(ReusedMixedTestCase, PandasOnSparkTestUtils):
+    def connect_conf(self, conf_dict):

Review Comment:
   for sure!



-- 
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-53696][PYTHON][CONNECT][SQL] Default to bytes for BinaryType in PySpark [spark]

Posted by "xianzhe-databricks (via GitHub)" <gi...@apache.org>.
xianzhe-databricks commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2435248776


##########
python/docs/source/migration_guide/pyspark_upgrade.rst:
##########
@@ -29,6 +29,17 @@ Upgrading from PySpark 4.0 to 4.1
 * In Spark 4.1, Arrow-optimized Python UDF supports UDT input / output instead of falling back to the regular UDF. To restore the legacy behavior, set ``spark.sql.execution.pythonUDF.arrow.legacy.fallbackOnUDT`` to ``true``.
 * In Spark 4.1, unnecessary conversion to pandas instances is removed when ``spark.sql.execution.pythonUDF.arrow.enabled`` is enabled. As a result, the type coercion changes when the produced output has a schema different from the specified schema. To restore the previous behavior, enable ``spark.sql.legacy.execution.pythonUDF.pandas.conversion.enabled``.
 * In Spark 4.1, unnecessary conversion to pandas instances is removed when ``spark.sql.execution.pythonUDTF.arrow.enabled`` is enabled. As a result, the type coercion changes when the produced output has a schema different from the specified schema. To restore the previous behavior, enable ``spark.sql.legacy.execution.pythonUDTF.pandas.conversion.enabled``.
+* In Spark 4.1, the data type ``BinaryType`` is mapped to Python ``bytes`` consistently in PySpark.
+  To restore the previous behavior, set ``spark.sql.execution.pyspark.binaryAsBytes`` to ``true``. The behavior before Spark 4.1.0 is illustrated in the following table:
+
+    ===============================================================================  ==============================
+    Case                                                                             Python type for ``BinaryType``
+    ===============================================================================  ==============================
+    Regular UDF and UDTF without Arrow optimization                                  ``bytearray``
+    DataFrame APIs (both Spark Classic and Spark Connect)                            ``bytearray``
+    Data sources                                                                     ``bytearray``
+    Arrow-optimized UDF and UDTF with unnecessary conversion to pandas instances     ``bytes``
+    ===============================================================================  ==============================

Review Comment:
   @ueshin @allisonwang-db migration guide is added!



-- 
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-53696][PYTHON]Default to bytes for BinaryType in PySpark arrow UDF [spark]

Posted by "xianzhe-databricks (via GitHub)" <gi...@apache.org>.
xianzhe-databricks commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2387721501


##########
python/pyspark/sql/tests/arrow/test_arrow_binary_as_bytes_udf.py:
##########
@@ -0,0 +1,165 @@
+#

Review Comment:
   Thanks! I actually found we don't have any test for `SQL_ARROW_TABLE_UDF`. I created a file `test_arrow_python_udtf.py` for that and add it to the modules.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-53696][PYTHON][CONNECT][SQL] Default to bytes for BinaryType in PySpark [spark]

Posted by "allisonwang-db (via GitHub)" <gi...@apache.org>.
allisonwang-db commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2433656755


##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3724,6 +3724,19 @@ object SQLConf {
       .booleanConf
       .createWithDefault(false)
 
+  val PYSPARK_BINARY_AS_BYTES =
+    buildConf("spark.sql.execution.pyspark.binaryAsBytes")
+      .doc("When true, BinaryType is mapped consistently to bytes in PySpark." +

Review Comment:
   "Mapped" here means the function input will be mapped as bytes right? 



##########
python/pyspark/sql/tests/test_udtf.py:
##########
@@ -3044,6 +3044,19 @@ def eval(self, v1, v2, v3, v4):
         for idx, field in enumerate(result_df.schema.fields):
             self.assertEqual(field.dataType, expected_output_types[idx])
 
+    def test_udtf_binary_type(self):

Review Comment:
   Can we also add one more test for test_arrow_udtf? 



-- 
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-53696][PYTHON][CONNECT][SQL] Default to bytes for BinaryType in PySpark [spark]

Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #52467:
URL: https://github.com/apache/spark/pull/52467#issuecomment-3404418229

   Let's fix up the linter failure and merge


-- 
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-53696][PYTHON]Default to bytes for BinaryType in PySpark [spark]

Posted by "xianzhe-databricks (via GitHub)" <gi...@apache.org>.
xianzhe-databricks commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2390358418


##########
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**|bytearray<br/>**Note:** When Arrow is enabled (`spark.sql.execution.arrow.pyspark.enabled=true`), BinaryType maps to `bytes` instead of `bytearray`.|BinaryType()|
 |**BooleanType**|bool|BooleanType()|

Review Comment:
   make sense. And to keep consistency, I actually should change the data type for all UDFs (including the non-arrow regular UDF)



-- 
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-53696][PYTHON][CONNECT][SQL]Default to bytes for BinaryType in PySpark [spark]

Posted by "xianzhe-databricks (via GitHub)" <gi...@apache.org>.
xianzhe-databricks commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2411744711


##########
python/pyspark/sql/tests/connect/arrow/test_parity_arrow_python_udf.py:
##########
@@ -46,6 +48,14 @@ def tearDownClass(cls):
         finally:
             super().tearDownClass()
 
+    @unittest.skip("Duplicate test as it is already tested in ArrowPythonUDFLegacyTests.")
+    def test_udf_binary_type(self):
+        super().test_udf_binary_type(self)
+
+    @unittest.skip("Duplicate test as it is already tested in ArrowPythonUDFLegacyTests.")
+    def test_udf_binary_type_in_nested_structures(self):
+        super().test_udf_binary_type_in_nested_structures(self)
+

Review Comment:
   class `ArrowPythonUDFParityLegacyTestsMixin` extends from `ArrowPythonUDFTestsMixin` and `ArrowPythonUDFTestsMixin` already contains this test.
   We already run this test in test_arrow_python_udf.py. It is not meaningful to re-run a test. We can marginally save some test resources.



-- 
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-53696][PYTHON][CONNECT][SQL] Default to bytes for BinaryType in PySpark [spark]

Posted by "xianzhe-databricks (via GitHub)" <gi...@apache.org>.
xianzhe-databricks commented on code in PR #52467:
URL: https://github.com/apache/spark/pull/52467#discussion_r2435056619


##########
python/pyspark/sql/tests/test_udtf.py:
##########
@@ -3063,6 +3076,19 @@ def tearDownClass(cls):
     not have_pandas or not have_pyarrow, pandas_requirement_message or pyarrow_requirement_message
 )
 class LegacyUDTFArrowTestsMixin(BaseUDTFTestsMixin):
+    def test_udtf_binary_type(self):
+        @udtf(returnType="type_name: string")
+        class BinaryTypeUDTF:
+            def eval(self, b):
+                # Check the type of the binary input and return type name as string
+                yield (type(b).__name__,)
+
+        # For Arrow Python UDTF with legacy conversion BinaryType is always mapped to bytes
+        for conf_value in ["true", "false"]:
+            with self.sql_conf({"spark.sql.execution.pyspark.binaryAsBytes": conf_value}):
+                result = BinaryTypeUDTF(lit(b"test")).collect()
+                self.assertEqual(result[0]["type_name"], "bytes")

Review Comment:
   @allisonwang-db arrow udtf with legacy conversion is tested here



##########
python/pyspark/sql/tests/test_udtf.py:
##########
@@ -3389,6 +3415,11 @@ def tearDownClass(cls):
 
 
 class UDTFArrowTestsMixin(LegacyUDTFArrowTestsMixin):
+    def test_udtf_binary_type(self):
+        # For Arrow Python UDTF with non-legacy conversionBinaryType is mapped to
+        # bytes or bytearray consistently with non-Arrow Python UDTF behavior.
+        BaseUDTFTestsMixin.test_udtf_binary_type(self)

Review Comment:
   @allisonwang-db arrow udtf without legacy conversion is tested here



-- 
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