srielau commented on code in PR #58549:
URL: https://github.com/apache/spark/pull/58549#discussion_r4000277932
##########
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:
Fixed in 4fba8e30b28. Connect now validates an already-materialized
StructType locally for every UDTF eval type, including regular SQL_TABLE_UDF,
while non-Arrow DDL strings still return before ddl_parse. The shared
regular-UDTF parity case now fails synchronously as intended.
##########
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:
Fixed in 4fba8e30b28. Grouping-key validation is now a shared preflight
invoked synchronously by both executor compute and driver init before socket
allocation or worker startup. The server keeps the same defensive check and
request-time value/list/map validation. Added a driver pre-init regression for
the classed error.
##########
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:
Fixed in c463ebb5c9d and 4fba8e30b28. The fallback retains valid direct
input indices, and the full Arrow-columnar exclusion now requires both
hasCharVarcharResult and applyCharVarcharChecks. Legacy unchecked outputs
therefore retain the optimized path.
##########
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:
Fixed in 4fba8e30b28. PythonUDFWithoutId now accepts, forwards, and copies
hasCharVarcharResult from the wrapped PythonUDF, so the typed pandas test UDF
exercises the checked columnar path.
##########
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:
Fixed in 4fba8e30b28. applySchemaToPythonRDD now passes its captured
applyCharVarcharChecks value into makeFromJava. The creation test now builds
under standard and legacy policies and collects under the opposite policy in
both directions.
##########
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:
Fixed in 4fba8e30b28. The lifted PythonUDF now preserves
hasCharVarcharResult as well as applyCharVarcharChecks. Added an Arrow-backed
transform regression covering CHAR padding and VARCHAR overflow.
--
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]
