andygrove opened a new issue, #6935: URL: https://github.com/apache/datafusion-comet/issues/6935
### What is the problem the feature request solves? Spark measures a task's CPU as the CPU time of the task thread. It reads `ThreadMXBean.getCurrentThreadCpuTime()` before and after `task.run` and stores the difference as the `executorCpuTime` task metric ([`Executor.scala`](https://github.com/apache/spark/blob/90568fb163775793260ecb13f2019ea10ef4e2f3/core/src/main/scala/org/apache/spark/executor/Executor.scala#L979-L980)), which is `Executor CPU Time` in the event log and `executorCpuTime` in the REST API. Since #3553 (0.14.0), Comet runs a native plan with no JVM input on its Tokio runtime, and the task thread waits for the plan's batches ([`BatchProducer`](https://github.com/apache/datafusion-comet/blob/474b5ece841a7fc6dff37ae04bd6787e654d7be7/native/core/src/execution/jni_api.rs#L1270-L1301), chosen [here](https://github.com/apache/datafusion-comet/blob/474b5ece841a7fc6dff37ae04bd6787e654d7be7/native/core/src/execution/jni_api.rs#L1493-L1497)). That covers the usual scan stage: a native Parquet or Iceberg scan, through filters, projections and a partial aggregate, into the native shuffle writer or a native write. Its work runs on the runtime's worker threads, which Spark's metric doesn't see, so the stage reports almost no CPU. A plan with a JVM input polls its stream on the task thread with `block_on` ([here](https://github.com/apache/datafusion-comet/blob/474b5ece841a7fc6dff37ae04bd6787e654d7be7/native/core/src/execution/jni_api.rs#L1522-L1530)), so its CPU is counted, apart fr om work in tasks it spawns. We hit this in a production job: an Iceberg scan, filter, sample and partial aggregate into native shuffle, over 245K tasks. Comet's stage recorded 20.8 minutes of executor CPU time in 109.4 hours of executor run time. Spark's run of the same stage recorded 7.98 hours of CPU in 150.2 hours, and a comparison built from these metrics concluded that Comet used 23x less CPU than Spark. Anything that takes CPU from Spark's metrics gets the same wrong answer: benchmark comparisons, CPU-efficiency reports from event logs, chargeback. The [metrics guide](https://github.com/apache/datafusion-comet/blob/main/docs/source/user-guide/latest/metrics.md) doesn't mention it. Comet's per-operator `elapsed_compute` metrics don't fill the gap either: they're wall time from `Instant`, not CPU time. Comet can't correct `executorCpuTime` itself. Spark sets it after `task.run` returns, and task completion listeners run inside `task.run` ([`Task.scala`](https://github.com/apache/spark/blob/90568fb163775793260ecb13f2019ea10ef4e2f3/core/src/main/scala/org/apache/spark/scheduler/Task.scala#L147)), so anything Comet added would be overwritten. To see it, in a `spark-shell` with Comet and native shuffle enabled: ```scala import java.util.concurrent.atomic.AtomicLong import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd} import org.apache.spark.sql.functions.{max, sum} val cpuNs = new AtomicLong val runMs = new AtomicLong spark.sparkContext.addSparkListener(new SparkListener { override def onTaskEnd(e: SparkListenerTaskEnd): Unit = Option(e.taskMetrics).foreach { m => cpuNs.addAndGet(m.executorCpuTime) runMs.addAndGet(m.executorRunTime) } }) spark.range(0, 200000000L) .selectExpr("id % 1000 AS k", "id AS v", "CAST(id AS STRING) AS s") .write.mode("overwrite").parquet("/tmp/comet_cpu") def measure(comet: Boolean): Unit = { spark.conf.set("spark.comet.enabled", comet) cpuNs.set(0) runMs.set(0) spark.read.parquet("/tmp/comet_cpu").groupBy("k").agg(sum("v"), max("s")).collect() Thread.sleep(2000) // let the listener bus deliver the last task-end events println(f"comet=$comet cpu=${cpuNs.get / 1e9}%.1f s run=${runMs.get / 1e3}%.1f s") } measure(false) measure(true) ``` With Comet, the CPU total should be a small fraction of the run time. On Spark it should be most of it. ### Describe the potential solution Report native CPU in a metric of its own: 1. **Per task.** In `BatchProducer`, read the thread's CPU clock (`clock_gettime(CLOCK_THREAD_CPUTIME_ID)`) before and after each poll of the plan's stream ([here](https://github.com/apache/datafusion-comet/blob/474b5ece841a7fc6dff37ae04bd6787e654d7be7/native/core/src/execution/jni_api.rs#L1301)). A poll runs on one thread from start to finish, so the difference is the CPU that poll used, whichever worker ran it. Sum the differences into a nanosecond SQL metric, for example `native cpu time`. SQL metric updates are in each task's end event, so event-log tools can add it to `executorCpuTime`. The same timer on the `block_on` path would measure CPU that `executorCpuTime` already counts, so it needs a separate metric there, or none. 2. **Per executor.** Work that a plan hands to other Tokio tasks doesn't run inside those polls: iceberg-rust's delete-file loading and delete filters, HTTP connection tasks (including TLS for object stores), and `spawn_blocking` calls. To catch it, the runtime's `on_thread_start` hook ([here](https://github.com/apache/datafusion-comet/blob/474b5ece841a7fc6dff37ae04bd6787e654d7be7/native/core/src/execution/jni_api.rs#L383)) could record each worker and blocking thread's CPU clock (`pthread_getcpuclockid`), and the executor could report their sum as a gauge in `CometExecutorPlugin`'s metric registry and in the `CometExecutorMemoryUsage` events that #6401 writes to the event log. It can't be split by stage or task, but it misses nothing. 3. **Docs.** Say in the metrics guide that `executorCpuTime` leaves out native execution for these stages. Until a metric exists, CPU can be measured as the pod's or container's CPU, or with the executors' `jvmCpuTime` gauge from Spark's `JVMCPUSource`, which counts every thread in the process. Questions to settle: - The overhead of two clock reads per poll. On Linux, `CLOCK_THREAD_CPUTIME_ID` is a system call rather than a vDSO read, and a stream waiting on I/O can be polled many times per batch. - Where the per-task metric lives: on the native block's root operator, or on the native shuffle writer and native write exec nodes. ### Additional context Before #3553, the task thread polled these plans itself, so their CPU was in `executorCpuTime`. -- 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]
