srielau commented on code in PR #58549:
URL: https://github.com/apache/spark/pull/58549#discussion_r4051436464
##########
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:
Fixed in 14093addc31. The pass-through queue is now a ConcurrentLinkedQueue,
which safely publishes entries between the pipelined writer thread and task
thread. The Arrow-backed legacy-output test now explicitly enables pipelined
execution.
##########
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:
Fixed in 14093addc31. The optimized path now requires every
child/pass-through column in the batch to be an ArrowColumnVector; mixed
batches fall back to the row-queue path.
##########
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:
Fixed in 14093addc31. Both the class-level path description and
evalArrowColumnar Scaladoc now explain that pass-through vectors are
transferred to independent views because the source may close the originals
before Python output is consumed.
--
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]
