srielau commented on code in PR #58549:
URL: https://github.com/apache/spark/pull/58549#discussion_r3969639206
##########
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:
Fixed in af66d51d5fc. Both standard and legacy tests now collect the same
full plan whose Arrow-backed columnar child is asserted, and read the UDF
columns at their original ordinals.
##########
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:
Fixed in af66d51d5fc. Fusion now uses the child expression captured policy
and retains the checked boundary only for constrained checked results. The
existing lazy/view transition regression covers resolution and execution under
opposite settings.
##########
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:
Fixed in af66d51d5fc. The lifted PythonUDF copies applyCharVarcharChecks,
with higher-order transform coverage for CHAR padding and VARCHAR overflow.
##########
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:
Fixed in af66d51d5fc by routing converted entries through
ArrayBasedMapBuilder. The regression verifies the duplicate error and LAST_WIN
behavior after CHAR normalization.
##########
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:
Fixed in af66d51d5fc. Arrow decoding uses a physical STRING schema, while
standard-mode checks project to the declared schema and legacy mode exposes
STRING attributes in both relation branches. The legacy test now asserts the
normalized output schema.
##########
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:
Added in af66d51d5fc. The threshold-0 regression forces the executor/RDD
branch under standard semantics and checks both CHAR padding and VARCHAR
overflow.
##########
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:
Added in af66d51d5fc. Two row UDF expressions capture opposite policies and
execute in one batch; the checked CHAR is padded while the unchecked
over-length VARCHAR is preserved at its own ordinal.
--
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]
