Weixin-Xu commented on code in PR #13492:
URL: https://github.com/apache/gluten/pull/13492#discussion_r4421524193
##########
backends-bolt/src/main/scala/org/apache/spark/api/python/ColumnarArrowEvalPythonExec.scala:
##########
@@ -103,9 +101,20 @@ class ColumnarArrowPythonRunner(
if (reader != null && batchLoaded) {
batchLoaded = reader.loadNextBatch()
if (batchLoaded) {
- val batch = new ColumnarBatch(vectors)
- batch.setNumRows(root.getRowCount)
- batch
+ // The reader reuses root across batches. Give each result its
own vectors and
+ // retained buffer references so prefetch and reader.close
cannot invalidate it.
+ val recordBatch = new VectorUnloader(root).getRecordBatch
+ val outputRoot = VectorSchemaRoot.create(root.getSchema,
allocator)
+ try {
+ new VectorLoader(outputRoot).load(recordBatch)
+ ArrowColumnVectors.toBatch(outputRoot)
Review Comment:
The reader now tracks the pending output batch while it still owns it.
Ownership tracking is cleared when next() transfers the batch downstream, and
teardown closes any unconsumed batch. Added an early-stop regression that calls
hasNext() to prefetch a result without consuming it.
--
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]
