cloud-fan commented on code in PR #58549:
URL: https://github.com/apache/spark/pull/58549#discussion_r3944526360
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/EvaluatePython.scala:
##########
@@ -207,6 +207,18 @@ object EvaluatePython {
case c: Int => c.toLong
}
+ case c: CharType => (obj: Any) => nullSafeConvert(obj) {
+ case _ =>
+ CharVarcharCodegenUtils.charTypeWriteSideCheck(
Review Comment:
**Blocking (P1):** These new CHAR/VARCHAR branches bypass
`CharVarcharUtils.shouldApplyWriteSideLengthCheck`. With
`spark.sql.legacy.charVarcharAsString=true` and both first-class modes off, a
`CharType(3)` UDF returning `"a"` is now padded and an over-length
`VarcharType` result now fails, although that mode explicitly promises no
padding or length check. Please gate this conversion and the new scalar output
projection with the existing helper so the legacy path remains unconstrained
STRING while first-class modes retain the checks.
**Recommended change:** Use the existing write-side-check policy for every
new Python scalar conversion and output projection.
**Why this works:** Evaluate shouldApplyWriteSideLengthCheck once for the
active SQLConf and select the unconstrained STRING conversion/projection only
when legacy-as-string is active without either first-class mode.
**Scope:** EvaluatePython.makeFromJava, EvalPythonEvaluatorFactory, and
focused legacy-mode regression tests for explicit-schema creation and scalar
UDF output.
**Compatibility:** Restores the documented legacy behavior without changing
standard-semantics or preserve-type-info behavior.
**Risks:** The bypass must remain limited to the exact policy helper result
so neither first-class mode loses assignment checks.
**Constraints:** Reuse CharVarcharUtils.shouldApplyWriteSideLengthCheck
rather than duplicating configuration precedence.
**Success:** Legacy mode preserves unpadded and over-length STRING values,
while both first-class modes still pad CHAR and reject over-length CHAR/VARCHAR
values.
##########
python/pyspark/sql/tests/arrow/test_arrow_python_udf.py:
##########
@@ -270,18 +272,53 @@ def f(v: float):
rounded = df.select(f("v").alias("d")).first().d
self.assertEqual(rounded, Decimal("1.233999999999999986"))
- def test_err_return_type(self):
- with self.assertRaises(PySparkNotImplementedError) as pe:
- udf(lambda x: x, VarcharType(10), useArrow=True)
-
- self.check_error(
- exception=pe.exception,
- errorClass="NOT_IMPLEMENTED",
- messageParameters={
- "feature": "Invalid return type with Arrow-optimized Python
UDF: VarcharType(10)"
- },
+ def test_char_varchar_results(self):
+ schema = StructType(
+ [
+ StructField("c", CharType(4)),
+ StructField("v", VarcharType(3)),
+ StructField("nested", ArrayType(CharType(2))),
+ StructField("m", MapType(CharType(2), VarcharType(3))),
+ ]
)
+ with self.sql_conf(
+ {
+ "spark.sql.charVarchar.standardSemantics.enabled": "true",
+ "spark.sql.execution.arrow.pythonUDF.columnarInput.enabled":
"true",
+ }
+ ):
+ result = self.spark.range(1).select(
+ udf(
+ lambda _: ("ab", "xyz", ["z"], {"k": "xy"}),
+ schema,
+ useArrow=True,
+ )("id").alias("s")
+ )
+ self.assertEqual(
+ result.first().s,
+ Row(c="ab ", v="xyz", nested=["z "], m={"k ": "xy"}),
+ )
+
+ pandas_result = self.spark.range(1).select(
+ pandas_udf(lambda values: values,
CharType(4))(lit("ab")).alias("c")
+ )
+ self.assertEqual(pandas_result.first().c, "ab ")
+
+ invalid = self.spark.range(1).select(
+ udf(lambda _: "abcd", VarcharType(3), useArrow=True)("id")
+ )
+ with self.assertRaisesRegex(Exception, "EXCEED_LIMIT_LENGTH"):
+ invalid.collect()
+
+ with tempfile.TemporaryDirectory() as path:
+ self.spark.range(1).write.parquet(path)
+ columnar_input = self.spark.read.parquet(path)
Review Comment:
**Non-blocking (P2):** A vectorized Parquet scan reaches the evaluator's
documented non-Arrow columnar path, so this case never exercises the new
`isArrow && !hasCharVarcharOutput` guard for `ArrowColumnVector` input. Please
add the CHAR/VARCHAR case to `ArrowColumnarPythonUDFSuite` using its
`readArrowSource` fixture, assert that `ArrowEvalPythonExec` still has an
Arrow-backed columnar child, and cover both padding and over-length rejection.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowConverters.scala:
##########
@@ -557,13 +559,15 @@ 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())
+ val projection = UnsafeProjection.create(checkedAttrs, attrs)
+ rows.map(row => projection(row).copy(): InternalRow)
Review Comment:
**Non-blocking (P2):** For schemas without CHAR/VARCHAR, `checkedAttrs` is
identical to `attrs`, but this large-input RDD branch still runs every row
through an `UnsafeProjection` and then deep-copies it. That adds CPU and
allocations to all Arrow-backed DataFrame creation above
`arrowLocalRelationThreshold`, even when the feature is unused. Please retain
the original direct `fromBatchIterator` path unless `hasCharVarchar(schema)`
requires the checked projection.
##########
python/pyspark/sql/pandas/types.py:
##########
@@ -133,7 +135,7 @@ def to_arrow_type(
arrow_type = pa.float64()
elif isinstance(dt, DecimalType):
arrow_type = pa.decimal128(dt.precision, dt.scale)
- elif isinstance(dt, StringType):
+ elif isinstance(dt, (StringType, CharType, VarcharType)):
Review Comment:
**Blocking (P1):** This shared mapping is also the capability check for
`mapInPandas`, `mapInArrow`, grouped/cogrouped map, and aggregate UDFs. Those
JVM output consumers still use identity projections and never call
`stringLengthCheck`, so a declared `VarcharType(3)` can return `"abcd"`
unchanged and an under-length CHAR remains unpadded. Please keep CHAR/VARCHAR
rejected for non-scalar eval types until each corresponding output consumer
converts from physical STRING and applies the recursive assignment checks.
**Recommended change:** Limit the new CHAR/VARCHAR acceptance to scalar eval
types and other boundaries whose consumers already enforce assignment semantics.
**Why this works:** Separate Arrow transport mapping from eval-type
capability validation and recursively reject CHAR/VARCHAR in non-scalar return
schemas before calling the shared mapper.
**Scope:** PySpark UDF return-type validation and focused negative tests for
map, grouped, cogrouped, and aggregate pandas/Arrow eval types.
**Compatibility:** Preserves newly implemented scalar and DataFrame/Arrow
support while restoring the prior unsupported-type failure for non-scalar APIs
that cannot yet enforce the contract.
**Risks:** Capability checks can drift from JVM support if eval-type
ownership is not kept explicit.
**Constraints:** Preserve recursive CHAR/VARCHAR detection inside structs,
arrays, and maps. Do not disable the shared mapping used by DataFrame creation,
toArrow, or supported scalar UDF paths.
**Success:** Every accepted CHAR/VARCHAR producer applies recursive padding
and overflow checks; non-scalar eval types remain rejected until their JVM
output paths provide those semantics.
--
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]
