apache / apache/datafusion-comet
Native metrics from several plan instances in one task overwrite each other, so a coalesced scan reports only its last partition
- Dominant language
- Scala
- Stars
- 1.3k
- Forks
- 373
- Avg merge
- 2d 4h
- Merged PRs (30d)
- 198
Description
### Describe the bug
`CometMetricNode.set` writes the absolute per-plan value that native `update_metrics` publishes into the SQL metric accumulator (`metric.set(v)`). When one task runs several native plan instances that share the same `CometMetricNode` tree, each instance overwrites the previous one, and Spark ships whichever value was written last.
`CometCoalesceExec` produces exactly that shape: `rdd.coalesce(n, shuffle = false)` computes several parent partitions of the same `CometExecRDD` inside one task, and each partition builds its own `CometExecIterator` over the shared metric tree.
### Steps to reproduce
```scala
spark.createDataFrame((0 until 10000).map(i => (i, s"e_$i"))).repartition(4).write.parquet(dir)
spark.read.parquet(dir).createOrReplaceTempView("t")
val df = sql("SELECT /*+ COALESCE(1) */ * FROM t")
df.collect()
val scan = find(stripAQEPlan(df.queryExecution.executedPlan))(_.isInstanceOf[CometNativeScanExec]).get
scan.metrics("output_rows").value // 2500, the last partition only
scan.metrics("bytes_scanned").value // one partition's bytes
```
Measured on main 3810936b4 with Spark 3.5: `output_rows=2500`, `bytes_scanned=21696` for a 10000-row, four-partition table coalesced into one task.
### Expected behavior
The SQL metrics on the scan (and every operator below the coalesce) cover all partitions the task read, 10000 rows here.
### Additional context
Task-level input metrics inherit the same value, so the `Input` column under-reports too. #5336 fixes the input metrics listener ordering and stops it replacing Spark's own counters, but it cannot recover the overwritten per-partition values; that needs the native side to accumulate across plan instances (or a per-instance snapshot on the JVM side) instead of setting absolute values.
`reportSpillMetrics` documents the same shape ("a coalesced partition registers the same tree once per parent partition") and claims each accumulator once per task, so spill metrics under coalesce are subject to the same last-writer-wins value.
Contributor guide
Research direction
Start with CometMetricNode.set and trace how native update_metrics values reach the shared SQL metric tree through CometExecIterator. Then inspect CometCoalesceExec and reportSpillMetrics to understand repeated registration for parent partitions. Done means coalesced scans and operators report totals across all partitions, including input and spill metrics, rather than the last partition only.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust, scala
- Domain
- backend, performance
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Active
- Clarity
- Mostly clear
- Newbie friendliness
- 55/100