cloud-fan commented on code in PR #58549:
URL: https://github.com/apache/spark/pull/58549#discussion_r4043362968
##########
python/pyspark/sql/connect/session.py:
##########
@@ -501,6 +504,13 @@ def createDataFrame(
_num_cols = len(schema.fields)
else:
_num_cols = 1
+ if _has_physical_type(schema, (CharType, VarcharType)):
Review Comment:
**Non-blocking (P2):** This guard runs before `_inferSchemaFromList`, so
`schema=None` can still infer a UDT whose storage contains CHAR/VARCHAR after
the check. The widened Arrow mapper then accepts those leaves as UTF8 and sends
an unsupported LocalRelation instead of producing the intended validation
error. Could we apply the same physical-type rejection to the finalized
inferred schema before `LocalDataToArrowConversion`?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/python/ArrowColumnarPythonUDFSuite.scala:
##########
@@ -103,6 +111,181 @@ class ArrowColumnarPythonUDFSuite extends
SharedSparkSession {
}
}
+ test("Arrow-backed source: CHAR/VARCHAR output checks") {
Review Comment:
**Non-blocking (P2):** The Arrow-backed cases here cover top-level results
with one policy at a time, so regressions in nested `physicalOutputSchema`
recursion or `checkedOutput.zip(udfs)` remain invisible. Please add a nested
constrained result and checked/unchecked sibling UDFs in the same Arrow-backed
plan, asserting padding/overflow and the policy at each output ordinal.
##########
python/pyspark/sql/connect/udtf.py:
##########
@@ -166,10 +171,33 @@ def __init__(
self._name = name or func.__name__
self.evalType = evalType
self.deterministic = deterministic
+ self._validated_return_type_session_ids: Set[str] = set()
Review Comment:
**Nit (P3):** This set retains every validated session UUID for the lifetime
of the UDTF object, including sessions that have been closed and can never
reuse the cached fact. Reusing one UDTF across many sessions therefore grows
state monotonically. Could this be a bounded current-session memo (or otherwise
evict expired owners) while retaining same-session reuse?
##########
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):
Review Comment:
**Non-blocking (P2):** The rejection matrix omits several independently
dispatched paths: row-at-a-time UDF UDT storage, `applyInPandasWithState`,
row-mode Connect UDTF DDL, Python DataSource UDT storage, and `toArrow` UDT
storage. Please add focused negative cases that assert the intended classed
validation error, so removing any one physical-type guard cannot leave the
suite green or fall through to an opaque Arrow failure.
##########
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(
+ {
+ "spark.sql.legacy.charVarcharAsString": "false",
+ "spark.sql.preserveCharVarcharTypeInfo": "false",
+ "spark.sql.charVarchar.standardSemantics.enabled": "false",
+ }
+ ):
+ for function in (scalar_char, iterator_char):
+ result = self.spark.range(1).select(function("id").alias("c"))
+ self.assertEqual(result.schema["c"].dataType, StringType())
+ self.assertEqual(result.first().c, "a ")
+ for function in (scalar_varchar, iterator_varchar):
+ with self.assertRaisesRegex(Exception, "EXCEED_LIMIT_LENGTH"):
+ self.spark.range(1).select(function("id")).collect()
+
+ with self.sql_conf({"spark.sql.charVarchar.standardSemantics.enabled":
"true"}):
+ for function in (scalar_char, iterator_char):
+ rows = self.spark.range(2).select(function("id")).collect()
Review Comment:
**Non-blocking (P2):** This standard-semantics case checks values but not
the first-class output type. Returning `StringType` would preserve the current
padding/overflow assertions while violating the feature's schema contract; the
same gap exists in the pandas, higher-order, and Python-RDD creation cases.
Please assert the retained CHAR/VARCHAR schemas, including the constrained
array element for the higher-order result.
##########
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
Review Comment:
**Nit (P3):** This contract says CHAR/VARCHAR are always replaced, but the
UDT branch returns raw `udt.sqlType` without recursively normalizing its
storage. Since UDT-storage assignment support is explicitly out of scope here,
please narrow the Scaladoc to state the raw-storage exception and that callers
must reject unsupported constrained UDT storage before invoking this conversion.
##########
python/pyspark/sql/types.py:
##########
@@ -2877,8 +2877,38 @@ def _has_type(dt: DataType, dts: Union[type, Tuple[type,
...]]) -> bool:
return _has_type(dt.elementType, dts)
elif isinstance(dt, MapType):
return _has_type(dt.keyType, dts) or _has_type(dt.valueType, dts)
+ else:
Review Comment:
**Non-blocking (P2):** Making `_has_type` stop at UDTs also changes existing
callers whose downstream Arrow operation unwraps UDT storage. In particular, a
UDT-backed year-month interval now bypasses the classed preflight and reaches
PyArrow's opaque lookup failure; the pandas legacy-struct preflight has the
same physical/logical mismatch. Could those two callers move to
`_has_physical_type` while logical capability checks keep the new behavior?
--
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]
