srielau commented on code in PR #58549:
URL: https://github.com/apache/spark/pull/58549#discussion_r4086850996
##########
python/pyspark/sql/types.py:
##########
@@ -3051,8 +3051,22 @@ def _has_type(dt: DataType, dts: Union[type, Tuple[type,
...]]) -> bool:
return _has_type(dt.elementType, dts)
elif isinstance(dt, MapType):
return _has_type(dt.keyType, dts) or _has_type(dt.valueType, dts)
+ else:
Review Comment:
Fixed in 69ef259f915. `_has_type` again descends through UDT storage,
preserving the nanosecond map-key guard and the existing year-month/pandas
preflight contracts. Logical CHAR/VARCHAR capability checks now use a separate
`_has_logical_type` helper, with focused helper coverage.
##########
python/pyspark/sql/tests/arrow/test_arrow.py:
##########
@@ -1935,7 +1944,143 @@ def
test_toPandas_nested_array_with_map_empty_outer(self):
pandas_requirement_message or pyarrow_requirement_message,
)
class ArrowTests(ArrowTestsMixin, ReusedSQLTestCase):
- pass
+ # These CHAR/VARCHAR cases are Classic-only: they exercise
standard-semantics
+ # createDataFrame/toArrow and assign a UDT-backed `_schema` directly,
neither of which is
Review Comment:
Fixed in 69ef259f915. The class comment now accurately says that the
Classic-only test overrides the cached `schema` attribute; it no longer refers
to Connect `_schema`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/ColumnarArrowEvalPythonEvaluatorFactory.scala:
##########
@@ -27,27 +27,43 @@ import org.apache.spark.api.python.ChainedPythonFunctions
import org.apache.spark.internal.config.Python.PYTHON_UDF_PIPELINED_EXECUTION
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util.CharVarcharUtils
import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.execution.RowToColumnConverter
import org.apache.spark.sql.execution.metric.SQLMetric
import org.apache.spark.sql.execution.python.EvalPythonExec.ArgumentMetadata
import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector
-import org.apache.spark.sql.types.{DataType, StructField, StructType,
UserDefinedType}
+import org.apache.spark.sql.types.{DataType, StructField, StructType}
import org.apache.spark.sql.types.DataType.equalsIgnoreCompatibleCollation
+import org.apache.spark.sql.util.ArrowUtils
import org.apache.spark.sql.vectorized.{ArrowColumnVector, ColumnarBatch,
ColumnVector}
import org.apache.spark.util.Utils
+private[python] object ColumnarArrowEvalPythonEvaluatorFactory {
+ def toArrowPhysicalType(dataType: DataType): DataType =
+ CharVarcharUtils.replaceCharVarcharWithStringForPhysicalType(dataType)
+
+ def canUseArrowColumnar(
+ inputColumnIndices: Option[Array[Int]],
+ isArrow: Boolean,
+ udfs: Seq[PythonUDF]): Boolean = {
+ inputColumnIndices.isDefined &&
+ isArrow &&
+ !udfs.exists(_.hasCharVarcharResult)
+ }
+}
+
/**
* Evaluator factory for Arrow Python UDFs: ColumnarBatch in, ColumnarBatch
out.
*
* Three execution paths based on input characteristics:
*
* 1. '''Arrow columnar path''' (UDF inputs are simple column refs AND
* columns are [[ArrowColumnVector]]): Arrow FieldVectors are extracted
- * directly and serialized to IPC. Pass-through columns are kept as
- * [[ColumnVector]] references (safe because Arrow vectors are
- * independently allocated per batch). Output is produced by columnar
- * combining: passThruCols ++ resultCols -> ColumnarBatch.
+ * directly and serialized to IPC. Pass-through columns are transferred
Review Comment:
Fixed in 69ef259f915. The Path 1 documentation now includes the checked
CHAR/VARCHAR result exclusion alongside the simple-column and Arrow-vector
requirements.
##########
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)) {
+ Some(dataType)
+ } else {
+ None
+ }
+ PythonUDF(
+ name,
+ func,
+ resolvedDataType,
+ e,
+ pythonEvalType,
+ udfDeterministic,
+ charVarcharCheckedResultType = charVarcharCheckedResultType)
Review Comment:
Fixed in 69ef259f915. `ConvertToCatalyst` now retains the original Python
expression when its `PythonUDF` owns checked CHAR/VARCHAR semantics, so
transpilation cannot bypass padding or overflow validation. Added focused
optimizer coverage.
--
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]
