cloud-fan commented on code in PR #58549:
URL: https://github.com/apache/spark/pull/58549#discussion_r4053513623
##########
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:
**Nit (P3):** This test does not assign `_schema`; it overrides Classic's
cached `schema` attribute through `df.__dict__["schema"]`. Connect's read-only
`_schema` is a different property, so the current explanation obscures the
actual reason this case is Classic-only. Could the comment name the cached
`schema` override explicitly?
##########
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:
**Nit (P3):** The Path 1 description omits the
`!udfs.exists(_.hasCharVarcharResult)` gate in `canUseArrowColumnar`. A checked
CHAR/VARCHAR result therefore takes the row-queue path even when inputs are
simple Arrow-backed columns. Please include that condition so the documented
path selection matches the executable predicate.
##########
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:
**Blocking (P1):** `_first_timestamp_nanos_map_key_type` still relies on
`_has_type` to prove that an entire map-key subtree has no nanosecond
timestamp. Since this change makes `_has_type` stop at a UDT, a UDT-backed
nanosecond key bypasses the guard and two distinct entries can collapse into
one Python `datetime` key during collection. Please restore the established
physical recursion and keep logical CHAR/VARCHAR capability checks on a
separately named helper.
##########
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:
**Blocking (P1):** `charVarcharCheckedResultType` is carried only by the
inner `PythonUDF`, but `ConvertToCatalyst` can replace the enclosing transpiled
node with its Catalyst alternative and discard that boundary. With ANSI mode
and UDF transpilation enabled, a checked `VARCHAR(3)` result can then be
truncated instead of raising `EXCEED_LIMIT_LENGTH`, and CHAR padding can be
skipped. Please retain the original Python expression whenever its result owns
checked CHAR/VARCHAR semantics.
##########
python/pyspark/sql/types.py:
##########
@@ -2877,8 +2877,38 @@ 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:
This regressed during the later conflict resolution: the current head again
makes `_has_type` stop at UDTs, while `ArrowTableToRowsConversion.convert`
still relies on physical recursion. Please restore the prior `_has_type`
contract and keep the logical CHAR/VARCHAR allow-list on a separate helper.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4043362978","thread_id":"inline:4043362978","verdict_sha256":"0bab4ded9b5ac5c3195a6a7257a0eb472f32efb4077ff10ae3625eec3825bee7"}
-->
--
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]
