mixermt opened a new pull request, #6092: URL: https://github.com/apache/datafusion-comet/pull/6092
## Which issue does this PR close? Closes #6091. ## Rationale for this change `Java_org_apache_comet_Native_executePlan` runs a plan that has any JVM-fed input (a broadcast build side, `CometSparkRowToColumnar`, a shuffle read) through a loop that polls the stream and, on `Pending`, pulls the next batches from the JVM. That pull blocks inside JNI while the JVM produces data, so the loop never spun as long as the JVM was the only thing worth waiting for. With native scans reading from S3 or HDFS the stream is also pending on asynchronous I/O, and once every JVM-fed scan holds a batch or has reached EOF the pull is a no-op: the loop re-polls at full speed for the duration of every read, pinning one core per task. On a production workload (Iceberg on HDFS joined with a broadcast relation) the scan stages used 87 core-hours against 6.6 for Spark alone, ran 3x to 6x longer, and the saturated cores caused HDFS ack timeouts and retries. Details in #6091. ## What changes are included in this PR? - `ScanExec::get_next_batch`, `ShuffleScanExec::get_next_batch` and `pull_input_batches` return whether a buffer was actually refilled. - In the ScanExec path of `executePlan`, a `Pending` with nothing pulled parks the `block_on` task until a waker registered by that poll fires (`park_until_woken`), then re-enters the loop. The park has a 100 ms safety timeout so a future waker-contract violation degrades to a slow poll rather than a hang. - Awaiting the stream directly was rejected: `ScanExec` returns `Pending` without a waker when an operator drains and re-polls it within a single poll, so that await could never resume. Parking for one wake-up keeps the JVM refill step reachable. - The metrics update interval is checked on every iteration instead of every 100 polls, since iterations are no longer spins; the poll counter is removed. - `tokio`'s `time` feature is enabled explicitly for the park timeout. ## How are these changes tested? - New unit test `park_until_woken_ends_on_a_registered_waker_or_the_timeout` in `jni_api.rs`: a waker registered by an earlier poll ends the park promptly, and with no waker the timeout ends it. - Existing Rust tests for the scan operators pass; `cargo clippy --all-targets -p datafusion-comet -- -D warnings` is clean. - JVM, with the rebuilt library: the `CometJoinSuite` broadcast hash and nested-loop join tests, the Iceberg DPP join test in `CometIcebergNativeSuite`, and the full `CometTaskMetricsSuite` (native scans, shuffle scans, spills). 35 tests, no hangs. - Not measured here: the CPU reduction on the production workload itself, which needs a run with this build. This touches the native execution loop, so the Spark SQL suites (`run-spark-4.1-tests`) are worth running before merge. ## AI Disclosure Drafted, implemented and tested with AI assistance (Claude Code); reviewed before submission. 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- 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]
