mixermt opened a new issue, #6091:
URL: https://github.com/apache/datafusion-comet/issues/6091
### Describe the bug
`Java_org_apache_comet_Native_executePlan` has two execution paths. A plan
with no JVM-fed inputs runs on a background Tokio task and the executor thread
parks in `rx.blocking_recv()` until a batch arrives. A plan with any JVM-fed
input (a `ScanExec` fed by a broadcast exchange or by
`CometSparkRowToColumnar`, a shuffle read, ...) instead runs this loop in
`native/core/src/execution/jni_api.rs`:
```rust
loop {
let poll_output = poll!(stream.next());
match poll_output {
Poll::Ready(Some(batch)) => return prepare_output(...),
Poll::Ready(None) => return Ok(-1),
Poll::Pending => tokio::task::block_in_place(||
pull_input_batches(exec_context))?,
}
}
```
`ScanExec` and `ShuffleScanExec` return `Poll::Pending` without registering
a waker when their buffer is empty and rely on this loop to refill them, which
is why the loop cannot simply await the stream. But `pull_input_batches` is a
no-op once every JVM-fed scan already holds an unconsumed batch or has reached
EOF. In that state, whenever the native stream is pending on asynchronous I/O
(a Parquet or Iceberg scan reading from S3 or HDFS), the loop re-polls
immediately: poll, `Pending`, no-op pull, poll, ... at 100% CPU on the executor
thread for as long as the read takes.
The common shape is a broadcast hash join over a native scan. The build side
is a JVM-fed `ScanExec` that reaches EOF after its first batches, after which
every I/O wait on the probe side becomes a spin.
Observed on a production workload (Iceberg table on HDFS read through the
native scan with the HDFS support from #5898, joined with a broadcast relation):
- the scan stages consumed 87.3 core-hours of CPU, against 6.6 core-hours
for the same stages on Spark alone, with every executor core pegged for the
duration of the stage;
- stage wall time was 3x to 6x longer than Spark's;
- with the cores saturated, HDFS reads on the Tokio workers started failing
with `Datanode connection closed while waiting for ack`, and the retries added
roughly 39 hours of aggregate task stall time.
### Steps to reproduce
1. Run a query whose native plan combines a JVM-fed input with a native scan
on high-latency storage, for example a broadcast hash join between a large
Parquet or Iceberg table on S3 or HDFS and a small dimension table.
2. Watch executor CPU while the stage runs: each running task holds one core
at 100% even though the task is waiting on storage reads.
3. Compare with the same scan executed without a JVM-fed input in the plan:
the executor thread parks in `blocking_recv()` and CPU drops to decode time.
### Expected behavior
When the JVM-fed scans have nothing to deliver and the native stream is
pending on I/O, the executor thread should sleep until the stream's waker
fires, as the background-task path already does, so CPU time tracks decode work
rather than I/O wait.
### Additional context
Proposed fix, contained in `jni_api.rs` and the two scan operators:
- `pull_input_batches`, `ScanExec::get_next_batch` and
`ShuffleScanExec::get_next_batch` report whether a buffer was actually refilled.
- When the stream is `Pending` and nothing was pulled, the loop parks the
`block_on` task until a waker registered by that poll fires, instead of
re-polling. Awaiting the stream directly would not be safe: an operator can
drain a JVM-fed buffer and re-poll that scan within a single poll, producing a
`Pending` that carries no waker, so the await would never resume. Parking for
one wake-up and then re-entering the loop keeps the refill step reachable.
- A short safety timeout on the park (100 ms) turns any future
waker-contract violation into a slow poll rather than a hang.
The change is implemented with unit tests and verified against the broadcast
join, Iceberg DPP and task metrics suites; a PR will follow.
A related but separate finding from the same workload:
`IcebergScanExec::execute_with_tasks` builds a fresh `FileIO`, and with it a
fresh storage client, for every task. That will be filed separately.
--
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]
