srielau commented on code in PR #58549:
URL: https://github.com/apache/spark/pull/58549#discussion_r3960787892
##########
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:
Fixed in 3234e913fb6. Return-type validation now explicitly allows only the
supported scalar eval types and recursively rejects CHAR/VARCHAR for every
other eval type. Added focused stateful and incremental aggregate cases.
##########
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 =>
+ CharVarcharUtils.stringLengthCheck(attr, attr.dataType)
Review Comment:
Fixed in 3234e913fb6. Concrete evaluator factories now declare whether
deserialization already owns CHAR/VARCHAR checks: batched evaluation opts out
of the common projection, while Arrow paths retain it. Per-UDF conversion
preserves exactly one check.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowConverters.scala:
##########
@@ -577,7 +592,7 @@ private[sql] object ArrowConverters extends Logging {
TaskContext.get())
// Project/copy it. Otherwise, the Arrow column vectors will be closed
and released out.
- val proj = UnsafeProjection.create(attrs, attrs)
+ val proj = UnsafeProjection.create(checkedAttrs, attrs)
Review Comment:
Fixed in 3234e913fb6. The Arrow row iterator now has idempotent explicit
cleanup, and local relation materialization closes it in finally after rows are
copied. Added an allocator regression for VARCHAR validation failure.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/ColumnarArrowEvalPythonEvaluatorFactory.scala:
##########
@@ -79,6 +81,22 @@ private[python] class
ColumnarArrowEvalPythonEvaluatorFactory(
sessionUUID: Option[String])
extends PartitionEvaluatorFactory[ColumnarBatch, ColumnarBatch] {
+ private val applyCharVarcharChecks =
+ CharVarcharUtils.shouldApplyWriteSideLengthCheck(SQLConf.get)
+ private val checkedOutput = if (applyCharVarcharChecks) {
+ childOutput ++ output.drop(childOutput.length).map { attr =>
+ CharVarcharUtils.stringLengthCheck(attr, attr.dataType)
+ }
+ } else {
+ output
+ }
+ private val hasCharVarcharOutput =
+ applyCharVarcharChecks &&
+ output.drop(childOutput.length).exists(attr =>
CharVarcharUtils.hasCharVarchar(attr.dataType))
+ private val physicalOutputSchema = CharVarcharUtils
Review Comment:
Fixed in 3234e913fb6. Columnar Arrow physical schema normalization now
recursively unwraps UserDefinedType values to sqlType before replacing
CHAR/VARCHAR. Added a regression with an ExamplePointUDT sibling.
##########
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,
Review Comment:
Added in 3234e913fb6. Owning-suite coverage now includes Pandas SCALAR_ITER
and Arrow SCALAR plus SCALAR_ITER, each checking CHAR padding and VARCHAR
overflow rejection.
##########
python/pyspark/sql/connect/udtf.py:
##########
@@ -167,9 +172,21 @@ def __init__(
self.evalType = evalType
self.deterministic = deterministic
+ def _check_return_type(self) -> None:
+ if self.returnType is None:
Review Comment:
Fixed in 3234e913fb6. Both classic and Connect paths now validate the
effective schema returned by analyze() for Arrow UDTFs, including nested
CHAR/VARCHAR. Added a dynamic analyze() regression.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/ColumnarArrowEvalPythonEvaluatorFactory.scala:
##########
@@ -79,6 +81,22 @@ private[python] class
ColumnarArrowEvalPythonEvaluatorFactory(
sessionUUID: Option[String])
extends PartitionEvaluatorFactory[ColumnarBatch, ColumnarBatch] {
+ private val applyCharVarcharChecks =
+ CharVarcharUtils.shouldApplyWriteSideLengthCheck(SQLConf.get)
Review Comment:
Fixed in 3234e913fb6. The resolved write-side policy is stored on each
PythonUDF/PythonUDTF expression and consumed by physical evaluators instead of
rereading ambient SQLConf. A view regression flips caller settings before
execution.
##########
python/pyspark/sql/connect/udtf.py:
##########
@@ -167,9 +172,21 @@ def __init__(
self.evalType = evalType
self.deterministic = deterministic
+ def _check_return_type(self) -> None:
+ if self.returnType is None:
+ return
+ return_type = (
+ _parse_datatype_string(self.returnType.data_type_string)
Review Comment:
Fixed in 3234e913fb6. Connect skips parsing entirely for non-Arrow UDTFs and
caches successful Arrow validation per owning session. A mocked call-count test
verifies zero regular-UDF RPCs and one repeated Arrow validation RPC.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/EvaluatePython.scala:
##########
@@ -146,10 +147,19 @@ 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 = {
+ val applyCharVarcharChecks =
+ CharVarcharUtils.shouldApplyWriteSideLengthCheck(SQLConf.get)
Review Comment:
Fixed in 3234e913fb6. The query-derived write-check policy is passed
explicitly through the TransformWithState runner, state server, and
deserializer for value, list, and map state. Added a legacy-policy state-server
regression.
--
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]
