cloud-fan commented on code in PR #58549:
URL: https://github.com/apache/spark/pull/58549#discussion_r3945912270
##########
python/pyspark/sql/udf.py:
##########
@@ -314,9 +317,24 @@ def _conf_is_true(key: str, default: Optional[str] = None)
-> bool:
@staticmethod
def _check_return_type(returnType: DataType, evalType: int) -> None:
+ char_varchar_supported_eval_types = (
+ PythonEvalType.SQL_ARROW_BATCHED_UDF,
+ PythonEvalType.SQL_SCALAR_PANDAS_UDF,
+ PythonEvalType.SQL_SCALAR_PANDAS_ITER_UDF,
+ PythonEvalType.SQL_SCALAR_ARROW_UDF,
+ PythonEvalType.SQL_SCALAR_ARROW_ITER_UDF,
+ )
+
+ def check_arrow_type() -> None:
+ if evalType not in char_varchar_supported_eval_types and _has_type(
Review Comment:
**Blocking (P1):** This guard is only reached from explicit eval-type
branches, and `_check_return_type` still has no branch for
`SQL_GROUPED_AGG_PANDAS_ITER_UDF` or `SQL_GROUPED_AGG_ARROW_ITER_UDF`. Both
variants are public and are planned with `ArrowAggregatePythonExec`, whose
output path never applies `stringLengthCheck`, so an under-length CHAR can
remain unpadded and an over-length VARCHAR can escape validation. Please reject
both iterator aggregate variants here and add them to the focused negative
eval-type matrix until their executor enforces recursive assignment semantics.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/EvalPythonEvaluatorFactory.scala:
##########
@@ -36,6 +38,16 @@ abstract class EvalPythonEvaluatorFactory(
output: Seq[Attribute])
extends PartitionEvaluatorFactory[InternalRow, InternalRow] {
+ private val applyCharVarcharChecks =
+ CharVarcharUtils.shouldApplyWriteSideLengthCheck(SQLConf.get)
+ private val checkedOutput = if (applyCharVarcharChecks) {
+ childOutput ++ output.drop(childOutput.length).map { attr =>
Review Comment:
**Blocking (P1):** This projection validates only the final PythonExec
output, but `collectFunctions` fuses a same-eval-type child UDF into the outer
function and the worker passes the inner raw value directly onward. For
`outer(inner(id))`, an inner `CHAR(3)` returning `"a"` is therefore observed as
length 1 instead of 3, and an over-length intermediate VARCHAR can be consumed
without `EXCEED_LIMIT_LENGTH`. Please preserve the declared intermediate
boundary and add padding/overflow regressions for both Arrow and pickled UDF
chains.
**Recommended change:** Stop fusing across an intermediate UDF whose
declared result recursively contains CHAR/VARCHAR while write-side checks are
active.
**Why this works:** Make Python UDF extraction materialize the constrained
inner UDF in its own PythonExec node, allowing the existing checked-output
projection to enforce its declared type before the outer UDF consumes it.
**Scope:** Python UDF expression extraction/fusion plus focused chained-UDF
tests for row and Arrow execution.
**Compatibility:** Keep existing fusion for all unconstrained result types
and for legacy-as-string mode, where write-side checks are intentionally
disabled.
**Risks:** Affected nested constrained-string expressions incur an
additional Python execution boundary. The extraction condition must behave
consistently for pickled, Arrow, pandas, and columnar-capable scalar evaluators.
**Constraints:** Gate only on the established write-side-check policy and
recursive CHAR/VARCHAR presence. Do not disable fusion for unrelated UDF chains
or change legacy-as-string results.
**Success:** The outer UDF observes padded CHAR values, over-length
intermediate CHAR/VARCHAR results fail before the outer UDF runs, and
unconstrained and legacy chains retain their current fusion behavior.
##########
python/pyspark/sql/pandas/types.py:
##########
@@ -133,7 +135,7 @@ def to_arrow_type(
arrow_type = pa.float64()
elif isinstance(dt, DecimalType):
arrow_type = pa.decimal128(dt.precision, dt.scale)
- elif isinstance(dt, StringType):
+ elif isinstance(dt, (StringType, CharType, VarcharType)):
Review Comment:
**Blocking (P1):** This shared mapping now also accepts CHAR/VARCHAR in
Arrow UDTF return schemas. The worker emits physical Arrow STRING fields, but
`ArrowEvalPythonUDTFExec` compares them with logical `CharType`/`VarcharType`
attributes and raises `arrowDataTypeMismatchError`; it also has no
assignment-checking projection. Please keep recursively constrained schemas
unsupported at the Arrow UDTF boundary until that executor implements physical
normalization and write-side checks.
**Recommended change:** Reject Arrow UDTF return schemas that recursively
contain CHAR/VARCHAR until end-to-end executor support exists.
**Why this works:** Add a recursive constrained-string capability check to
Arrow UDTF return-type validation before the shared Arrow mapper is used,
returning the established unsupported-return-type error.
**Scope:** PySpark Arrow UDTF validation and focused negative tests for
direct and nested CHAR/VARCHAR return fields.
**Compatibility:** Restore the pre-change unsupported boundary only for
Arrow UDTFs; retain the shared mapping for scalar UDF, DataFrame creation, and
toArrow paths that have matching consumers.
**Risks:** The validation must produce a stable PySpark error for both
parsed DDL strings and DataType objects.
**Constraints:** Detect CHAR/VARCHAR recursively through structs, arrays,
and maps. Do not remove the shared physical STRING mapping or disable supported
scalar and DataFrame paths.
**Success:** Arrow UDTFs with direct or nested CHAR/VARCHAR return fields
fail during return-type validation instead of reaching a
logical-versus-physical runtime mismatch.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/python/ArrowColumnarPythonUDFSuite.scala:
##########
@@ -103,6 +104,39 @@ class ArrowColumnarPythonUDFSuite extends
SharedSparkSession {
}
}
+ test("Arrow-backed source: CHAR/VARCHAR output checks") {
+ assume(shouldTestPandasUDFs)
+ withSQLConf(
+ SQLConf.ARROW_PYSPARK_EXECUTION_ENABLED.key -> "true",
+ SQLConf.ARROW_PYSPARK_UDF_COLUMNAR_INPUT_ENABLED.key -> "true",
+ SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
Review Comment:
**Non-blocking (P2):** This owning Arrow-backed-input suite covers only
standard semantics. In legacy-as-string mode, the same policy also controls
whether output skips `checkedOutput` and retains the full Arrow-columnar path.
Please add a companion case using `readArrowSource` that asserts an
under-length CHAR remains unpadded, an over-length VARCHAR is accepted, and the
Arrow-backed child stays columnar.
##########
python/pyspark/sql/tests/arrow/test_arrow_python_udf.py:
##########
@@ -270,18 +271,45 @@ def f(v: float):
rounded = df.select(f("v").alias("d")).first().d
self.assertEqual(rounded, Decimal("1.233999999999999986"))
- def test_err_return_type(self):
- with self.assertRaises(PySparkNotImplementedError) as pe:
- udf(lambda x: x, VarcharType(10), useArrow=True)
-
- self.check_error(
- exception=pe.exception,
- errorClass="NOT_IMPLEMENTED",
- messageParameters={
- "feature": "Invalid return type with Arrow-optimized Python
UDF: VarcharType(10)"
- },
+ 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.charVarchar.standardSemantics.enabled": "true",
Review Comment:
**Non-blocking (P2):** This Arrow-specific test covers only the
standard-semantics arm, while the added legacy UDF cases explicitly use
`useArrow=False`. A regression that always applies `checkedOutput` to Arrow
UDFs would pad `CHAR(3)` and reject an over-length `VARCHAR(3)` in
legacy-as-string mode without failing this suite. Please add a `useArrow=True`
legacy case asserting that `"a"` stays unpadded and `"abcd"` remains accepted.
--
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]
