mixermt commented on PR #6092: URL: https://github.com/apache/datafusion-comet/pull/6092#issuecomment-5774360736
Thanks both. Everything below is in the latest push, and the two threads changed the design rather than patching it. - **Wakers instead of a timeout** (@mbutrovich). `ScanStream` and `ShuffleScanStream` now register `cx.waker()` when the buffer is empty, and `get_next_batch` wakes it after refilling, through an `AtomicWaker` shared by the exec's clones. Every `Pending` carries a waker, so the 100 ms timeout, the `bool` return from `get_next_batch` and the `tokio` `time` feature are gone; `park_until_woken` is a `poll_fn` that yields once. I dropped the timeout rather than logging its firings: with wakers in place a firing could not be told apart from a legitimately slow read, so a counter would not have found a lost wake. - **EOF is sticky** (@andygrove). `poll_next` leaves `InputBatch::EOF` in the buffer, so a re-poll returns `Ready(None)` again and `get_next_batch` stays a no-op instead of making another JNI round trip into a drained reader. The resting-state invariant in #6091 is now what the code does. - **A loop test that fails without the fix** (@mbutrovich). The poll, pull and park steps moved into `next_batch(stream, on_pending)`; `update_metrics` and `prepare_output` stay in `executePlan`. `next_batch_parks_while_the_stream_waits_on_native_io` drives it with a stream pending on `tokio::time::sleep` and asserts the pull closure ran a few times; with the park deleted it ran 135,311 times in one 50 ms wait. `next_batch_resumes_on_a_refill_and_stops_pulling_after_eof` runs a real `ScanExec` in test mode: each park ends only on the refill's wake, under a timeout that turns a lost wake into a failure, and a re-poll after EOF pulls nothing; with EOF cleared again it fails. `refill_wakes_the_pending_poll_and_eof_stays_buffered` covers `ShuffleScanStream`. - **Tracing density** (@andygrove). `log_memory_usage` sits behind the same interval check as `update_metrics`, in `update_metrics_on_interval`, which runs on every pending poll and once per returned batch, so in-loop trace density no longer depends on how often the loop turns. - **Docs** (@mbutrovich). The JVM data source paragraph in `development.md` describes the pull-then-park loop. `poll_fn` is imported next to `task::Poll` and the `Park` struct is gone. Verified with the rebuilt library: clippy with `-D warnings` is clean, the 433 core unit tests pass, and `CometTaskMetricsSuite`, `CometJoinSuite`, `CometNativeShuffleInputRDDSuite`, `CometNativeShuffleSuite` and `CometIcebergNativeSuite` pass (235 tests, no hangs). This rewrites the native execution loop, so the Spark SQL suites should report here rather than in the merge queue. Could a committer add `run-spark-4.1-tests`? -- 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]
