wForget commented on code in PR #3930:
URL: https://github.com/apache/datafusion-comet/pull/3930#discussion_r3091301286
##########
spark/src/main/scala/org/apache/comet/CometExecIterator.scala:
##########
@@ -79,14 +80,22 @@ class CometExecIterator(
private val taskAttemptId = TaskContext.get().taskAttemptId
private val taskCPUs = TaskContext.get().cpus()
private val cometTaskMemoryManager = new CometTaskMemoryManager(id,
taskAttemptId)
+ // When true, executePlan stashes output batches natively and returns handles
+ // instead of exporting via Arrow FFI. Used when output feeds a native
ShuffleWriter.
+ private var stashMode: Boolean = false
+ private var pendingHandle: Long = -1L
// Build a mixed array of iterators: CometShuffleBlockIterator for shuffle
// scan indices, CometBatchIterator for regular scan indices.
- private val inputIterators: Array[Object] = inputs.zipWithIndex.map {
- case (_, idx) if shuffleBlockIterators.contains(idx) =>
- shuffleBlockIterators(idx).asInstanceOf[Object]
- case (iterator, _) =>
- new CometBatchIterator(iterator, nativeUtil).asInstanceOf[Object]
- }.toArray
+ private val inputIterators: Array[Object] = if (handleInputs.nonEmpty) {
+ handleInputs
Review Comment:
assert inputs.isEmpty
--
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]
