srielau commented on code in PR #58549:
URL: https://github.com/apache/spark/pull/58549#discussion_r4051344920
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/ColumnarArrowEvalPythonEvaluatorFactory.scala:
##########
@@ -190,10 +216,22 @@ private[python] class
ColumnarArrowEvalPythonEvaluatorFactory(
val passThruQueue =
new ArrayDeque[(Array[ColumnVector], Int)]()
Review Comment:
With pipelined execution enabled, `bufferedIter` adds entries from the
writer thread while `resultIter` peeks and polls them from the task thread.
`ArrayDeque` provides no synchronization or memory publication, unlike the
`HybridRowQueue` used by paths 2 and 3, so this can observe an empty or stale
queue. Please use a thread-safe SPSC/concurrent queue and add Arrow-columnar
pipelined coverage.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/ColumnarArrowEvalPythonEvaluatorFactory.scala:
##########
@@ -190,10 +216,22 @@ private[python] class
ColumnarArrowEvalPythonEvaluatorFactory(
val passThruQueue =
new ArrayDeque[(Array[ColumnVector], Int)]()
+ context.addTaskCompletionListener[Unit] { _ =>
+ while (!passThruQueue.isEmpty) {
+ passThruQueue.poll()._1.foreach(_.close())
+ }
+ }
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:
This new lifetime explanation contradicts the existing Path 1 Scaladoc,
which still says raw `ColumnVector` references are safe because batches
allocate independently. The implementation now creates transferred views
specifically because the reader can close the originals first. Please update
both the class-level and method-level descriptions.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/ColumnarArrowEvalPythonEvaluatorFactory.scala:
##########
@@ -190,10 +216,22 @@ private[python] class
ColumnarArrowEvalPythonEvaluatorFactory(
val passThruQueue =
new ArrayDeque[(Array[ColumnVector], Int)]()
+ context.addTaskCompletionListener[Unit] { _ =>
+ while (!passThruQueue.isEmpty) {
+ passThruQueue.poll()._1.foreach(_.close())
+ }
+ }
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
+ // the input iterator. Create independent vector views whose buffers
remain valid until
+ // the corresponding output batch is closed.
+ val passThruCols = childOutput.indices.map { i =>
+ val vector =
batch.column(i).asInstanceOf[ArrowColumnVector].getValueVector
Review Comment:
The optimized-path gate checks only that column 0 is Arrow-backed, but this
loop now casts every child column. A legal mixed `ColumnarBatch`, such as an
Arrow-backed scan with an appended constant or metadata vector, enters this
path and then fails with `ClassCastException`. Require every pass-through
column to be an `ArrowColumnVector` before choosing path 1; otherwise use the
row-queue 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]
