dwsmith1983 opened a new pull request, #5880: URL: https://github.com/apache/datafusion-comet/pull/5880
## Which issue does this PR close? Closes #5336. ## Rationale for this change `CometMetricNode.reportScanInputMetrics` feeds the `Input` column on the Stages and Executors tabs. It registered its completion listener after `super.compute` had created the `CometExecIterator`, and Spark runs completion listeners in reverse registration order, so the report ran before the iterator's `close` published the final native metrics. A fully consumed iterator closes itself on exhaustion, which hid the problem for plain scans and joins. A block with a JVM input only publishes on `spark.comet.metrics.updateInterval`, and a consumer that stops early never exhausts the iterator: a broadcast join with `LIMIT 3` reported zero bytes and zero rows at the default interval. The listener also used `setBytesRead`/`setRecordsRead`, replacing bytes a fallback Spark scan had accumulated in the same task. ## What changes are included in this PR? - Register the report before `super.compute` at the three sites (`CometNativeScanExec`, `CometIcebergNativeScanExec`, the `CometExecRDD` override in `operators.scala`), matching what the native shuffle writer and native write already do. - Add to the task's counters with `incBytesRead`/`incRecordsRead`, and claim each accumulator once per task through the per-task identity registry `reportSpillMetrics` already uses (generalised from `SeenSpillMetrics` to `SeenMetrics`, with a shared `claimMetricValue` helper that also treats an unset size metric as zero). - Failed attempts now report too, since the listener is registered before anything that can throw. ## What this does not cover - The scan-input gate in `buildNativeContext` still matches only `CometNativeScanExec`, so an Iceberg (or CSV, or contrib) scan fused into a parent native block registers no report at all. That is the gap #5265 closes by widening the match to every `CometLeafExec`; this PR leaves that line untouched so the two do not collide. - Several native plan instances in one task, as `CometCoalesceExec` produces, set absolute values into the same accumulators and the last one wins, in the SQL metrics themselves as well as the input column. That needs a native-side change and is tracked in #5879. ## How are these changes tested? New tests in `CometTaskMetricsSuite`, all asserting against vanilla Spark's numbers or against the plan shape: - Broadcast join plus `LIMIT 3` over a native scan, at update interval -1 and the default. Reported 0 bytes before this change. - Native scan block plus `LIMIT 3` at both intervals. - A coalesced `UNION ALL` of a native scan and a fallback JSON scan in one task. Reported 5000 of 10000 rows before this change. - A coalesced union of a native scan and a cached input read from the block manager. - A coalesced union of two native scans, each block reporting its own bytes. Reports 5000 of 10000 with `set` semantics. - A failed attempt (ANSI division by zero after several batches) still reports the bytes and rows scanned. - Unit tests on `TaskContext.empty()`: overlapping trees count each accumulator once and keep counters a Spark scan already set, the ordering contract, and scans that never ran report nothing. Each guard was checked by reverting one part of the fix at a time: the ordering, the incrementing setters, and the once-per-task claim each make a distinct test fail. The full suite passes on Spark 3.5 and 4.0, along with the Iceberg native scan input metrics test. -- 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]
