comphead opened a new issue, #7582:
URL: https://github.com/apache/datafusion-comet/issues/7582
### Describe the bug
A map task whose plan is a native scan feeding a native shuffle write
(`CometNativeScan` -> `CometExchange`, map RDD `CometNativeShuffleInputRDD`)
reports no metrics while it runs. Spark's live UI shows blank Input and Shuffle
Write columns, and the plan's SQL metrics stay at 0 until the task ends. In one
job, 50 such tasks each read about 4 GiB of Parquet and showed nothing for over
30 minutes.
Two things combine:
1. For a plan without JVM input, `execute_plan` publishes metrics only after
its `BatchProducer` returns a batch. `ShuffleWriterExec` returns none until it
has written all of its input, so `spark.comet.metrics.updateInterval` never
fires, and the first publish is the final one in `releasePlan`. A 100 ms
interval changes nothing.
2. `CometMetricNode.reportScanInputMetrics` adds the scan's bytes and rows
to the task's input metrics only in a task completion listener, so fresh SQL
metrics would still not reach the Stages tab.
### Steps to reproduce
Run on `local-cluster`. In `local[N]`, Spark's live status store never shows
in-flight task metrics, for any engine. Set
`spark.comet.shuffle.native.partitioning.roundrobin.enabled=true` and
`spark.sql.files.maxPartitionBytes=64g`, then:
```scala
spark.range(0, 30000000, 1, 2)
.selectExpr("id", "id * 1.5 AS d", "md5(CAST(id AS STRING)) AS s1",
"sha1(CAST(id AS STRING)) AS s2", "concat('user_', CAST(id % 100000 AS
STRING)) AS s3")
.write.parquet(path)
spark.read.parquet(path).repartition(50).write.format("noop").mode("overwrite").save()
```
While the map stage runs, sample `sparkContext.statusStore.activeStages()`,
which is what the Stages tab renders. With Spark, Input grew to 30M records and
Shuffle Write to 2.6 GiB during the run. With Comet, both stayed blank in every
sample, and the running tasks' heartbeats carried no non-zero accumulator at
all. Comet's values (2.4 GiB and 30M records read, 2.7 GiB written) appeared
only at `TaskEnd`.
### Expected behavior
While the task runs, its SQL metrics update every
`spark.comet.metrics.updateInterval` and its Input grows, as with Spark.
Shuffle Write bytes are known only once the data file is complete, so they can
stay at task end, like Spark's sort-based shuffle writer.
### Additional context
#6313 made plans without JVM input publish on the interval when a batch
arrives, and noted that a block ending in a native shuffle write was unaffected
because it hands back at most one batch. Waiting for that batch is the problem
here.
--
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]
