srielau commented on code in PR #58549:
URL: https://github.com/apache/spark/pull/58549#discussion_r4041491685
##########
python/pyspark/sql/types.py:
##########
@@ -2868,7 +2868,7 @@ def _has_nulltype(dt: DataType) -> bool:
def _has_type(dt: DataType, dts: Union[type, Tuple[type, ...]]) -> bool:
- """Return whether there are specified types"""
+ """Return whether `dt` logically contains any of `dts`. Does not descend
UDTs."""
Review Comment:
`_has_type`'s existing contract intentionally includes UDT storage:
`ArrowTableToRowsConversion.convert` relies on it to turn a year-month interval
inside a UDT into `NOT_IMPLEMENTED` instead of PyArrow `KeyError: 21`, and
pandas conversion uses it for UDT-backed structs. Changing it to logical-only
silently regresses those callers. Please retain `_has_type`'s physical
recursion and introduce `_has_logical_type` for the CHAR/VARCHAR allow-list.
##########
python/pyspark/sql/session.py:
##########
@@ -1611,6 +1617,13 @@ def createDataFrame( # type: ignore[misc]
elif isinstance(schema, (list, tuple)):
# Must re-encode any unicode strings to be consistent with
StructField names
schema = [x.encode("utf-8") if not isinstance(x, str) else x for x
in schema]
+ if isinstance(schema, DataType) and _has_char_varchar_in_udt(schema):
Review Comment:
This guard only sees a user-supplied schema. With `schema=None`,
`_infer_type` returns `obj.__UDT__`; Classic and Connect then continue without
rerunning the UDT check, and Connect's widened mapper admits the storage
CHAR/VARCHAR. Please validate the effective schema after inference in both
clients and add inferred-UDT rejection tests.
##########
python/pyspark/sql/udf.py:
##########
@@ -29,9 +29,16 @@
from pyspark.sql.pandas.types import to_arrow_type
from pyspark.sql.pandas.utils import require_minimum_pandas_version,
require_minimum_pyarrow_version
from pyspark.sql.types import (
+ ArrayType,
Review Comment:
`ArrayType`, `MapType`, and `_has_physical_type` are unused and trigger Ruff
F401. Please remove them.
##########
python/pyspark/sql/tests/test_udf.py:
##########
@@ -55,10 +63,198 @@
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:
+ def test_char_varchar_results(self):
+ schema = StructType(
+ [
+ StructField("c", CharType(4)),
+ StructField("v", VarcharType(3)),
+ StructField("nested", ArrayType(CharType(2))),
+ StructField("m", MapType(CharType(2), VarcharType(3))),
+ ]
+ )
+
+ with self.sql_conf(
+ {
+ "spark.sql.legacy.charVarcharAsString": "false",
+ "spark.sql.preserveCharVarcharTypeInfo": "false",
+ "spark.sql.charVarchar.standardSemantics.enabled": "false",
+ }
+ ):
+ default_result = self.spark.range(1).select(
+ udf(lambda _: "a", CharType(3),
useArrow=False)("id").alias("c")
+ )
+ self.assertEqual(default_result.schema["c"].dataType, StringType())
+ self.assertEqual(default_result.first().c, "a ")
+
+ with self.sql_conf({"spark.sql.charVarchar.standardSemantics.enabled":
"true"}):
+ result = self.spark.range(1).select(
+ udf(
+ lambda _: ("ab", "xyz", ["z"], {"k": "xy"}),
+ schema,
+ useArrow=False,
+ )("id").alias("s")
+ )
+ self.assertEqual(
+ result.first().s,
+ Row(c="ab ", v="xyz", nested=["z "], m={"k ": "xy"}),
+ )
+
+ invalid = self.spark.range(1).select(
+ udf(lambda _: "abcd", VarcharType(3), useArrow=False)("id")
+ )
+ with self.assertRaisesRegex(Exception, "EXCEED_LIMIT_LENGTH"):
+ invalid.collect()
+
+ def test_char_varchar_legacy_as_string(self):
+ with self.sql_conf(
+ {
+ "spark.sql.legacy.charVarcharAsString": "true",
+ "spark.sql.preserveCharVarcharTypeInfo": "false",
+ "spark.sql.charVarchar.standardSemantics.enabled": "false",
+ }
+ ):
+ result = self.spark.range(1).select(
+ udf(lambda _: "a", CharType(3),
useArrow=False)("id").alias("c"),
+ udf(lambda _: "abcd", VarcharType(3),
useArrow=False)("id").alias("v"),
+ )
+ self.assertEqual(result.first(), Row(c="a", v="abcd"))
+
+ def test_char_varchar_intermediate_udf_results(self):
+ inner_char = udf(lambda _: "a", CharType(3), useArrow=False)
+ inner_varchar = udf(lambda _: "abcd", VarcharType(3), useArrow=False)
+ outer = udf(lambda value: value, StringType(), useArrow=False)
+
+ with self.sql_conf({"spark.sql.charVarchar.standardSemantics.enabled":
"true"}):
+ padded =
self.spark.range(1).select(outer(inner_char("id")).alias("result"))
+ self.assertEqual(padded.first().result, "a ")
+
+ invalid = self.spark.range(1).select(outer(inner_varchar("id")))
+ with self.assertRaisesRegex(Exception, "EXCEED_LIMIT_LENGTH"):
+ invalid.collect()
+
+ with self.sql_conf(
+ {
+ "spark.sql.legacy.charVarcharAsString": "true",
+ "spark.sql.preserveCharVarcharTypeInfo": "false",
+ "spark.sql.charVarchar.standardSemantics.enabled": "false",
+ }
+ ):
+ result = self.spark.range(1).select(
+ outer(inner_char("id")).alias("c"),
+ outer(inner_varchar("id")).alias("v"),
+ )
+ self.assertEqual(result.first(), Row(c="a", v="abcd"))
+
+ def test_char_varchar_view_keeps_resolved_semantics(self):
+ with self.temp_view("char_varchar_udf_view"):
+ with
self.sql_conf({"spark.sql.charVarchar.standardSemantics.enabled": "true"}):
+ self.spark.range(1).select(
+ udf(lambda _: "a", CharType(3),
useArrow=False)("id").alias("c"),
+ udf(lambda _: "abcd", VarcharType(3),
useArrow=False)("id").alias("v"),
+ ).createOrReplaceTempView("char_varchar_udf_view")
+
+ with self.sql_conf(
+ {
+ "spark.sql.legacy.charVarcharAsString": "true",
+ "spark.sql.preserveCharVarcharTypeInfo": "false",
+ "spark.sql.charVarchar.standardSemantics.enabled": "false",
+ }
+ ):
+ self.assertEqual(
+ self.spark.sql("SELECT c FROM
char_varchar_udf_view").collect()[0].c,
+ "a ",
+ )
+ with self.assertRaisesRegex(Exception, "EXCEED_LIMIT_LENGTH"):
+ self.spark.sql("SELECT v FROM
char_varchar_udf_view").collect()
+
+ def test_char_varchar_mixed_captured_policies_in_one_batch(self):
+ char_udf = udf(lambda _: "a", CharType(3), useArrow=False)
+ varchar_udf = udf(lambda _: "abcd", VarcharType(3), useArrow=False)
+ with self.sql_conf({"spark.sql.charVarchar.standardSemantics.enabled":
"true"}):
+ checked_char = char_udf("id").alias("c")
+ with self.sql_conf(
+ {
+ "spark.sql.legacy.charVarcharAsString": "true",
+ "spark.sql.preserveCharVarcharTypeInfo": "false",
+ "spark.sql.charVarchar.standardSemantics.enabled": "false",
+ }
+ ):
+ unchecked_varchar = varchar_udf("id").alias("v")
+
+ self.assertEqual(
+ self.spark.range(1).select(checked_char,
unchecked_varchar).collect()[0],
+ Row(c="a ", v="abcd"),
+ )
+
+ def test_char_varchar_non_scalar_return_types_unsupported(self):
+ nested_return_type = StructType([StructField("nested",
ArrayType(CharType(3)))])
+ struct_eval_types = [
Review Comment:
The rejection matrix omits `SQL_GROUPED_MAP_PANDAS_ITER_UDF`,
`SQL_GROUPED_MAP_ARROW_ITER_UDF`, and `SQL_GROUPED_MAP_PANDAS_UDF_WITH_STATE`,
although production handles all three. Add them so every unsupported public
mapper consumer has a regression.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CharVarcharUtils.scala:
##########
@@ -33,6 +33,43 @@ object CharVarcharUtils extends Logging with
SparkCharVarcharUtils {
// visible for testing
private[sql] val CHAR_VARCHAR_TYPE_STRING_METADATA_KEY =
"__CHAR_VARCHAR_TYPE_STRING"
+ /**
+ * Returns whether a type contains CHAR/VARCHAR, including inside UDT
storage. This is intended
+ * for validation at boundaries that do not support CHAR/VARCHAR in UDT
storage.
+ */
+ private[sql] def physicalTypeHasCharVarchar(dt: DataType): Boolean = dt
match {
+ case ArrayType(elementType, _) => physicalTypeHasCharVarchar(elementType)
+ case MapType(keyType, valueType, _) =>
+ physicalTypeHasCharVarchar(keyType) ||
physicalTypeHasCharVarchar(valueType)
+ case StructType(fields) => fields.exists(f =>
physicalTypeHasCharVarchar(f.dataType))
+ case udt: UserDefinedType[_] => physicalTypeHasCharVarchar(udt.sqlType)
+ case _: CharType | _: VarcharType => true
+ case _ => false
+ }
+
+ /**
+ * Replaces CHAR/VARCHAR with their unconstrained string representation
regardless of session
+ * configuration. Use this only at physical boundaries, such as Arrow, that
encode all character
+ * string types as UTF8.
+ */
Review Comment:
This currently implies that UDT storage is recursively normalized, but line
67 unwraps it without visiting the returned storage type. It also uses `UTF8`
where prose should use `UTF-8`.
```suggestion
/**
* Replaces logical CHAR/VARCHAR with their unconstrained string
representation regardless of
* session configuration. Use this only at physical boundaries, such as
Arrow, that encode all
* character string types as UTF-8. UDTs are unwrapped without normalizing
their storage types,
* so callers must reject CHAR/VARCHAR inside UDT storage first.
*/
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/PythonUDF.scala:
##########
@@ -337,9 +337,14 @@ case class PythonUDF(
// single lambda, and one more for each enclosing lambda when the UDF is
lifted out of a nested
// lambda (e.g. `transform(arr, i -> transform(i, x -> f(x)))` lifts `f`
to depth 2). Ignored
// for every non-element-wise eval type, where it stays at its default of
1.
- elementwiseNestingDepth: Int = 1)
+ elementwiseNestingDepth: Int = 1,
+ // Original CHAR/VARCHAR result type when write-side checks apply. Absent
for unconstrained
+ // results so CHAR policy is not part of PythonUDF equality.
Review Comment:
Clarify that the policy covers both constrained types:
```suggestion
// results so CHAR/VARCHAR policy is not part of PythonUDF equality.
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/EvaluatePython.scala:
##########
@@ -146,10 +147,47 @@ object EvaluatePython {
* Make a converter that converts `obj` to the type specified by the data
type, or returns
* null if the type of obj is unexpected. Because Python doesn't enforce the
type.
*/
- def makeFromJava(dataType: DataType): Any => Any =
-
TypeApiOps(dataType).flatMap(_.makeFromJava).getOrElse(makeFromJavaDefault(dataType))
+ def makeFromJava(dataType: DataType): Any => Any = {
Review Comment:
The earlier thread concerned TransformWithState policy capture; this is a
different unchanged consumer. The generic overload now changes
`ApplyInPandasWithStatePythonRunner` state conversion through
`PythonSQLUtils.toJVMRow`, although that state API is outside this PR's support
fence. Keep this overload behavior-neutral (`false`) and let supported paths
pass captured policy explicitly.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/EvaluatePython.scala:
##########
@@ -227,19 +277,20 @@ object EvaluatePython {
}
case MapType(keyType, valueType, _) =>
- val keyFromJava = makeFromJava(keyType)
- val valueFromJava = makeFromJava(valueType)
+ val keyFromJava = makeFromJava(keyType, applyCharVarcharChecks)
+ val valueFromJava = makeFromJava(valueType, applyCharVarcharChecks)
(obj: Any) => nullSafeConvert(obj) {
case javaMap: java.util.Map[_, _] =>
- ArrayBasedMapData(
- javaMap,
- (key: Any) => keyFromJava(key),
- (value: Any) => valueFromJava(value))
+ val builder = new ArrayBasedMapBuilder(keyType, valueType)
Review Comment:
`ArrayBasedMapBuilder` is needed when CHAR padding coalesces map keys, but
this now affects every `MapType`, even when checks are disabled or the key
contains no CHAR/VARCHAR. That changes duplicate/null behavior and adds hashing
or ordering overhead to all Python map results. Preserve `ArrayBasedMapData`
unless captured checks are active and the key type contains CHAR/VARCHAR.
##########
python/pyspark/sql/tests/arrow/test_arrow_udf_scalar.py:
##########
@@ -58,6 +60,50 @@
@unittest.skipIf(not have_pyarrow, pyarrow_requirement_message)
class ScalarArrowUDFTestsMixin:
+ def test_char_varchar_scalar_results(self):
+ import pyarrow as pa
+
+ @arrow_udf(CharType(3), ArrowUDFType.SCALAR)
+ def scalar_char(values):
+ return pa.array(["a"] * len(values))
+
+ @arrow_udf(CharType(3), ArrowUDFType.SCALAR_ITER)
+ def iterator_char(batches):
+ for values in batches:
+ yield pa.array(["a"] * len(values))
+
+ @arrow_udf(VarcharType(3), ArrowUDFType.SCALAR)
+ def scalar_varchar(values):
+ return pa.array(["abcd"] * len(values))
+
+ @arrow_udf(VarcharType(3), ArrowUDFType.SCALAR_ITER)
+ def iterator_varchar(batches):
+ for values in batches:
+ yield pa.array(["abcd"] * len(values))
+
+ with self.sql_conf(
Review Comment:
The frozen scope requires each supported surface under
standard/default/legacy policy. Missing cases include legacy native Arrow and
pandas scalar/iterator, default/legacy HOF, legacy `toArrow`, and parts of the
pandas/PyArrow creation cross-product. Please complete that matrix, including
non-iterator pandas scalar.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
