cloud-fan commented on code in PR #58549:
URL: https://github.com/apache/spark/pull/58549#discussion_r3960245814
##########
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):** The CHAR/VARCHAR rejection is reachable only from the
eval-type branches below, so public incremental aggregate and
TransformWithState variants currently fall through without either rejection or
assignment checks in their consumers. A VARCHAR(3) result can therefore emit
`abcd`, and a CHAR result remains unpadded. Please make this capability
dispatch exhaustive: only explicitly supported scalar modes should accept
CHAR/VARCHAR, and every other eval type should reject recursively unless its
executor enforces the same semantics. Add focused incremental/stateful cases so
future eval types cannot silently fall through.
##########
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:
**Blocking (P1):** `returnType=None` is the supported dynamic-schema form
for a regular UDTF with `analyze()`, including the Arrow-optimized
SQL_ARROW_TABLE_UDF path. Returning here means a schema containing CHAR/VARCHAR
is never passed to the new rejection and instead reaches execution with a
logical constrained type backed by an Arrow STRING vector. Please validate the
effective schema after analyze in both classic and Connect, or explicitly
disable Arrow optimization for this schema source, and cover nested constrained
types.
**Recommended change:** Apply the same recursive Arrow UDTF capability check
to the schema produced by analyze before constructing the resolved UDTF.
**Why this works:** Place validation at the shared effective-schema boundary
reached by classic and Connect analyze resolution; reject CHAR/VARCHAR before
Arrow planning while leaving non-Arrow analyze behavior unchanged.
**Scope:** Classic and Connect dynamic UDTF resolution plus focused
analyze-based Arrow UDTF tests.
**Compatibility:** Explicit-schema behavior and supported non-Arrow analyze
UDTFs remain unchanged; unsupported Arrow constrained-string schemas fail early
with the intended error.
**Risks:** Validating only one frontend would leave classic and Connect
behavior inconsistent. Validating after physical planning would preserve the
misleading runtime type mismatch.
**Constraints:** Use the effective analyzed schema rather than requiring an
explicit returnType. Preserve analyze support for schemas without CHAR/VARCHAR.
**Success:** Explicit and analyze-derived Arrow UDTF schemas receive
identical recursive CHAR/VARCHAR rejection in classic and Connect.
##########
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:
**Blocking (P1):** When any sibling output contains CHAR/VARCHAR, the
row-queue path converts the entire joined output with this schema. Unlike
`outputTypes` below, this schema leaves UserDefinedType values wrapped, but
RowToColumnConverter only accepts their physical SQL types. Thus a valid Arrow
UDF returning ExamplePointUDT starts failing merely because a sibling UDF
returns CHAR. Please recursively replace each UDT with its `sqlType` before
replacing CHAR/VARCHAR, matching `outputTypes`, and add a mixed CHAR-plus-UDT
columnar-input regression.
##########
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:
**Non-blocking (P2):** This parse happens before checking whether
`_check_arrow_udtf_return_type` applies. In Connect, even an ordinary
SQL_TABLE_UDF with a DDL schema now sends a synchronous `ddl_parse` Analyze
request on every invocation, then discards the parsed type; it also parses
through `SparkSession.active()` rather than the session used for registration.
Please short-circuit non-Arrow eval types before parsing, and make necessary
Arrow validation session-bound and reusable instead of repeating the RPC on
every plan construction.
**Recommended change:** Gate parsing on the two Arrow UDTF eval types, then
bind and cache any required validation against the operation's owning Connect
session.
**Why this works:** Return before DDL parsing for non-Arrow modes; for Arrow
modes, retain the parsed type or validated result on the UDTF instance with the
session context that performed the analysis.
**Scope:** Connect UDTF return-type validation, invocation/registration
session ownership, and RPC-count regression coverage.
**Compatibility:** Ordinary UDTFs recover lazy local plan construction;
Arrow UDTFs retain the intended early recursive rejection without cross-session
parsing.
**Risks:** Caching a parse without session identity could reuse
catalog-dependent resolution across sessions. Gating too broadly could skip
required validation for SQL_ARROW_TABLE_UDF.
**Constraints:** Do not add a server request to non-Arrow UDTF plan
construction. Do not use an unrelated globally active session for registration
validation.
**Success:** Repeated ordinary UDTF calls perform no validation RPC, while
Arrow validation runs through the correct session no more often than necessary.
##########
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:
**Non-blocking (P2):** In the local driver branch, `fromBatchIterator` can
open an allocator and VectorSchemaRoot before this projection is evaluated, and
there is no TaskContext listener. If the new CHAR/VARCHAR projection throws
EXCEED_LIMIT_LENGTH, `toArray` abandons the iterator without exhausting its
close-on-end path, so repeated expected failures retain native Arrow resources.
Please materialize under deterministic cleanup that closes the iterator on
projection or deserialization failure as well as on normal exhaustion.
**Recommended change:** Give the local Arrow iterator an explicit close
boundary and materialize it under try/finally-style resource management.
**Why this works:** Expose or wrap InternalRowIterator cleanup so the driver
branch closes its allocator and vector root after success and after any
projection or deserialization exception.
**Scope:** Arrow batch iterator lifecycle and the local explicit-schema
DataFrame conversion branch, with an exceptional cleanup regression test.
**Compatibility:** Successful local and RDD conversion results are
unchanged; exceptional local conversion releases resources deterministically.
**Risks:** Closing before copied rows are fully materialized would
invalidate retained Arrow-backed values. Duplicated close calls must remain
safe on normal iterator exhaustion.
**Constraints:** Keep the RDD TaskContext cleanup path intact. Copy all
retained rows before closing the Arrow vectors.
**Success:** Both successful exhaustion and any validation/deserialization
failure close all driver-side Arrow resources exactly once.
##########
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:
**Blocking (P1):** This ambient SQLConf read is also used by
`PythonSQLUtils.toJVMRow` on the TransformWithState state-server thread. The
runner captures the query configuration, but the separate server thread never
installs it, so `SQLConf.get` falls back to defaults. A legacy-mode query
updating `VARCHAR(3)` state with `abcd` can therefore raise EXCEED_LIMIT_LENGTH
even though that mode requires plain STRING behavior. Please pass an explicit
derived policy into conversion, or install the runner's captured SQLConf around
the state-server loop, and exercise legacy CHAR/VARCHAR state updates.
**Recommended change:** Bind state conversion to the runner's query
configuration by passing an explicit write-check policy through the
state-server conversion boundary.
**Why this works:** Derive the policy while the query SQLConf is available
and supply it to toJVMRow/makeFromJava, avoiding an ambient ThreadLocal read on
the helper thread.
**Scope:** TransformWithState Python runner/state-server conversion plumbing
and focused legacy-mode state tests.
**Compatibility:** Standard semantics retain padding and overflow checks;
legacy charVarcharAsString queries retain unbounded STRING behavior independent
of thread placement.
**Risks:** Installing a broad SQLConf around a long-lived server loop could
unintentionally affect unrelated configuration reads. A policy passed to only
some state operations could leave value, list, and map state inconsistent.
**Constraints:** Apply one query-bound policy consistently to every state
schema conversion. Do not rely on the helper thread inheriting SQLConf.
**Success:** Value, list, and map state conversions honor the query's
standard or legacy CHAR/VARCHAR semantics on every state-server thread.
##########
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:
**Blocking (P1):** The standard-semantics config is PERSISTED because it
determines how a view body resolves, but this factory recomputes the
enforcement decision from the caller's session at physical execution. A view
created with standard semantics can therefore return over-length VARCHAR or
unpadded CHAR when queried from a legacy-mode session. Please carry the
resolved enforcement decision into the logical/physical Python node instead of
rereading ambient SQLConf, and add a view test that flips the caller
configuration before execution.
**Recommended change:** Carry a plan-bound CHAR/VARCHAR enforcement decision
from resolution into Python physical execution.
**Why this works:** Record the persisted semantic mode, or the derived
write-check boolean, on the resolved Python operator and pass it into evaluator
construction instead of consulting SQLConf.get there.
**Scope:** Python logical/physical operator configuration plumbing, columnar
evaluator construction, and a persisted-view regression test.
**Compatibility:** Direct queries continue to use their own resolved mode,
while views keep the semantics captured when their bodies were resolved as
required by the binding policy.
**Risks:** Binding only the columnar evaluator could leave row-based Python
execution inconsistent. Serializing a session object instead of a small
semantic value would broaden plan state unnecessarily.
**Constraints:** Bind a minimal semantic value rather than ambient session
state. Keep all Python evaluator variants consistent with the resolved plan.
**Success:** Changing the querying session's CHAR/VARCHAR settings cannot
change the results or validation behavior of an already resolved view body.
##########
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:
**Non-blocking (P2):** This common projection also wraps
BatchEvalPythonEvaluatorFactory, whose unpickling path already calls
`EvaluatePython.makeFromJava(resultType)`. The new converter recursively
validates CHAR/VARCHAR in arrays, maps, structs, and UDT storage types, so a
batched UDF returning ARRAY<VARCHAR> now traverses and rebuilds the complete
nested value twice. Please let concrete evaluators declare whether
deserialization already enforces assignment semantics, and apply this
projection only to paths that still need it.
**Recommended change:** Move ownership of the post-result check to
evaluator-specific capability rather than applying it unconditionally in the
common base.
**Why this works:** Have concrete factories opt in or out of checkedOutput
based on whether their deserializer already performs recursive CHAR/VARCHAR
assignment validation.
**Scope:** Python evaluator factory capability plumbing and a nested
batched-UDF regression that confirms one enforcement layer.
**Compatibility:** All evaluator modes retain identical padding and overflow
behavior; only redundant traversal on already-validating batch paths is removed.
**Risks:** An incorrect opt-out could remove the only enforcement layer from
an Arrow path. A coarse factory flag could miss mixed conversion paths.
**Constraints:** Document which concrete conversion boundary owns assignment
enforcement. Keep exactly one recursive check on every supported result path.
**Success:** Nested results are validated once on batch and Arrow paths,
with unchanged CHAR padding and VARCHAR overflow failures.
##########
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:
**Non-blocking (P2):** The tuple now promises CHAR/VARCHAR support for
SCALAR_ITER pandas UDFs and both scalar Arrow UDF modes, but the new
integration cases execute only SQL_ARROW_BATCHED_UDF and default
SQL_SCALAR_PANDAS_UDF. Deleting one of these three entries or breaking its
distinct worker serialization would leave the suite green. Please add
owning-suite cases for `PandasUDFType.SCALAR_ITER`, `ArrowUDFType.SCALAR`, and
`ArrowUDFType.SCALAR_ITER`, asserting padded CHAR output and
EXCEED_LIMIT_LENGTH for an overlong VARCHAR result in each mode.
--
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]
