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
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 bytesMeasured on main 3810936 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.
Describe the bug
CometMetricNode.setwrites the absolute per-plan value that nativeupdate_metricspublishes into the SQL metric accumulator (metric.set(v)). When one task runs several native plan instances that share the sameCometMetricNodetree, each instance overwrites the previous one, and Spark ships whichever value was written last.CometCoalesceExecproduces exactly that shape:rdd.coalesce(n, shuffle = false)computes several parent partitions of the sameCometExecRDDinside one task, and each partition builds its ownCometExecIteratorover the shared metric tree.Steps to reproduce
Measured on main 3810936 with Spark 3.5:
output_rows=2500,bytes_scanned=21696for 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
Inputcolumn 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.reportSpillMetricsdocuments 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.