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

Open
#5,879 0 comments 0 reactions 0 assignees View on GitHub
requires-triage
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.