This is an automated email from the ASF dual-hosted git repository.
dtenedor pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new fd4d81687402 [SPARK-59275][PYTHON] Reject CHAR/VARCHAR at Python/Arrow
host boundaries
fd4d81687402 is described below
commit fd4d81687402eeb00703c1d817cf6de5b876ffb8
Author: Serge Rielau <[email protected]>
AuthorDate: Thu Oct 8 15:44:10 2026 -0700
[SPARK-59275][PYTHON] Reject CHAR/VARCHAR at Python/Arrow host boundaries
### What changes were proposed in this pull request?
Reject CHAR/VARCHAR at Python/Arrow host boundaries with one recursive,
classed error.
CHAR/VARCHAR stay SQL types. Host APIs must declare STRING.
- Add `_check_no_char_varchar` and raise
`CHAR_VARCHAR_NOT_SUPPORTED_IN_PYTHON`.
- Apply it to Python UDF return types (all eval types), UDAF buffer
schemas, and UDTF return types on Classic and Connect.
- Reject the same schemas on the JVM `UserDefinedPythonFunction` /
`UserDefinedPythonTableFunction` builders so Connect or direct construction
cannot skip the client check.
- Recurse through struct, array, map, and CHAR/VARCHAR hidden in UDT
storage (Python `_has_type` and JVM `hasCharVarcharIncludingUDT`).
- Reject analyze-derived Python UDTF schemas after the worker returns the
schema.
- Split JVM return vs buffer diagnostics so a buffer-only failure reports
`Python UDAF buffer schemas` and the buffer type.
This PR does not add CHAR/VARCHAR assignment for Python UDFs, Arrow,
pandas, `createDataFrame`, or `toArrow`. SQL assignment and Connect schema
transport (SPARK-59276) are unchanged. No new configuration is added.
JIRA: https://issues.apache.org/jira/browse/SPARK-59275
### Why are the changes needed?
Python/Arrow have only unconstrained strings. Admitting CHAR/VARCHAR in a
shared mapper requires either full write-side assignment on every consumer or a
per-protocol rejection matrix. That ballooned earlier SPARK-59275 work without
improving SQL semantics.
Fail-closed host APIs keep SQL CHAR/VARCHAR correct, and they fail with a
stable error instead of unchecked values or `arrowDataTypeMismatchError`.
### Does this PR introduce _any_ user-facing change?
Yes. Declaring CHAR/VARCHAR as a Python UDF return type, UDAF buffer
schema, or UDTF return type (fixed or analyze-derived) now fails before
execution with `CHAR_VARCHAR_NOT_SUPPORTED_IN_PYTHON`, including nested types
and UDT storage. Previously some paths failed later with a generic
unsupported-type or JVM message, and some pickled UDF paths could run without
SQL length checks.
The error text tells callers to declare STRING. SQL CHAR/VARCHAR assignment
(`INSERT`, `CAST`, table schemas) is unchanged. SPARK-59276 Connect
local-relation CHAR/VARCHAR schema transport is unchanged.
### How was this patch tested?
- Updated Classic UDF CHAR/VARCHAR return-type tests to assert the new
error class, including nested array/map/struct DDL.
- Updated the Arrow-optimized UDF `VarcharType` rejection test.
- Added `_check_no_char_varchar` coverage for scalar, nested, and
UDT-storage types.
- Added UDTF CHAR rejection for StructType, Classic parsed DDL, Connect
server-parsed DDL, and analyze-derived schemas (Classic/Connect parity mixin).
- Added Classic/Connect incremental Aggregator bufferSchema and outputType
rejection tests.
- Added JVM `PythonCharVarcharRejectSuite` for builder return, buffer, UDT
storage, and UDTF checks.
`PythonCharVarcharRejectSuite` passes locally. Python `py_compile` on the
changed files passes. Focused PySpark runtime suites were not run locally in
this environment; CI will exercise them.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor Auto (Grok 4.6)
Closes #58549 from srielau/serge-rielau_data/SPARK-59275.
Authored-by: Serge Rielau <[email protected]>
Signed-off-by: Daniel Tenedorio <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 6 ++
python/pyspark/errors/error-conditions.json | 5 +
python/pyspark/sql/connect/udf.py | 14 ++-
python/pyspark/sql/connect/udtf.py | 5 +-
.../tests/arrow/test_arrow_python_aggregator.py | 69 ++++++++++++-
.../sql/tests/arrow/test_arrow_python_udf.py | 5 +-
python/pyspark/sql/tests/test_types.py | 45 +++++++++
python/pyspark/sql/tests/test_udf.py | 82 +++++++++++++---
.../pyspark/sql/tests/test_udf_transpile_unit.py | 36 +++++++
python/pyspark/sql/tests/test_udtf.py | 57 +++++++++++
python/pyspark/sql/types.py | 13 +++
python/pyspark/sql/udf.py | 12 +++
python/pyspark/sql/udtf.py | 3 +-
.../spark/sql/catalyst/util/CharVarcharUtils.scala | 12 +++
.../spark/sql/errors/QueryCompilationErrors.scala | 6 ++
.../python/UserDefinedPythonFunction.scala | 20 ++++
.../python/PythonCharVarcharRejectSuite.scala | 109 +++++++++++++++++++++
17 files changed, 482 insertions(+), 17 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index a58631f31bce..3a796a031afa 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -1047,6 +1047,12 @@
},
"sqlState" : "XX000"
},
+ "CHAR_VARCHAR_NOT_SUPPORTED_IN_PYTHON" : {
+ "message" : [
+ "CHAR/VARCHAR is not supported in <feature>. Use STRING in the
declaration. Data type: <data_type>."
+ ],
+ "sqlState" : "0A000"
+ },
"CHECKPOINT_DIRECTORY_NOT_SET" : {
"message" : [
"Cannot checkpoint because no checkpoint directory is configured.",
diff --git a/python/pyspark/errors/error-conditions.json
b/python/pyspark/errors/error-conditions.json
index d60a4c7e3c05..5a1442cf5f17 100644
--- a/python/pyspark/errors/error-conditions.json
+++ b/python/pyspark/errors/error-conditions.json
@@ -154,6 +154,11 @@
"Cannot <condition1> without <condition2>."
]
},
+ "CHAR_VARCHAR_NOT_SUPPORTED_IN_PYTHON": {
+ "message": [
+ "CHAR/VARCHAR is not supported in <feature>. Use STRING in the
declaration. Data type: <data_type>."
+ ]
+ },
"CLASSIC_OPERATION_NOT_SUPPORTED_ON_DF": {
"message": [
"Calling property or member '<member>' is not supported in PySpark
Classic, please use Spark Connect instead."
diff --git a/python/pyspark/sql/connect/udf.py
b/python/pyspark/sql/connect/udf.py
index 9eb2bf5e0510..cc7f26f4dd9e 100644
--- a/python/pyspark/sql/connect/udf.py
+++ b/python/pyspark/sql/connect/udf.py
@@ -33,7 +33,13 @@ from pyspark.sql.connect.expressions import (
PythonUDF,
)
from pyspark.sql.pandas.utils import require_minimum_pandas_version,
require_minimum_pyarrow_version
-from pyspark.sql.types import DataType, StringType, StructType,
_parse_datatype_string
+from pyspark.sql.types import (
+ DataType,
+ StringType,
+ StructType,
+ _check_no_char_varchar,
+ _parse_datatype_string,
+)
from pyspark.sql.udf import (
UDFRegistration as PySparkUDFRegistration,
)
@@ -191,6 +197,12 @@ class UserDefinedFunction:
# (see :class:`pyspark.sql.aggregator.Aggregator`); ``None``
otherwise. A first-class field
# so it survives ``_wrapped()``, ``asNondeterministic()`` and
``spark.udf.register``.
self.bufferSchema = bufferSchema
+ if isinstance(returnType, DataType):
+ # Constrained SQL strings are rejected at construction. Eval-type
Arrow/Pandas
+ # conversion stays on ``returnType`` so unrelated types fail only
when consumed.
+ _check_no_char_varchar(returnType, "Python UDF return types")
+ if bufferSchema is not None:
+ _check_no_char_varchar(bufferSchema, "Python UDAF buffer schemas")
@property
def returnType(self) -> DataType:
diff --git a/python/pyspark/sql/connect/udtf.py
b/python/pyspark/sql/connect/udtf.py
index aa7d86c0f72d..89496d083621 100644
--- a/python/pyspark/sql/connect/udtf.py
+++ b/python/pyspark/sql/connect/udtf.py
@@ -32,7 +32,7 @@ from pyspark.sql.connect.table_arg import TableArg
from pyspark.sql.connect.types import UnparsedDataType
from pyspark.sql.connect.utils import get_python_ver
from pyspark.sql.pandas.utils import require_minimum_pandas_version,
require_minimum_pyarrow_version
-from pyspark.sql.types import DataType, StructType
+from pyspark.sql.types import DataType, StructType, _check_no_char_varchar
from pyspark.sql.udtf import AnalyzeArgument, AnalyzeResult,
_validate_udtf_handler # noqa: F401
from pyspark.sql.udtf import UDTFRegistration as PySparkUDTFRegistration
from pyspark.util import PythonEvalType
@@ -155,6 +155,9 @@ class UserDefinedTableFunction:
) -> None:
_validate_udtf_handler(func, returnType)
+ if isinstance(returnType, DataType):
+ _check_no_char_varchar(returnType, "Python UDTF return types")
+
self.func = func
self.returnType: Optional[DataType] = (
None
diff --git a/python/pyspark/sql/tests/arrow/test_arrow_python_aggregator.py
b/python/pyspark/sql/tests/arrow/test_arrow_python_aggregator.py
index 96b16aea9a99..2d745252c497 100644
--- a/python/pyspark/sql/tests/arrow/test_arrow_python_aggregator.py
+++ b/python/pyspark/sql/tests/arrow/test_arrow_python_aggregator.py
@@ -17,9 +17,10 @@
import unittest
from decimal import Decimal
-from pyspark.errors import AnalysisException, PySparkValueError
+from pyspark.errors import AnalysisException, PySparkNotImplementedError,
PySparkValueError
from pyspark.sql import functions as sf
from pyspark.sql.types import (
+ CharType,
DecimalType,
DoubleType,
LongType,
@@ -242,6 +243,72 @@ class ArrowPythonAggregatorTestsMixin:
exp = {r["k"]: r["m"] for r in expected}
self.assertEqual(got, exp)
+ def test_char_varchar_buffer_schema_rejected(self):
+ class BadBuffer(Aggregator):
+ @property
+ def bufferSchema(self):
+ return StructType([StructField("s", CharType(3))])
+
+ @property
+ def outputType(self):
+ return DoubleType()
+
+ def zero(self):
+ return ("",)
+
+ def reduce(self, buffer, value):
+ return buffer
+
+ def merge(self, b1, b2):
+ return b1
+
+ def finish(self, buffer):
+ return 0.0
+
+ with self.assertRaises(PySparkNotImplementedError) as pe:
+ udaf(BadBuffer())
+ self.check_error(
+ exception=pe.exception,
+ errorClass="CHAR_VARCHAR_NOT_SUPPORTED_IN_PYTHON",
+ messageParameters={
+ "feature": "Python UDAF buffer schemas",
+ "data_type": "struct<s:char(3)>",
+ },
+ )
+
+ def test_char_varchar_output_type_rejected(self):
+ class BadOutput(Aggregator):
+ @property
+ def bufferSchema(self):
+ return StructType([StructField("n", LongType())])
+
+ @property
+ def outputType(self):
+ return CharType(3)
+
+ def zero(self):
+ return (0,)
+
+ def reduce(self, buffer, value):
+ return buffer
+
+ def merge(self, b1, b2):
+ return b1
+
+ def finish(self, buffer):
+ return "a"
+
+ with self.assertRaises(PySparkNotImplementedError) as pe:
+ udaf(BadOutput())
+ self.check_error(
+ exception=pe.exception,
+ errorClass="CHAR_VARCHAR_NOT_SUPPORTED_IN_PYTHON",
+ messageParameters={
+ "feature": "Python UDF return types",
+ "data_type": "char(3)",
+ },
+ )
+
def test_named_arguments(self):
# A named argument (both DataFrame and SQL forms) must feed the
aggregator's value tuple,
# not be silently dropped.
diff --git a/python/pyspark/sql/tests/arrow/test_arrow_python_udf.py
b/python/pyspark/sql/tests/arrow/test_arrow_python_udf.py
index e56fc33ae163..417f1082cdc4 100644
--- a/python/pyspark/sql/tests/arrow/test_arrow_python_udf.py
+++ b/python/pyspark/sql/tests/arrow/test_arrow_python_udf.py
@@ -276,9 +276,10 @@ class ArrowPythonUDFTestsMixin(BaseUDFTestsMixin):
self.check_error(
exception=pe.exception,
- errorClass="NOT_IMPLEMENTED",
+ errorClass="CHAR_VARCHAR_NOT_SUPPORTED_IN_PYTHON",
messageParameters={
- "feature": "Invalid return type with Arrow-optimized Python
UDF: VarcharType(10)"
+ "feature": "Python UDF return types",
+ "data_type": "varchar(10)",
},
)
diff --git a/python/pyspark/sql/tests/test_types.py
b/python/pyspark/sql/tests/test_types.py
index 1b477ef5d5e2..5db16c9bbc2a 100644
--- a/python/pyspark/sql/tests/test_types.py
+++ b/python/pyspark/sql/tests/test_types.py
@@ -76,6 +76,7 @@ from pyspark.sql.types import (
_array_signed_int_typecode_ctype_mappings,
_array_type_mappings,
_array_unsigned_int_typecode_ctype_mappings,
+ _check_no_char_varchar,
_create_row,
_infer_type,
_make_type_verifier,
@@ -659,6 +660,50 @@ class TypesTestsMixin:
self.assertEqual(StringType("UTF8_LCASE").simpleString(), "string
collate UTF8_LCASE")
self.assertEqual(StringType("UNICODE").simpleString(), "string collate
UNICODE")
+ def test_check_no_char_varchar(self):
+ from pyspark.testing.objects import ExamplePointUDT
+
+ _check_no_char_varchar(StringType(), "Python UDF return types")
+ _check_no_char_varchar(ArrayType(StringType()), "Python UDF return
types")
+
+ cases = [
+ CharType(3),
+ VarcharType(8),
+ ArrayType(CharType(5)),
+ MapType(VarcharType(4), StringType()),
+ MapType(StringType(), VarcharType(4)),
+ MapType(StringType(), ArrayType(CharType(3))),
+ StructType([StructField("c", CharType(3))]),
+ ]
+ for dt in cases:
+ with self.subTest(dt=dt.simpleString()):
+ with self.assertRaises(PySparkNotImplementedError) as pe:
+ _check_no_char_varchar(dt, "Python UDF return types")
+ self.check_error(
+ exception=pe.exception,
+ errorClass="CHAR_VARCHAR_NOT_SUPPORTED_IN_PYTHON",
+ messageParameters={
+ "feature": "Python UDF return types",
+ "data_type": dt.simpleString(),
+ },
+ )
+
+ class CharStorageUDT(ExamplePointUDT):
+ @classmethod
+ def sqlType(cls):
+ return StructType([StructField("x", CharType(3))])
+
+ with self.assertRaises(PySparkNotImplementedError) as pe:
+ _check_no_char_varchar(CharStorageUDT(), "Python UDF return types")
+ self.check_error(
+ exception=pe.exception,
+ errorClass="CHAR_VARCHAR_NOT_SUPPORTED_IN_PYTHON",
+ messageParameters={
+ "feature": "Python UDF return types",
+ "data_type": CharStorageUDT().simpleString(),
+ },
+ )
+
def test_char_varchar_type_collations(self):
from pyspark.sql.types import _parse_datatype_json_string
diff --git a/python/pyspark/sql/tests/test_udf.py
b/python/pyspark/sql/tests/test_udf.py
index b1a7f9214fbb..9fd365f131e1 100644
--- a/python/pyspark/sql/tests/test_udf.py
+++ b/python/pyspark/sql/tests/test_udf.py
@@ -27,7 +27,12 @@ import time
import unittest
from contextlib import redirect_stdout
-from pyspark.errors import AnalysisException, PySparkTypeError, PythonException
+from pyspark.errors import (
+ AnalysisException,
+ PySparkNotImplementedError,
+ PySparkTypeError,
+ PythonException,
+)
from pyspark.logger import PySparkLogger
from pyspark.sql import Column, Row, SparkSession
from pyspark.sql.functions import assert_true, col, lit, rand, udf
@@ -35,6 +40,7 @@ from pyspark.sql.types import (
ArrayType,
BinaryType,
BooleanType,
+ CharType,
DayTimeIntervalType,
DoubleType,
IntegerType,
@@ -46,8 +52,10 @@ from pyspark.sql.types import (
TimestampNTZType,
VariantType,
VariantVal,
+ _parse_datatype_string,
)
from pyspark.sql.udf import UserDefinedFunction
+from pyspark.sql.utils import is_remote
from pyspark.testing.objects import ExamplePoint, ExamplePointUDT
from pyspark.testing.sqlutils import (
ReusedSQLTestCase,
@@ -55,7 +63,7 @@ from pyspark.testing.sqlutils import (
test_not_compiled_message,
)
from pyspark.testing.utils import assertDataFrameEqual, eventually, timeout
-from pyspark.util import is_remote_only
+from pyspark.util import PythonEvalType, is_remote_only
class BaseUDFTestsMixin:
@@ -1468,6 +1476,7 @@ class BaseUDFTestsMixin:
(array_with_char_type, array_with_char_type_value),
(array_with_varchar_type, array_with_varchar_value),
(map_type, map_value),
+ (f"map<string, {varchar_type}>", {"a": "b"}),
(struct_type, struct_value),
(
f"struct<f1: {array_with_char_type}, f2:
{array_with_varchar_type}, "
@@ -1483,17 +1492,68 @@ class BaseUDFTestsMixin:
]
for return_type, return_value in pairs:
- with self.assertRaisesRegex(
- Exception,
- "(Please use a different output data type for your UDF or
DataFrame|"
- "Invalid return type with Arrow-optimized Python UDF)",
- ):
+ with self.subTest(return_type=return_type):
+ with self.assertRaises(PySparkNotImplementedError) as pe:
+
+ @udf(return_type)
+ def my_udf():
+ return return_value
+
+ self.spark.range(1).select(my_udf().alias("result"))
+
+ parsed = _parse_datatype_string(return_type)
+ self.check_error(
+ exception=pe.exception,
+ errorClass="CHAR_VARCHAR_NOT_SUPPORTED_IN_PYTHON",
+ messageParameters={
+ "feature": "Python UDF return types",
+ "data_type": parsed.simpleString(),
+ },
+ )
+
+ def test_char_varchar_rejected_at_udf_construction(self):
+ if is_remote():
+ from pyspark.sql.connect.udf import UserDefinedFunction as UDF
+ else:
+ from pyspark.sql.udf import UserDefinedFunction as UDF
+
+ with self.assertRaises(PySparkNotImplementedError) as pe:
+ UDF(lambda x: x, CharType(3))
+ self.check_error(
+ exception=pe.exception,
+ errorClass="CHAR_VARCHAR_NOT_SUPPORTED_IN_PYTHON",
+ messageParameters={
+ "feature": "Python UDF return types",
+ "data_type": "char(3)",
+ },
+ )
- @udf(return_type)
- def my_udf():
- return return_value
+ def test_eval_type_return_check_is_lazy(self):
+ if is_remote():
+ from pyspark.sql.connect.udf import UserDefinedFunction as UDF
+ else:
+ from pyspark.sql.udf import UserDefinedFunction as UDF
- self.spark.range(1).select(my_udf().alias("result")).show()
+ # Grouped-map Pandas UDFs require a StructType. Construction must
succeed for STRING;
+ # the eval-type check runs only when returnType is consumed.
+ udf_obj = UDF(
+ lambda x: x,
+ StringType(),
+ evalType=PythonEvalType.SQL_GROUPED_MAP_PANDAS_UDF,
+ )
+ with self.assertRaises(PySparkTypeError) as pe:
+ _ = udf_obj.returnType
+ self.check_error(
+ exception=pe.exception,
+ errorClass="INVALID_RETURN_TYPE_FOR_PANDAS_UDF",
+ messageParameters={
+ "eval_type": (
+ "SQL_GROUPED_MAP_PANDAS_UDF or
SQL_GROUPED_MAP_PANDAS_ITER_UDF or "
+ "SQL_GROUPED_MAP_PANDAS_UDF_WITH_STATE"
+ ),
+ "return_type": str(StringType()),
+ },
+ )
def test_udf_binary_type(self):
def get_binary_type(x):
diff --git a/python/pyspark/sql/tests/test_udf_transpile_unit.py
b/python/pyspark/sql/tests/test_udf_transpile_unit.py
index dce2e9eeefc4..6a650a08a767 100644
--- a/python/pyspark/sql/tests/test_udf_transpile_unit.py
+++ b/python/pyspark/sql/tests/test_udf_transpile_unit.py
@@ -179,6 +179,42 @@ class UDFTranspileUnitTests(ReusedSQLTestCase):
"when both transpilePyUDFs and ANSI mode are enabled",
)
+ def test_udf_transpile_does_not_mask_char_varchar_ddl(self):
+ # DDL CHAR/VARCHAR is parsed via ``returnType`` inside the optional
+ # transpilation try. That structured error must not become a
+ # transpilation UserWarning (which warnings-as-errors would raise).
+ import warnings
+
+ from pyspark.errors import PySparkNotImplementedError
+
+ def identity(x):
+ return x
+
+ def _assert_char_varchar(return_type, data_type):
+ with self.assertRaises(PySparkNotImplementedError) as pe:
+ UserDefinedFunction(identity, return_type)
+ self.check_error(
+ exception=pe.exception,
+ errorClass="CHAR_VARCHAR_NOT_SUPPORTED_IN_PYTHON",
+ messageParameters={
+ "feature": "Python UDF return types",
+ "data_type": data_type,
+ },
+ )
+
+ with self.sql_conf(_TRANSPILE_ON):
+ with warnings.catch_warnings(record=True) as caught:
+ warnings.simplefilter("always")
+ _assert_char_varchar("char(3)", "char(3)")
+ self.assertFalse(
+ any("Exception transpiling" in str(w.message) for w in caught),
+ "CHAR/VARCHAR DDL must not be reported as a transpilation
failure",
+ )
+
+ with warnings.catch_warnings():
+ warnings.simplefilter("error")
+ _assert_char_varchar("varchar(8)", "varchar(8)")
+
def test_udf_transpile_falls_back_for_unsupported_patterns(self):
# The transpiler intentionally only handles a small subset of
# Python AST today. Everything outside that subset must
diff --git a/python/pyspark/sql/tests/test_udtf.py
b/python/pyspark/sql/tests/test_udtf.py
index 10641b0769a0..5e3df713ca3c 100644
--- a/python/pyspark/sql/tests/test_udtf.py
+++ b/python/pyspark/sql/tests/test_udtf.py
@@ -30,6 +30,7 @@ from pyspark.errors import (
AnalysisException,
IllegalArgumentException,
PySparkAttributeError,
+ PySparkNotImplementedError,
PySparkPicklingError,
PySparkTypeError,
PythonException,
@@ -53,6 +54,7 @@ from pyspark.sql.functions import (
from pyspark.sql.types import (
ArrayType,
BooleanType,
+ CharType,
DataType,
IntegerType,
LongType,
@@ -64,6 +66,7 @@ from pyspark.sql.types import (
StructType,
VariantVal,
)
+from pyspark.sql.utils import is_remote
from pyspark.testing import assertDataFrameEqual, assertSchemaEqual
from pyspark.testing.objects import ExamplePoint, ExamplePointUDT
from pyspark.testing.sqlutils import ReusedSQLTestCase
@@ -85,6 +88,60 @@ class BaseUDTFTestsMixin:
func = udtf(TestUDTF, returnType="c1: string, c2: string")
assertDataFrameEqual(func(), [Row(c1="hello", c2="world")])
+ def test_udtf_char_varchar_return_type(self):
+ class TestUDTF:
+ def eval(self):
+ yield ("a",)
+
+ schema = StructType([StructField("c", CharType(3))])
+ with self.assertRaises(PySparkNotImplementedError) as pe:
+ udtf(TestUDTF, returnType=schema)()
+ self.check_error(
+ exception=pe.exception,
+ errorClass="CHAR_VARCHAR_NOT_SUPPORTED_IN_PYTHON",
+ messageParameters={
+ "feature": "Python UDTF return types",
+ "data_type": schema.simpleString(),
+ },
+ )
+
+ def test_udtf_char_varchar_return_type_ddl(self):
+ class TestUDTF:
+ def eval(self):
+ yield ("a",)
+
+ def invoke():
+ return udtf(TestUDTF, returnType="c: char(3)")()
+
+ if is_remote():
+ with self.assertRaises(AnalysisException) as pe:
+ invoke().collect()
+ self.assertEqual(pe.exception.getCondition(),
"CHAR_VARCHAR_NOT_SUPPORTED_IN_PYTHON")
+ else:
+ with self.assertRaises(PySparkNotImplementedError) as pe:
+ invoke()
+ self.check_error(
+ exception=pe.exception,
+ errorClass="CHAR_VARCHAR_NOT_SUPPORTED_IN_PYTHON",
+ messageParameters={
+ "feature": "Python UDTF return types",
+ "data_type": "struct<c:char(3)>",
+ },
+ )
+
+ def test_udtf_char_varchar_analyze_schema(self):
+ class TestUDTF:
+ @staticmethod
+ def analyze(*args):
+ return AnalyzeResult(StructType([StructField("c",
CharType(3))]))
+
+ def eval(self, *args):
+ yield ("a",)
+
+ with self.assertRaises(AnalysisException) as pe:
+ udtf(TestUDTF)().collect()
+ self.assertEqual(pe.exception.getCondition(),
"CHAR_VARCHAR_NOT_SUPPORTED_IN_PYTHON")
+
def test_udtf_yield_single_row_col(self):
class TestUDTF:
def eval(self, a: int):
diff --git a/python/pyspark/sql/types.py b/python/pyspark/sql/types.py
index cabbe4e2b87b..a291ea5481c1 100644
--- a/python/pyspark/sql/types.py
+++ b/python/pyspark/sql/types.py
@@ -3231,6 +3231,19 @@ def _has_type(dt: DataType, dts: Union[type, Tuple[type,
...]]) -> bool:
return False
+def _check_no_char_varchar(dt: DataType, feature: str) -> None:
+ """Reject CHAR/VARCHAR at Python/Arrow host boundaries.
+
+ CHAR/VARCHAR are SQL types. Host APIs must declare STRING instead.
+ Nested struct/array/map and UDT storage types are included.
+ """
+ if _has_type(dt, (CharType, VarcharType)):
+ raise PySparkNotImplementedError(
+ errorClass="CHAR_VARCHAR_NOT_SUPPORTED_IN_PYTHON",
+ messageParameters={"feature": feature, "data_type":
dt.simpleString()},
+ )
+
+
def _first_timestamp_nanos_map_key_type(dt: DataType) -> Optional["DataType"]:
"""Return the key type of the first map (depth-first) whose key carries a
nanosecond timestamp
type, or ``None`` if ``dt`` contains no such map.
diff --git a/python/pyspark/sql/udf.py b/python/pyspark/sql/udf.py
index c2bc85c962fe..e0ff7281ef12 100644
--- a/python/pyspark/sql/udf.py
+++ b/python/pyspark/sql/udf.py
@@ -32,6 +32,7 @@ from pyspark.sql.types import (
DataType,
StringType,
StructType,
+ _check_no_char_varchar,
_parse_datatype_string,
)
from pyspark.sql.utils import get_active_spark_context
@@ -213,12 +214,18 @@ class UserDefinedFunction:
)
self.evalType = evalType
self.deterministic = deterministic
+ if isinstance(returnType, DataType):
+ # Constrained SQL strings are rejected at construction. Eval-type
Arrow/Pandas
+ # conversion stays on ``returnType`` so unrelated types fail only
when consumed.
+ _check_no_char_varchar(returnType, "Python UDF return types")
# Schema of the intermediate aggregation buffer, set only for an
incremental Python
# aggregator (see :class:`pyspark.sql.aggregator.Aggregator`);
``None`` otherwise. It is a
# first-class field so it survives reconstruction paths such as
``_wrapped()``,
# ``asNondeterministic()`` and ``spark.udf.register``, and is threaded
to the JVM in
# ``_create_judf`` so ``PythonAggregate`` can plan the two-stage
aggregation.
self.bufferSchema = bufferSchema
+ if bufferSchema is not None:
+ _check_no_char_varchar(bufferSchema, "Python UDAF buffer schemas")
# Extract Python UDF details if transpilation is enabled.
self.transpiled: list = []
self._transpiled_param_names: list[str] = []
@@ -306,6 +313,10 @@ class UserDefinedFunction:
if not self.transpiled:
detail = f": {errors}" if errors else ""
warnings.warn(f"Unable to transpile UDF {func}{detail}")
+ except PySparkNotImplementedError:
+ # ``self.returnType`` parses DDL and rejects CHAR/VARCHAR. That is
+ # not a transpilation failure; do not swallow it as a warning.
+ raise
except Exception as e:
# An inability to transpile must never break a working UDF -- fall
# back to interpreted Python execution and surface the failure as a
@@ -321,6 +332,7 @@ class UserDefinedFunction:
@staticmethod
def _check_return_type(returnType: DataType, evalType: int) -> None:
+ _check_no_char_varchar(returnType, "Python UDF return types")
if evalType == PythonEvalType.SQL_ARROW_BATCHED_UDF:
try:
to_arrow_type(returnType, timezone="UTC")
diff --git a/python/pyspark/sql/udtf.py b/python/pyspark/sql/udtf.py
index e8dc25dc907e..034668294ed2 100644
--- a/python/pyspark/sql/udtf.py
+++ b/python/pyspark/sql/udtf.py
@@ -32,7 +32,7 @@ from pyspark.errors import (
PySparkTypeError,
)
from pyspark.sql.pandas.utils import require_minimum_pandas_version,
require_minimum_pyarrow_version
-from pyspark.sql.types import DataType, StructType, _parse_datatype_string
+from pyspark.sql.types import DataType, StructType, _check_no_char_varchar,
_parse_datatype_string
from pyspark.sql.udf import _wrap_function
from pyspark.util import PythonEvalType
@@ -369,6 +369,7 @@ class UserDefinedTableFunction:
"return_type": f"{parsed}",
},
)
+ _check_no_char_varchar(parsed, "Python UDTF return types")
self._returnType_placeholder = parsed
return self._returnType_placeholder
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CharVarcharUtils.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CharVarcharUtils.scala
index c9fc3bfd87f1..dc628ac5551d 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CharVarcharUtils.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CharVarcharUtils.scala
@@ -34,6 +34,18 @@ object CharVarcharUtils extends Logging with
SparkCharVarcharUtils {
// visible for testing
private[sql] val CHAR_VARCHAR_TYPE_STRING_METADATA_KEY =
"__CHAR_VARCHAR_TYPE_STRING"
+ /**
+ * Like [[hasCharVarchar]], but also inspects [[UserDefinedType]] storage
types.
+ * Python/Arrow host checks must match PySpark `_has_type`, which unwraps
UDTs.
+ */
+ def hasCharVarcharIncludingUDT(dt: DataType): Boolean = {
+ dt.existsRecursively {
+ case _: CharType | _: VarcharType => true
+ case u: UserDefinedType[_] => hasCharVarcharIncludingUDT(u.sqlType)
+ case _ => false
+ }
+ }
+
/**
* Creates a StringRPad expression with the pad literal inheriting the
collation from the
* str expression's data type. This is necessary because StringRPad may be
created after
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
index ef3d3a9a12bd..c16a7d20dfac 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
@@ -84,6 +84,12 @@ private[sql] object QueryCompilationErrors extends
QueryErrorsBase with Compilat
)
}
+ def charVarcharNotSupportedInPython(feature: String, dataType: String):
Throwable = {
+ new AnalysisException(
+ errorClass = "CHAR_VARCHAR_NOT_SUPPORTED_IN_PYTHON",
+ messageParameters = Map("feature" -> feature, "data_type" -> dataType))
+ }
+
def positionalAndNamedArgumentDoubleReference(
routineName: String, parameterName: String): Throwable = {
val errorClass =
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/python/UserDefinedPythonFunction.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/python/UserDefinedPythonFunction.scala
index 80abe7623e8c..94961872bba9 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/python/UserDefinedPythonFunction.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/python/UserDefinedPythonFunction.scala
@@ -31,6 +31,7 @@ import
org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
import org.apache.spark.sql.catalyst.expressions.{Alias, Ascending,
Descending, Expression, FunctionTableSubqueryArgumentExpression,
NamedArgumentExpression, NullsFirst, NullsLast, PythonAggregate, PythonUDAF,
PythonUDF, PythonUDTF, PythonUDTFAnalyzeResult, PythonUDTFSelectedExpression,
SortOrder, TranspiledPythonUDF, TranspiledUDFParameter,
UnresolvedPolymorphicPythonUDTF, UnresolvedTableArgPlanId}
import org.apache.spark.sql.catalyst.parser.ParserInterface
import org.apache.spark.sql.catalyst.plans.logical.{Generate, LogicalPlan,
NamedParametersSupport, OneRowRelation}
+import org.apache.spark.sql.catalyst.util.CharVarcharUtils
import org.apache.spark.sql.classic.{DataFrame, Dataset, SparkSession}
import org.apache.spark.sql.classic.ClassicConversions._
import org.apache.spark.sql.classic.ColumnConversions
@@ -63,6 +64,14 @@ case class UserDefinedPythonFunction(
bufferType: DataType = null) {
def builder(e: Seq[Expression]): Expression = {
+ if (CharVarcharUtils.hasCharVarcharIncludingUDT(dataType)) {
+ throw QueryCompilationErrors.charVarcharNotSupportedInPython(
+ "Python UDF return types", dataType.catalogString)
+ }
+ if (bufferType != null &&
CharVarcharUtils.hasCharVarcharIncludingUDT(bufferType)) {
+ throw QueryCompilationErrors.charVarcharNotSupportedInPython(
+ "Python UDAF buffer schemas", bufferType.catalogString)
+ }
if (pythonEvalType == PythonEvalType.SQL_BATCHED_UDF
|| pythonEvalType ==PythonEvalType.SQL_ARROW_BATCHED_UDF
|| pythonEvalType == PythonEvalType.SQL_SCALAR_PANDAS_UDF
@@ -236,6 +245,13 @@ case class UserDefinedPythonTableFunction(
*/
NamedParametersSupport.splitAndCheckNamedArguments(exprs, name,
SQLConf.get.resolver)
+ returnType.foreach { rt =>
+ if (CharVarcharUtils.hasCharVarcharIncludingUDT(rt)) {
+ throw QueryCompilationErrors.charVarcharNotSupportedInPython(
+ "Python UDTF return types", rt.catalogString)
+ }
+ }
+
// Check which argument is a table argument here since it will be replaced
with
// `UnresolvedAttribute` to construct lateral join.
val tableArgs = exprs.map {
@@ -375,6 +391,10 @@ class UserDefinedPythonTableFunctionAnalyzeRunner(
val schema = DataType.fromJson(
PythonWorkerUtils.readUTF(length, dataIn)).asInstanceOf[StructType]
+ if (CharVarcharUtils.hasCharVarcharIncludingUDT(schema)) {
+ throw QueryCompilationErrors.charVarcharNotSupportedInPython(
+ "Python UDTF return types", schema.catalogString)
+ }
// Receive the pickled AnalyzeResult buffer, if any.
val pickledAnalyzeResult: Array[Byte] = PythonWorkerUtils.readBytes(dataIn)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/python/PythonCharVarcharRejectSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/python/PythonCharVarcharRejectSuite.scala
new file mode 100644
index 000000000000..8517b0bca028
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/python/PythonCharVarcharRejectSuite.scala
@@ -0,0 +1,109 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.execution.python
+
+import org.apache.spark.api.python.PythonEvalType
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.util.CharVarcharUtils
+import org.apache.spark.sql.test.{ExamplePointUDT, SharedSparkSession}
+import org.apache.spark.sql.types._
+
+class PythonCharVarcharRejectSuite extends SharedSparkSession {
+
+ private class CharStorageUDT extends ExamplePointUDT {
+ override def sqlType: DataType = CharType(3)
+ }
+
+ private def dummyPythonUdf(
+ dataType: DataType,
+ bufferType: DataType = null): UserDefinedPythonFunction = {
+ UserDefinedPythonFunction(
+ name = "dummyUDF",
+ func = new DummyUDF,
+ dataType = dataType,
+ pythonEvalType =
PythonEvalType.SQL_GROUPED_AGG_ARROW_INCREMENTAL_FINAL_UDF,
+ udfDeterministic = true,
+ bufferType = bufferType)
+ }
+
+ test("hasCharVarcharIncludingUDT unwraps UserDefinedType storage") {
+ assert(!CharVarcharUtils.hasCharVarchar(new CharStorageUDT()))
+ assert(CharVarcharUtils.hasCharVarcharIncludingUDT(new CharStorageUDT()))
+ assert(!CharVarcharUtils.hasCharVarcharIncludingUDT(StringType))
+ }
+
+ test("builder rejects CHAR return types separately from STRING buffers") {
+ val udf = dummyPythonUdf(CharType(3), StringType)
+ checkError(
+ exception = intercept[AnalysisException] { udf.builder(Nil) },
+ condition = "CHAR_VARCHAR_NOT_SUPPORTED_IN_PYTHON",
+ parameters = Map(
+ "feature" -> "Python UDF return types",
+ "data_type" -> "char(3)"))
+ }
+
+ test("builder rejects CHAR/VARCHAR buffer schemas with buffer feature name")
{
+ val buffer = StructType(StructField("s", VarcharType(3)) :: Nil)
+ val udf = dummyPythonUdf(StringType, buffer)
+ checkError(
+ exception = intercept[AnalysisException] { udf.builder(Nil) },
+ condition = "CHAR_VARCHAR_NOT_SUPPORTED_IN_PYTHON",
+ parameters = Map(
+ "feature" -> "Python UDAF buffer schemas",
+ "data_type" -> buffer.catalogString))
+ }
+
+ test("builder rejects UDT storage that is CHAR") {
+ val udf = dummyPythonUdf(new CharStorageUDT())
+ checkError(
+ exception = intercept[AnalysisException] { udf.builder(Nil) },
+ condition = "CHAR_VARCHAR_NOT_SUPPORTED_IN_PYTHON",
+ parameters = Map(
+ "feature" -> "Python UDF return types",
+ "data_type" -> "char(3)"))
+ }
+
+ test("builder reports return type before buffer when both are CHAR/VARCHAR")
{
+ val buffer = StructType(StructField("s", VarcharType(3)) :: Nil)
+ val udf = dummyPythonUdf(CharType(3), buffer)
+ checkError(
+ exception = intercept[AnalysisException] { udf.builder(Nil) },
+ condition = "CHAR_VARCHAR_NOT_SUPPORTED_IN_PYTHON",
+ parameters = Map(
+ "feature" -> "Python UDF return types",
+ "data_type" -> "char(3)"))
+ }
+
+ test("UDTF builder rejects CHAR return types") {
+ val schema = StructType(StructField("c", CharType(3)) :: Nil)
+ val udtf = UserDefinedPythonTableFunction(
+ name = "dummyUDTF",
+ func = new DummyUDF,
+ returnType = Some(schema),
+ pythonEvalType = PythonEvalType.SQL_TABLE_UDF,
+ udfDeterministic = true)
+ checkError(
+ exception = intercept[AnalysisException] {
+ udtf.builder(Nil, spark.sessionState.sqlParser)
+ },
+ condition = "CHAR_VARCHAR_NOT_SUPPORTED_IN_PYTHON",
+ parameters = Map(
+ "feature" -> "Python UDTF return types",
+ "data_type" -> schema.catalogString))
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
