fix: let a Comet plan's output be read on a thread other than the task's

[msafonov]

Problem

CometExecIterator.getNextBatch called TaskContext.get() on every batch. When a consumer reads a partition's iterator on its own thread, TaskContext.get() returns null there, and the first batch fails with a NullPointerException in CometTaskContextShim.killTaskIfInterrupted.

The datasci ClickHouse RowBinary writer is one such consumer: insertRowsRowBinary reads the rows inside ClickHouseWriter.write, and the ClickHouse client calls that on its own thread. The client swallowed the NPE, so each insert wrote 0 rows while the Spark task still succeeded. This happened to platform.app-performance and platform.published_product_current on the night of 2026-10-15 (joom-1.1-10): Expected count is 1309796, actual 0. The job that counted partition sizes read every row because it ran on the task thread. The shuffle blocks were fetched (fetching starts in compute, on the task thread), but 0 records were decoded.

Vanilla Spark handles this case. apache/main has the same per-batch TaskContext.get().

Fix

Capture the task's TaskContext once, in the iterator's constructor, which always runs on the task thread. Use that captured context everywhere the iterator needs one: getNextBatch, the kill watch, the completion listener and createPlan.

Tests

• New CometOffTaskThreadSuite (added to the exec CI group). It covers a reused native or JVM Comet shuffle with AQE on and off, using the same pattern as the ClickHouse writer: repartitionByRange → .rdd → a sizes job → runJob on single partitions from a thread pool, with each partition's iterator read on a new thread. It also covers a native scan read on a foreign thread. Without the fix all 5 tests fail with the NPE; with the fix all 5 pass.
• Locally (Spark 3.5, libcomet 1.1-10, native unchanged), all with 0 failures:
  • CometExecSuite, CometNativeShuffleSuite, CometShuffleSuite, DisableAQECometShuffleSuite, CometShuffleReadCoalesceSuite, CometShuffleManagerSuite and CometShuffleEncryptionSuite
  • CometNativeTaskKillSuite and CometTaskMetricsSuite
  • CometNativeColumnarToRowSuite, CometJoinSuite and CometAggregateSuite

🤖 Generated with Claude Code