andygrove opened a new issue, #5336: URL: https://github.com/apache/datafusion-comet/issues/5336
## Describe the bug `CometMetricNode.reportScanInputMetrics` (`spark/src/main/scala/org/apache/spark/sql/comet/CometMetricNode.scala:69`) populates Spark's task-level `inputMetrics`, which drive the `Input` column on the UI's Stages and Executors tabs. It is unreliable when a native block contains a native scan *alongside* a JVM input, for example an Iceberg or Parquet scan joined against a shuffled side, or against a fallback Spark scan reaching the block through `CometSparkToColumnarExec`. There are two independent problems. Both are long-standing. They were found while reviewing #5265, which does not introduce either one, but which does widen the set of plans that reach this reporting path from Parquet-only to any `CometLeafExec`. ## Problem 1: the completion listener is registered too late Spark's `TaskContextImpl` keeps task completion listeners in a `Stack` and invokes them in reverse registration order: > Using a stack causes us to process listeners in reverse order of registration. As listeners are invoked, they are popped from the stack. `CometNativeExec.executeColumnarWithContext` (`spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:626-629`) calls `reportScanInputMetrics` *after* `super.compute(split, context)`. That is where the `CometExecIterator` is constructed and registers its own close listener. So the reporting listener runs *before* the iterator's `close()`, and `close()` is where `nativeLib.releasePlan` performs the final `update_metrics` (`native/core/src/execution/jni_api.rs:965`). The same ordering appears at `CometNativeScanExec.scala:282` and `CometIcebergNativeScanExec.scala:239`. Whether this actually loses data depends on which native execution path the block takes: - **No JVM data sources.** `jni_api.rs:822` takes the `batch_receiver` branch, which calls `update_metrics` on every batch (`jni_api.rs:872`). The metric is current when the listener runs, so reporting is correct. - **With a JVM data source.** Execution takes the busy-poll branch, where `update_metrics` only fires when `spark.comet.metrics.updateInterval` has elapsed (`jni_api.rs:903-905`). The default is 3000ms, so a task finishing inside that window reads a stale or zero `bytes_scanned`. `CometConf.COMET_METRICS_UPDATE_INTERVAL` documents that a negative interval means metrics are updated only on task completion, which would report zero every time. The codebase already documents the correct ordering elsewhere. `CometNativeShuffleWriter.scala:130` registers before constructing its iterator, and `CometNativeWriteExec.scala:223` does the same with an explicit comment. The scaladoc on `CometMetricNode.reportNativeWriteOutputMetrics` states the requirement outright. The scan-input path is the one site that does not follow it. ## Problem 2: `setBytesRead` overwrites rather than accumulates `reportScanInputMetrics` uses `setBytesRead` / `setRecordsRead` (`CometMetricNode.scala:81-82`) rather than the incrementing variants. If a fallback Spark scan reaches the same native block through `CometSparkToColumnarExec`, Spark's `FileScanRDD` has already accumulated bytes for that side into the task's `inputMetrics`. Comet's completion listener then discards it. ## How the two compound Together they turn a missing number into a wrong one. For a block with an Iceberg scan and a fallback Spark scan, Spark's accumulated `bytesRead` is overwritten by a native value that Problem 1 can leave at zero, so the `Input` column goes from partially correct to empty. ## Suggested fix Move the `reportScanInputMetrics` call above `super.compute(...)` at the three sites listed above, so reverse-order invocation puts it after the iterator's `close()`. This matches `CometNativeShuffleWriter` and `CometNativeWriteExec`. Separately, consider whether the incrementing metric setters are more appropriate than `setBytesRead` / `setRecordsRead`. ## Test coverage gap Existing coverage exercises only the pure-native shape, where per-batch metric updates mask Problem 1. That includes the tests added in #5265 and `"native shuffle reports task input metrics for its scan child"` in `CometTaskMetricsSuite`. A test for a native block combining a native scan with a JVM input would cover both problems. ## Additional context Found while reviewing #5265, which fixes a related but distinct bug where `hasScanInput` matched only `CometNativeScanExec` and so skipped Iceberg, CSV and contrib scans entirely. -- 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]
