cloud-fan commented on code in PR #58549:
URL: https://github.com/apache/spark/pull/58549#discussion_r4113743112
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/ColumnarArrowEvalPythonEvaluatorFactory.scala:
##########
@@ -189,12 +216,64 @@ private[python] class
ColumnarArrowEvalPythonEvaluatorFactory(
columnIndices: Array[Int]): Iterator[ColumnarBatch] = {
val passThruQueue =
- new ArrayDeque[(Array[ColumnVector], Int)]()
+ new ConcurrentLinkedQueue[(Array[ColumnVector], Int)]()
+ val passThruClosed = new AtomicBoolean(false)
+
+ def closeCols(cols: Array[ColumnVector]): Unit = {
+ var i = 0
+ while (i < cols.length) {
+ cols(i).close()
+ i += 1
+ }
+ }
+
+ def drainPassThru(): Unit = {
+ passThruClosed.set(true)
+ var entry = passThruQueue.poll()
+ while (entry != null) {
+ closeCols(entry._1)
+ entry = passThruQueue.poll()
+ }
+ }
+
+ context.addTaskCompletionListener[Unit] { _ =>
Review Comment:
**Non-blocking (P2):** The added tests consume one input batch completely,
so they never exercise repeated queue pairing or task completion while
transferred vectors remain queued. A multi-batch case plus an early-termination
case is needed to make ordering errors and completion-time native-vector leaks
regression-sensitive.
**Verification:**
- **Regression:** Drive at least two Arrow input batches in one task and
assert pass-through values align with every Python result batch.
- **Regression:** Terminate consumption with queued batches outstanding and
assert task completion returns native allocation to baseline.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/ColumnarArrowEvalPythonEvaluatorFactory.scala:
##########
@@ -189,12 +216,64 @@ private[python] class
ColumnarArrowEvalPythonEvaluatorFactory(
columnIndices: Array[Int]): Iterator[ColumnarBatch] = {
val passThruQueue =
- new ArrayDeque[(Array[ColumnVector], Int)]()
+ new ConcurrentLinkedQueue[(Array[ColumnVector], Int)]()
+ val passThruClosed = new AtomicBoolean(false)
+
+ def closeCols(cols: Array[ColumnVector]): Unit = {
+ var i = 0
+ while (i < cols.length) {
+ cols(i).close()
+ i += 1
+ }
+ }
+
+ def drainPassThru(): Unit = {
+ passThruClosed.set(true)
+ var entry = passThruQueue.poll()
+ while (entry != null) {
+ closeCols(entry._1)
+ entry = passThruQueue.poll()
+ }
+ }
+
+ context.addTaskCompletionListener[Unit] { _ =>
+ drainPassThru()
+ }
val bufferedIter = inputIter.map { batch =>
- val passThruCols = childOutput.indices.map(
- i => batch.column(i)).toArray
- passThruQueue.add((passThruCols, batch.numRows()))
+ // The input reader owns and may close its vectors as soon as the
Python runner consumes
+ // the input iterator. Create independent vector views whose buffers
remain valid until
+ // the corresponding output batch is closed.
+ val cols = Array.ofDim[ColumnVector](childOutput.length)
+ var allocated = 0
+ try {
+ var i = 0
+ while (i < childOutput.length) {
+ val vector =
batch.column(i).asInstanceOf[ArrowColumnVector].getValueVector
+ val transferPair = vector.getTransferPair(ArrowUtils.rootAllocator)
+ transferPair.splitAndTransfer(0, batch.numRows())
Review Comment:
**Non-blocking (P2):** `getTo` has already allocated the destination vector
here, but the catch can close it only after the later wrapper assignment
increments `allocated`. If transfer or wrapping throws, this current target
leaks off-heap memory; establish cleanup ownership immediately after acquiring
the target and release it on either failure.
**Verification:**
- **Regression:** Inject a failure after target allocation and verify all
acquired Arrow buffers return to the allocator baseline.
##########
python/pyspark/sql/tests/connect/test_parity_udf.py:
##########
@@ -46,6 +46,10 @@ def test_udf_with_input_file_name_for_hadooprdd(self):
def test_same_accumulator_in_udfs(self):
super().test_same_accumulator_in_udfs()
+ @unittest.skip("Spark Connect resolves both UDFs when the plan is
submitted.")
Review Comment:
**Non-blocking (P2):** This skip exposes a supported Connect correctness
gap: the plan carries the logical result type but not the expression-local
CHAR/VARCHAR check policy. When expressions created under different policies
are submitted together, the server rebuilds both under the final SQLConf, which
can drop CHAR padding or VARCHAR overflow validation for an individual
expression.
**Recommended change:** Serialize captured CHAR/VARCHAR checked-result state
with each Connect PythonUDF and consume it during server reconstruction.
**Why this works:** Resolve the policy when the Connect expression is
created, encode the checked logical result (or an equivalent explicit policy),
and have the server build PythonUDF from that state instead of recomputing it
from submission-time SQLConf.
**Scope:** Extend the Connect UDF representation, Python plan serialization,
JVM reconstruction, and mixed-policy parity coverage.
**Compatibility:** Pure legacy-as-string expressions remain unconstrained,
while standard and preserve-type-info expressions remain checked.
**Risks:** Wire compatibility must preserve behavior for older clients that
omit the new state.
**Constraints:** Do not change non-Connect Classic capture timing or
grouped/non-scalar capability rules.
**Success:** Each Connect scalar UDF retains its construction-time
CHAR/VARCHAR assignment policy. Checked and unchecked UDFs submitted in one
plan produce the same result as Classic.
**Verification:**
- **Regression:** Construct checked CHAR and unchecked VARCHAR expressions
under opposite policies, submit them together after a configuration change, and
assert padding and overflow behavior remain expression-local.
- **Compatibility:** Verify a client that omits the new optional state
retains the established default reconstruction behavior.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/UserDefinedPythonFunction.scala:
##########
@@ -86,6 +87,9 @@ case class UserDefinedPythonFunction(
val optionInputTypes: List[List[String]] =
transpiledInputTypes.asScala.map(_.asScala.toList).toList
+ if (CharVarcharUtils.hasCharVarcharInUDT(dataType)) {
Review Comment:
**Non-blocking (P2):** This authoritative JVM boundary only rejects
CHAR/VARCHAR hidden in UDT storage before grouped aggregates branch into
`PythonUDAF`. A direct JVM or Connect caller can still supply a logical
CHAR/VARCHAR aggregate result and bypass the Python client fence, producing
unchecked output instead of the promised unsupported-return error.
**Verification:**
- **Regression:** Construct each grouped aggregate eval type through the
authoritative JVM boundary with CHAR/VARCHAR output and assert the classed
unsupported-return error.
- **Compatibility:** Verify scalar row, Arrow, and pandas eval types
continue to accept and check supported CHAR/VARCHAR output.
##########
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:
Fixed in 69ef259f915. `_has_type` again descends through UDT storage, while
logical CHAR/VARCHAR capability checks continue to use `_has_logical_type`.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4043362978","thread_id":"inline:4043362978","verdict_sha256":"346739c2117e6dac2e9b2e51f34743d7c2bf72cb63a4aa89bf3f0b6bb75b6578"}
-->
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/UserDefinedPythonFunction.scala:
##########
@@ -106,9 +110,32 @@ case class UserDefinedPythonFunction(
throw
QueryCompilationErrors.invalidIncrementalPythonAggregatorBufferError(
name, bufferType)
}
+ if (CharVarcharUtils.physicalTypeHasCharVarchar(bufferStruct)) {
+ throw
QueryCompilationErrors.invalidPythonAggregatorBufferType(bufferStruct)
+ }
PythonAggregate(name, func, dataType, e, udfDeterministic, bufferStruct)
} else {
- PythonUDF(name, func, dataType, e, pythonEvalType, udfDeterministic)
+ val conf = SQLConf.get
+ val resolvedDataType = if (conf.charVarcharFirstClassTypes) {
Review Comment:
**Non-blocking (P2):** The policy matrix does not cover preservation enabled
together with standard semantics disabled and legacy-as-string enabled. That
interaction is the precedence case in this branch; without a scalar-UDF and
explicit-schema Arrow assertion, a regression can silently lower the schema to
STRING or remove padding and overflow checks while every adjacent mode test
stays green.
**Verification:**
- **Regression:** Disable standard semantics, enable legacy-as-string and
preservation together, then assert first-class schema exposure, padding,
overflow rejection, and checked Arrow routing.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/EvaluatePython.scala:
##########
@@ -267,19 +314,27 @@ object EvaluatePython {
}
case MapType(keyType, valueType, _) =>
- val keyFromJava = makeFromJava(keyType)
- val valueFromJava = makeFromJava(valueType)
+ val keyFromJava = makeFromJava(keyType, applyCharVarcharChecks)
+ val valueFromJava = makeFromJava(valueType, applyCharVarcharChecks)
(obj: Any) => nullSafeConvert(obj) {
case javaMap: java.util.Map[_, _] =>
- ArrayBasedMapData(
- javaMap,
- (key: Any) => keyFromJava(key),
- (value: Any) => valueFromJava(value))
+ if (applyCharVarcharChecks &&
CharVarcharUtils.hasCharVarchar(keyType)) {
Review Comment:
**Non-blocking (P2):** `keyType` and `applyCharVarcharChecks` are fixed when
this converter is built, but `hasCharVarchar(keyType)` recursively rescans the
schema for every converted map. Compute the combined branch condition once
outside the returned closure so row conversion does not repeat invariant schema
work.
**Verification:**
- **Behavior:** Exercise checked CHAR keys, unconstrained keys,
duplicate-after-padding keys, and repeated row conversion with identical
results and one schema decision.
--
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]
