srielau commented on code in PR #58549:
URL: https://github.com/apache/spark/pull/58549#discussion_r4109944386
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -1316,7 +1316,13 @@ object ConvertToCatalyst extends Rule[LogicalPlan] {
def keepPython: Expression = s.pythonUDFExpr.mapChildren(recurse(_,
parentIsUdf = true))
// We _shouldn't_ have these nodes if ANSI is not enabled or
transpilation is disabled
// but if someone changed it while running we'll want to strip the
nodes out.
- if (!conf.getConf(SQLConf.ANSI_ENABLED)) {
+ val hasCheckedCharVarcharResult = s.pythonUDFExpr match {
+ case udf: PythonUDF => udf.hasCharVarcharResult
+ case _ => false
+ }
+ if (hasCheckedCharVarcharResult) {
+ s.pythonUDFExpr.mapChildren(applyExpr(_, parentIsUdf = true))
Review Comment:
Fixed. The checked CHAR/VARCHAR arm now uses `keepPython`, so nested calls
keep `inLambda` and `preEvaluate`. Added `ConvertToCatalystSuite` coverage for
a checked UDF wrapping a nested call inside a lambda.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/ColumnarArrowEvalPythonEvaluatorFactory.scala:
##########
@@ -189,11 +215,25 @@ private[python] class
ColumnarArrowEvalPythonEvaluatorFactory(
columnIndices: Array[Int]): Iterator[ColumnarBatch] = {
val passThruQueue =
- new ArrayDeque[(Array[ColumnVector], Int)]()
+ new ConcurrentLinkedQueue[(Array[ColumnVector], Int)]()
+ context.addTaskCompletionListener[Unit] { _ =>
+ var entry = passThruQueue.poll()
+ while (entry != null) {
+ entry._1.foreach(_.close())
+ entry = passThruQueue.poll()
+ }
+ }
val bufferedIter = inputIter.map { batch =>
- val passThruCols = childOutput.indices.map(
- i => batch.column(i)).toArray
+ // The input reader owns and may close its vectors as soon as the
Python runner consumes
Review Comment:
Fixed. Partial `splitAndTransfer` construction now closes already-allocated
vectors, and a closed flag plus post-enqueue drain closes late or cancelled
publications instead of leaking them.
##########
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:
Fixed without widening assignment. `udaf()` now rejects CHAR/VARCHAR in
`bufferSchema` (including UDT storage), and the JVM `UserDefinedPythonFunction`
builder rejects the same for Connect/direct construction. Covered in the Python
aggregator mixin and `ExtractPythonUDFsSuite`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/PythonUDF.scala:
##########
@@ -456,9 +456,14 @@ 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,
+ // The original CHAR/VARCHAR result type when write-side checks apply.
This is absent for
+ // unconstrained results, so the CHAR/VARCHAR policy is not part of
PythonUDF equality.
Review Comment:
Fixed the comment: generated equality includes
`charVarcharCheckedResultType`, and canonicalized copies retain it, so checked
and unchecked UDFs stay distinct.
##########
python/pyspark/sql/tests/arrow/test_arrow.py:
##########
@@ -1935,7 +1944,158 @@ 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
Review Comment:
Moved the reusable pandas/PyArrow createDataFrame and toArrow CHAR/VARCHAR
cases into `ArrowTestsMixin` so Connect parity inherits them. Classic-only
coverage (RDD `localRelationThreshold=0`, UDT cached-schema, and
`__arrow_c_stream__`) stays on `ArrowTests`.
--
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]
