You are viewing a plain text version of this content. The canonical link for it is here.

Posted to github@datafusion.apache.org by "andygrove (via GitHub)" <gi...@apache.org> on 2026/09/28 00:04:06 UTC

[I] Cancelling a task doesn't stop its native plan until the plan produces its next batch [datafusion-comet]

andygrove opened a new issue, #6295:
URL: https://github.com/apache/datafusion-comet/issues/6295

### Describe the bug

When Spark kills a task it interrupts the task thread, and the task stops at its next interruption check. A Comet task waiting for its native plan's next batch is parked in native code instead, in `blocking_recv` for a plan with no JVM input or in `Handle::block_on` otherwise (`jni_api.rs:1200` and `1226`). `Thread.interrupt` wakes neither. The task stays there until the plan produces a batch or finishes, and it keeps its core and its memory until then. A plan with no JVM input also keeps running on its Tokio task after that, until it next tries to send. That is the cause of #2453, which #6261 fixes.

Measured on `main` at `634e37d08` (`local[4]`, one 2.4M-row file, `setJobGroup(..., interruptOnCancel = true)`):

- **Native scan feeding a native sort, read with `foreachPartition(_.hasNext)`, cancelled 2 s after the task started:** the job failed 49 ms after the cancel, but the task kept running for another 13.1 s.
- **The same query with Comet disabled:** the task ended 324 ms after the cancel.
- **Native scan feeding a native shuffle write, cancelled after 1.5 s:** the map task ran to completion, 16.0 s after the cancel.

Anything that relies on killing tasks to free their slots waits on this: job cancellation, speculative copies that lose, and stages that are no longer needed.

### Expected behavior

A killed task stops its native plan in roughly the time Spark takes to stop its own operators.

### Additional context

`CelebornNativeShuffleDestination.watchForCancellation` (`CometNativeShuffleWriter.scala:564-580`) already polls `TaskContext.isInterrupted()` on a scheduled executor and aborts the pusher. The same pattern could call a native entry point that cancels the plan. For a plan with no JVM input, that entry point could stop the producer and drop its stream, as #6261's `BatchProducer::stop` does. For the `block_on` path, it could set a flag that makes `next_batch` return an error and wake it. #4175 covers the narrower case of JVM UDF dispatch.

--
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: github-unsubscribe@datafusion.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [I] Cancelling a task doesn't stop its native plan until the plan produces its next batch [datafusion-comet]

Posted by "viirya (via GitHub)" <gi...@apache.org>.

viirya commented on issue #6295:
URL: https://github.com/apache/datafusion-comet/issues/6295#issuecomment-6510298337

#7181 adds a kill path for native Arrow UDFs, and part of it could be reused for plan cancellation. `CometExecIterator` binds a token to the plan with `Native.setPythonUdfTask` before the plan first executes. `PythonUdfMonitor` polls `TaskContext.isInterrupted` with one thread per task and passes that token to `NativeBase.killPythonUdfs`. Two details there apply to cancelling a plan too:

- The token is generated by the JVM because a task attempt id is unique only within one SparkContext.
- The watch has to keep running after the task completes, until the task's plans are released. Spark marks the task completed before its completion listeners release the plan, and releasing a plan waits for a batch that is still being computed.

The rest is specific to Python UDFs and would not carry over: the `hasPythonUdf` gate, the `spark.python.task.killTimeout` grace period, and `spark.comet.exec.nativeArrowPythonUDF.interruptOnKill.enabled`.

--
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: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [I] Cancelling a task doesn't stop its native plan until the plan produces its next batch [datafusion-comet]

Posted by "andygrove (via GitHub)" <gi...@apache.org>.

andygrove commented on issue #6295:
URL: https://github.com/apache/datafusion-comet/issues/6295#issuecomment-6223479588

   Measured again on `main` at c230bf4e98 (release build, Spark 4.1.3, `local[8]`). Each query is cancelled with `cancelJobGroup` 2 s after it starts, with `interruptOnCancel = true`. The table is the time from the cancel to the last task of the job ending, from two runs each.
   
   | Query (203-column, 4M-row Parquet table, 8 tasks of about 400 MB) | Comet | Spark |
   |---|---:|---:|
   | native scan feeding a native hash shuffle write (`REPARTITION(32, id)`) | 2.2 to 2.7 s | 51 to 260 ms |
   | native scan feeding a native sort (`ORDER BY`) | 4.2 to 4.4 s | 2 to 5 ms |
   
   The job failed within 70 ms of the cancel in both engines, so Comet's slots stay busy after the job is reported cancelled. The Comet times are each task's remaining work: every task ran to completion. That is the map-stage and sort counterpart of the direct-read kill check #6805 added.
   


-- 
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: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [I] Cancelling a task doesn't stop its native plan until the plan produces its next batch [datafusion-comet]

Posted by "rich7420 (via GitHub)" <gi...@apache.org>.

rich7420 commented on issue #6295:
URL: https://github.com/apache/datafusion-comet/issues/6295#issuecomment-6444621871

   Prototype: after cancellation, `executePlan` returned within 27.3ms; `releasePlan` took ~1.48s.
   


-- 
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: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org