cloud-fan commented on code in PR #58549:
URL: https://github.com/apache/spark/pull/58549#discussion_r3962447628
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/ExtractPythonUDFs.scala:
##########
@@ -200,6 +201,8 @@ object ExtractPythonUDFs extends Rule[LogicalPlan] with
Logging {
case Seq(child: PythonUDF) =>
correctEvalType(e, pythonUDFArrowFallbackOnUDT) ==
correctEvalType(child, pythonUDFArrowFallbackOnUDT) &&
+ !(CharVarcharUtils.shouldApplyWriteSideLengthCheck(conf) &&
Review Comment:
**Blocking (P1):** This decision re-reads the querying session's SQLConf
even though the child UDF now carries its resolved policy. If a view is
resolved under standard semantics and queried from a legacy session, this can
fuse away the child's JVM boundary, so the outer UDF sees an unpadded CHAR or
an over-length VARCHAR avoids its error. Please gate on
`child.applyCharVarcharChecks && hasCharVarchar(child.dataType)` and add a
construction-versus-execution config-transition regression.
**Recommended change:** Make nested-UDF fusion depend on the resolved child
expression's captured policy rather than optimizer-time SQLConf.
**Why this works:** Replace the ambient write-side policy read with
child.applyCharVarcharChecks while retaining the constrained-type predicate, so
checked intermediates keep a JVM conversion boundary.
**Scope:** ExtractPythonUDFs nested-chain extraction and a persisted or
lazy-plan configuration-transition test.
**Compatibility:** Preserve current fusion for unchecked children and
unconstrained result types; only prevent fusion where the resolved child
already requires assignment checks.
**Risks:** Blocking fusion adds a Python evaluation boundary for constrained
checked intermediates. A regression test that does not separate resolution from
execution would miss the lifecycle defect.
**Constraints:** Do not re-resolve policy from the current SQLConf. Keep the
change local to constrained child results whose captured flag is true.
**Success:** The same resolved nested-UDF plan returns identical checked
results when later queried under an opposite CHAR/VARCHAR session policy.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/PythonUDF.scala:
##########
@@ -337,7 +337,8 @@ case class PythonUDF(
// single lambda, and one more for each enclosing lambda when the UDF is
lifted out of a nested
// lambda (e.g. `transform(arr, i -> transform(i, x -> f(x)))` lifts `f`
to depth 2). Ignored
// for every non-element-wise eval type, where it stays at its default of
1.
- elementwiseNestingDepth: Int = 1)
+ elementwiseNestingDepth: Int = 1,
+ applyCharVarcharChecks: Boolean = false)
Review Comment:
**Blocking (P1):** The lambda rewrite builds a replacement `PythonUDF` but
does not copy this newly captured flag, so its default becomes false. A
CHAR/VARCHAR UDF inside `transform` can then return an unpadded CHAR or accept
an over-length VARCHAR. Please pass `applyCharVarcharChecks =
udf.applyCharVarcharChecks` when constructing the lifted UDF and add
higher-order-function coverage for both outcomes.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/python/ArrowColumnarPythonUDFSuite.scala:
##########
@@ -103,6 +116,71 @@ 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") {
+ val charUDF = TestTypedScalarPandasUDF(
+ name = "arrow_char_udf", returnType = CharType(4))
+ val varcharUDF = TestTypedScalarPandasUDF(
+ name = "arrow_varchar_udf", returnType = VarcharType(3))
+ registerTestUDF(charUDF, spark)
+ registerTestUDF(varcharUDF, spark)
+
+ val df = readArrowSource(numRows = 10)
+ val padded = df.selectExpr(
+ "id", "name", "value", "data",
+ "arrow_char_udf(id) as udf_id")
+ val arrowExec = collectNodes[ArrowEvalPythonExec](
+ padded.queryExecution.executedPlan).head
+ assert(arrowExec.child.supportsColumnar,
+ "ArrowEvalPythonExec should retain its Arrow-backed columnar child")
+ assert(padded.select("udf_id").collect().map(_.getString(0)).toSeq ===
Review Comment:
**Non-blocking (P2):** This inspects `padded` but collects
`padded.select("udf_id")`, which is a separately optimized query. The added
projection can prune the Arrow source and make the collected query fall back to
row input, so the test can pass without exercising the asserted columnar path.
Please collect `padded` itself, read the UDF field at its original ordinal, and
make the same change in the legacy case.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/EvaluatePython.scala:
##########
@@ -227,8 +277,8 @@ object EvaluatePython {
}
case MapType(keyType, valueType, _) =>
- val keyFromJava = makeFromJava(keyType)
- val valueFromJava = makeFromJava(valueType)
+ val keyFromJava = makeFromJava(keyType, applyCharVarcharChecks)
Review Comment:
**Blocking (P1):** Key conversion can now coalesce distinct Python keys: for
`MapType(CharType(2), ...)`, both `"a"` and `"a "` become the same Catalyst
key. Passing those entries directly to `ArrayBasedMapData` violates its
documented no-duplicates precondition. Please build through the duplicate-aware
map builder, or otherwise detect post-normalization collisions and honor
`spark.sql.mapKeyDedupPolicy`.
**Recommended change:** Route converted map entries through Spark's
duplicate-aware map construction policy after key normalization.
**Why this works:** Normalize each key first, then use ArrayBasedMapBuilder
or equivalent collision detection so equal normalized keys follow
spark.sql.mapKeyDedupPolicy.
**Scope:** Python-to-Catalyst map conversion for constrained key types and
collision regression tests.
**Compatibility:** Preserve behavior for maps whose normalized keys remain
unique; make newly created collisions follow the same policy as other Catalyst
map constructors.
**Risks:** Applying deduplication before CHAR normalization would miss the
collision. Ignoring the configured LAST_WIN policy or exception behavior would
diverge from Spark SQL map semantics.
**Constraints:** Do not admit duplicate keys into ArrayBasedMapData. Apply
the configured duplicate-key policy to the normalized Catalyst keys.
**Success:** Keys such as `a` and `a ` either raise the configured
duplicate-key error or deterministically deduplicate under LAST_WIN, and all
unique maps remain unchanged.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/BatchEvalPythonExec.scala:
##########
@@ -106,7 +106,13 @@ class BatchEvalPythonEvaluatorFactory(
StructType(udfs.map(u => StructField("", u.dataType, u.nullable)))
}
- val fromJava = EvaluatePython.makeFromJava(resultType)
+ val fromJava = if (udfs.length == 1) {
+ EvaluatePython.makeFromJava(resultType, udfs.head.applyCharVarcharChecks)
+ } else {
+ EvaluatePython.makeFromJava(
Review Comment:
**Non-blocking (P2):** This sequence is consumed positionally, but current
multi-UDF coverage makes every flag false. Please construct two independent
expressions under opposite policies, evaluate them together in one
`BatchEvalPythonExec`, and assert that the checked CHAR field is padded while
the unchecked over-length VARCHAR field is preserved at the matching ordinal.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowConverters.scala:
##########
@@ -557,13 +574,19 @@ private[sql] object ArrowConverters extends Logging {
val rdd = session.sparkContext
.parallelize(batchesInDriver.toImmutableArraySeq,
batchesInDriver.length)
.mapPartitions { batchesInExecutors =>
- ArrowConverters.fromBatchIterator(
+ val rows = ArrowConverters.fromBatchIterator(
batchesInExecutors,
schema,
timeZoneId,
errorOnDuplicatedFieldNames,
largeVarTypes,
TaskContext.get())
+ if (applyCharVarcharChecks) {
Review Comment:
**Non-blocking (P2):** This is a separately implemented executor-side
projection, but the standard-semantics tests all stay on the local branch; the
only test forcing threshold 0 enables legacy mode, where this block is skipped.
Please force the RDD branch under standard semantics and assert both CHAR
padding and an `EXCEED_LIMIT_LENGTH` failure during materialization.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowConverters.scala:
##########
@@ -548,6 +557,14 @@ private[sql] object ArrowConverters extends Logging {
errorOnDuplicatedFieldNames: Boolean,
largeVarTypes: Boolean): DataFrame = {
val attrs = toAttributes(schema)
+ val applyCharVarcharChecks =
+ CharVarcharUtils.hasCharVarchar(schema) &&
+
CharVarcharUtils.shouldApplyWriteSideLengthCheck(session.sessionState.conf)
+ val checkedAttrs = if (applyCharVarcharChecks) {
Review Comment:
**Blocking (P1):** In legacy-as-string mode this flag is false, but `attrs`
and the schema passed to both relation branches still contain CHAR/VARCHAR
while first-class types are disabled. `CheckAnalysis` rejects those leaf
outputs before the intended unchecked values can be read. Please normalize the
relation schema and attributes to STRING under the existing legacy policy,
retaining the declared logical schema only where standard-mode projection needs
it.
**Recommended change:** Separate the declared schema used for standard
assignment checks from the policy-normalized schema exposed by Arrow relations.
**Why this works:** Under legacy-as-string mode, derive relation attributes
and the LogicalRDD or LocalRelation schema from the existing
CHAR/VARCHAR-to-STRING normalization while keeping the original schema only for
Arrow decoding or standard checks where required.
**Scope:** ArrowConverters.toDataFrame relation construction and any
Python-side schema caching that exposes the unnormalized logical schema.
**Compatibility:** Restore the established legacy contract that constrained
strings appear as STRING and remain unchecked; preserve first-class logical
types under standard semantics.
**Risks:** Normalizing the decoding schema too early could lose information
needed for standard-mode checks. Using different attribute sequences for
projection and relation construction can cause expression-ID or type mismatches.
**Constraints:** Both the RDD and local relation branches must expose the
same policy-normalized schema. Do not normalize away CHAR/VARCHAR in standard
semantics.
**Success:** Both Arrow relation branches analyze and return unchecked
STRING values in legacy mode, while standard mode still pads CHAR and rejects
over-length VARCHAR.
--
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]
