cloud-fan commented on code in PR #58549:
URL: https://github.com/apache/spark/pull/58549#discussion_r3991646192
##########
sql/core/src/main/scala/org/apache/spark/sql/classic/SparkSession.scala:
##########
@@ -887,11 +887,21 @@ class SparkSession private(
private[sql] def applySchemaToPythonRDD(
rdd: RDD[Array[Any]],
schema: StructType): DataFrame = {
+ val applyCharVarcharChecks =
Review Comment:
**Blocking (P1):** `outputSchema` is selected with the boolean captured
above, but this lazy closure calls the overload that reads `SQLConf` again when
an action runs. If the DataFrame is created under standard semantics and
collected after that scope, it can put unchecked/unpadded values into the
retained CHAR/VARCHAR schema; the reverse transition can unexpectedly pad or
reject values for an output already normalized to STRING. Please pass the
captured `applyCharVarcharChecks` into `makeFromJava` here and cover
create-under-one-policy/collect-under-the-other in both directions.
##########
python/pyspark/sql/connect/udtf.py:
##########
@@ -166,10 +171,32 @@ def __init__(
self._name = name or func.__name__
self.evalType = evalType
self.deterministic = deterministic
+ self._validated_return_type_session_ids: Set[str] = set()
+
+ def _check_return_type(self, session: "SparkSession") -> None:
+ if self.returnType is None or self.evalType not in (
Review Comment:
**Blocking (P1):** This early return skips the new recursive return-type
rejection for every regular `SQL_TABLE_UDF`. The shared parity suite calls a
regular UDTF with nested CHAR/VARCHAR inside `assertRaises`, but Connect only
builds a lazy DataFrame here, so that assertion cannot fire later during server
analysis. Could we validate already-materialized `StructType` values for
regular UDTFs before returning, while preserving the no-`ddl_parse`-RPC path
for DDL strings?
##########
sql/core/src/test/scala/org/apache/spark/sql/IntegratedUDFTestUtils.scala:
##########
@@ -443,12 +443,31 @@ object IntegratedUDFTestUtils extends SQLHelper {
children: Seq[Expression],
evalType: Int,
udfDeterministic: Boolean,
- resultId: ExprId)
- extends PythonUDF(name, func, dataType, children, evalType,
udfDeterministic, resultId) {
+ resultId: ExprId,
+ elementwiseNestingDepth: Int,
+ applyCharVarcharChecks: Boolean)
Review Comment:
**Blocking (P1):** `PythonUDFWithoutId` copies the neighboring nesting-depth
and policy fields but not `hasCharVarcharResult`, so `TestTypedScalarPandasUDF`
resets the marker to false. The new true-columnar CHAR/VARCHAR cases then take
the unchecked path instead of the production fallback that applies padding and
overflow checks, either failing or validating the wrong route. Please forward
the wrapped UDF's marker as well.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/ExtractPythonUDFFromLambda.scala:
##########
@@ -441,7 +441,8 @@ object ExtractPythonUDFFromLambda extends Rule[LogicalPlan]
{
// `PythonUDF.liftedElementwiseEvalType`.
PythonUDF.liftedElementwiseEvalType(udf.evalType),
udf.udfDeterministic,
- elementwiseNestingDepth = newDepth)
+ elementwiseNestingDepth = newDepth,
+ applyCharVarcharChecks = udf.applyCharVarcharChecks)
Review Comment:
**Blocking (P1):** The lifted UDF also needs to preserve
`udf.hasCharVarcharResult`. Leaving it at the false default makes an
Arrow-backed `transform` look unconstrained to
`ColumnarArrowEvalPythonEvaluatorFactory`, which can then take
`evalArrowColumnar` and bypass the checked output projection. That leaves
under-length CHAR unpadded and accepts over-length VARCHAR. Please copy the
marker and add an Arrow-backed higher-order-function regression for both
outcomes.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/streaming/TransformWithStateInPySparkStateServer.scala:
##########
@@ -80,6 +82,14 @@ class TransformWithStateInPySparkStateServer(
import PythonResponseWriterUtils._
+ private def validateStateSchema(schema: StructType, schemaKind: String):
Unit = {
+ if (CharVarcharUtils.hasCharVarchar(schema)) {
+ throw QueryCompilationErrors.invalidPythonStateSchema(schema, schemaKind)
+ }
+ }
+
+ validateStateSchema(groupingKeySchema, "grouping key")
Review Comment:
**Blocking (P1):** This constructor check does not run on either production
runner's synchronous initialization path. Both open the state-server socket
first and construct this object inside a listener thread; if this throws, that
thread exits before `run()`/`accept()`, while the Python worker can wait on the
still-advertised but unserved socket and the driver pre-init waits for the
worker response. Please validate the immutable grouping-key schema
synchronously before socket publication/worker startup in both runner paths,
while keeping dynamic value/list/map-state validation in the live request loop.
**Recommended change:** Move or duplicate grouping-key schema validation to
a shared runner-side preflight that runs before initStateServer and worker
startup, and ensure the server constructor cannot be the first place that
rejects the grouping schema.
**Why this works:** Validate the immutable groupingKeySchema on the calling
runner thread, throw the classed unsupported-schema error before allocating or
advertising a socket, and leave request-time validation for dynamically
registered value/list/map state schemas inside the serving loop where errors
are returned over the protocol.
**Scope:**
sql/core/src/main/scala/org/apache/spark/sql/execution/python/streaming,
sql/core/src/test/scala/org/apache/spark/sql/execution/python/streaming
**Compatibility:** Supported TransformWithStateInPySpark initialization and
request-time state schema errors continue to use their existing worker and
socket protocols.
**Risks:** The driver pre-init and executor runner could drift if validation
is duplicated rather than shared. Moving all state-schema validation out of the
server would be incorrect because value/list/map schemas arrive dynamically in
requests.
**Constraints:** Use the existing recursive CharVarcharUtils predicate and
classed invalidPythonStateSchema error. Apply the preflight to both driver
pre-initialization and executor compute paths. Do not open or expose the
state-server socket before the grouping-key preflight succeeds.
**Success:** An unsupported grouping-key schema fails synchronously with
UNSUPPORTED_FEATURE.PYTHON_STATE_CHAR_VARCHAR_SCHEMA. No Python worker waits on
an unserved state-server socket after that failure. Supported grouping schemas
retain current driver and executor state-server lifecycle. Dynamic unsupported
value/list/map state schemas still receive an error response through the live
state-server protocol.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/ColumnarArrowEvalPythonEvaluatorFactory.scala:
##########
@@ -151,15 +173,17 @@ private[python] class
ColumnarArrowEvalPythonEvaluatorFactory(
batch.column(0).isInstanceOf[ArrowColumnVector]
}
- if (inputColumnIndices.isDefined && isArrow) {
+ if (inputColumnIndices.isDefined && isArrow && !hasCharVarcharOutput) {
Review Comment:
**Non-blocking (P2):** `hasCharVarcharResult` remains true even when
`applyCharVarcharChecks` is false and the resolved result type is already
STRING, so this condition unnecessarily removes the full Arrow-columnar path
from legacy UDFs. In the fallback below, the same marker also replaces valid
`inputColumnIndices` with `None`, forcing Path 3's per-row `MutableProjection`,
although Path 2 converges on the same `checkedOutput` projection. Please gate
the full-columnar exclusion on checks actually being active and retain direct
indices for checked outputs.
--
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]
