srielau commented on code in PR #58549:
URL: https://github.com/apache/spark/pull/58549#discussion_r4099933629
##########
python/pyspark/sql/pandas/types.py:
##########
@@ -988,7 +990,7 @@ def _to_corrected_pandas_ext_type(dt: DataType) ->
Optional[Any]:
return pd.Float64Dtype()
elif isinstance(dt, BooleanType):
return pd.BooleanDtype()
- elif isinstance(dt, StringType):
+ elif isinstance(dt, (StringType, CharType, VarcharType)):
Review Comment:
This branch cannot run: `_to_corrected_pandas_ext_type` is only called
inside an `isinstance(data_type, IntegralType)` branch, so CHAR/VARCHAR can
never reach it.
```suggestion
elif isinstance(dt, StringType):
```
##########
python/pyspark/sql/pandas/types.py:
##########
@@ -157,7 +159,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:
`SparkArrowCStreamer` calls `to_arrow_schema(df.schema)` and then uses
`mapInArrow` with a binary result schema, so none of the new output capability
gates run. This changes `DataFrame.__arrow_c_stream__` from rejection to
positive CHAR/VARCHAR support even though the scope fence permits only
`toArrow`. Please add a recursive consumer-local rejection, including UDT
storage, and a C-stream regression.
##########
python/pyspark/sql/pandas/types.py:
##########
@@ -959,7 +961,7 @@ def _to_corrected_pandas_type(dt: DataType) ->
Optional[Any]:
return np.dtype("timedelta64[ns]")
else:
return np.dtype("timedelta64[us]")
- elif isinstance(dt, StringType):
+ elif isinstance(dt, (StringType, CharType, VarcharType)):
Review Comment:
`_create_converter_to_pandas` is also used by both Arrow and row
`DataFrame.toPandas()` paths. On pandas 3 this now forces first-class
CHAR/VARCHAR columns to `StringDtype(na_value=np.nan)` where they were
previously left uncorrected, creating a public behavior change outside the
stated `toArrow` and scalar-UDF scope. Please keep this correction caller-local
to worker paths, or explicitly document and directly test the `toPandas()`
contract.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/ColumnarArrowEvalPythonEvaluatorFactory.scala:
##########
@@ -212,10 +252,11 @@ private[python] class
ColumnarArrowEvalPythonEvaluatorFactory(
val numRows = resultBatch.numRows()
val resultCols = (0 until resultBatch.numCols()).map(
i => resultBatch.column(i)).toArray
- val (passThruCols, passThruRows) = passThruQueue.poll()
+ val passThruRows = passThruQueue.peek()._2
Review Comment:
Scalar pandas/Arrow iterator UDFs explicitly allow prefetching and
rebatching as long as the total output row count matches, and the worker
validates only that total. Pairing every result batch with exactly one queued
input batch makes legal split or combined outputs fail this assertion; a
zero-row result on an empty partition also dereferences an empty queue here.
Please either exclude iterator eval types from path 1 so the row queue joins by
row, or re-batch the pass-through vectors to the output boundaries, with a
rechunking regression.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/UserDefinedPythonFunction.scala:
##########
@@ -108,7 +108,27 @@ case class UserDefinedPythonFunction(
}
PythonAggregate(name, func, dataType, e, udfDeterministic, bufferStruct)
} else {
- PythonUDF(name, func, dataType, e, pythonEvalType, udfDeterministic)
+ val conf = SQLConf.get
+ val resolvedDataType = if (conf.charVarcharFirstClassTypes) {
+ dataType
+ } else {
+ CharVarcharUtils.replaceCharVarcharWithString(dataType)
+ }
+ val charVarcharCheckedResultType =
+ if (CharVarcharUtils.shouldApplyWriteSideLengthCheck(conf) &&
+ CharVarcharUtils.hasCharVarchar(dataType)) {
Review Comment:
The UDT-storage rejection currently exists only in PySpark
`_check_return_type`, but Connect reconstructs this object directly in
`SparkConnectPlanner.createUserDefinedPythonFunction`. Because
`hasCharVarchar(dataType)` does not descend through `UserDefinedType.sqlType`,
a direct proto can reach Arrow with physical STRING storage and no checked
result type. Please add authoritative JVM-side UDT-storage validation before
the eval-type branches and cover the Connect/JVM construction path.
--
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]
