comphead opened a new pull request, #7584: URL: https://github.com/apache/datafusion-comet/pull/7584
## Which issue does this PR close? Closes #7582. ## Rationale for this change A map task whose plan is a native scan feeding a native shuffle write reported no metrics until it ended. The live Stages tab showed a blank Input column for the whole task, and the SQL tab showed zeros. On a job whose map tasks ran for over 30 minutes, nothing showed progress. ## What changes are included in this PR? - For a plan without JVM input, `execute_plan` now waits for the next batch no longer than until metrics are due, and publishes when the wait times out. A publish while waiting does not restart the interval, so the batch that ends a wait is still published as it was after #6313. With a negative `spark.comet.metrics.updateInterval` nothing changes. - The wait parks the Spark task thread instead of using a Tokio timer. A Tokio timer fires only once a runtime worker is free, and by default the runtime has one worker per executor core, which the shuffle writers' synchronous spills and final writes can all hold. With a Tokio timeout, Input in the repro below stopped updating after 2.5 s while the tasks ran until 5 s. - Each publish runs the task's in-flight reports. `CometNativeShuffleWriter` uses the new `reportScanInputMetricsWhileRunning` when its plan has no JVM input, so the scan's bytes and rows reach the task's input metrics as they grow. A plan with a JVM input keeps reporting at completion only, because a JVM scan in the same task sets its counters from a snapshot taken when it starts, which would drop what Comet added in the meantime. - Reports now claim what an accumulator gained since it was last claimed for the task, so a report while running and the one at completion count each byte once. At completion the values are final, so existing reports add the same totals as before. Shuffle Write still appears when the task ends. Its bytes are the size of the finished data file, and the UI hides the column while they are 0, as with Spark's sort-based shuffle writer. ## How are these changes tested? - New native test `execute_plan_publishes_metrics_while_it_waits_for_a_batch`. It fails with a single publish without the change. - New test in `CometTaskMetricsSuite` that input reported while a task runs is not counted again at completion. - `CometTaskMetricsSuite` (31 tests), `CometNativeShuffleSuite` (83 tests) and the `jni_api` native tests (46) pass locally. - Manually on `local-cluster` with the repro from #7582 (30M rows read by two tasks, `repartition(50)`, 500 ms update interval): Input grew from 767 MiB at 1.5 s to 2.4 GiB at 3 s while the tasks ran, where it stayed blank before. The running tasks' heartbeats carried 31 non-zero SQL metrics instead of none. Input and Shuffle Write totals at `TaskEnd` were byte-identical to the build without the change. -- 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]
