andygrove opened a new pull request, #6894: URL: https://github.com/apache/datafusion-comet/pull/6894
## Which issue does this PR close? Closes #5336 and #6313 on `branch-1.1`. #5880 and #6314 already closed them on `main`. ## Rationale for this change This backports #5880 and #6314 to `branch-1.1`, as two commits in the order they merged. A native block whose leaves are all native scans publishes its whole metric tree to the JVM after every output batch, ignoring `spark.comet.metrics.updateInterval` (#6313). Each publish walks the plan, aggregates every metric, encodes a protobuf and calls into the JVM. #6313 measured 19 to 38 µs per publish, or 40 to 78 ms for a task that returns 2,048 batches. It affects pure-native blocks whose output reaches the JVM batch by batch: a scan feeding a Spark write or a collect, a broadcast build side, JVM shuffle or a fallback operator. It has been this way since 0.14.0, so it is not a 1.1.0 regression, but the config doesn't do what its documentation says. #6314 can only come to this branch together with #5880. Here, `reportScanInputMetrics`, which feeds the Input column on the Stages and Executors tabs, registers its completion listener after `super.compute` creates the `CometExecIterator`. Spark runs completion listeners in reverse order, so the report runs before the iterator's close publishes the final metrics (#5336). A block with a JVM input already reports 0 input bytes when a limit stops it early. Pure-native blocks escape that only because they publish after every batch, so taking #6314 alone would make them report 0 too. #5880 registers the listener first, so it runs after the final publish, which is the order #6314's description relies on. The tests below show both halves of that. Not proposed for `branch-1.0`. Both bugs are there too, but they cost CPU time and Spark UI accuracy rather than query results, and neither source PR has a `backport-1.0` label. ## What changes are included in this PR? Two cherry-picks with `-x`. See the source PRs for the details. - #5880 applies cleanly. It registers the scan input report before `super.compute` at its three sites (`CometNativeScanExec`, `CometIcebergNativeScanExec` and the `CometExecRDD` override in `operators.scala`), adds to the task's `bytesRead` and `recordsRead` instead of replacing them, and claims each accumulator once per task through the registry that `reportSpillMetrics` already uses. - #6314 makes the channel branch of `executePlan`, which serves native blocks with no JVM input, call `update_metrics_on_interval` instead of `update_metrics`. `releasePlan` still publishes the final values, as it does on `main`. One adaptation, recorded in #6314's commit message: this branch still feeds a native block with no JVM input through the `batch_receiver` channel, which `main` has since replaced with `batch_producer`, so the same one-line change lands in the channel branch. Apart from that hunk's context, both diffs are identical to `main`, tests included. `reportSpillMetrics`, the other listener #6314 relies on, is already registered first in `CometExecRDD.compute` on this branch. ## How are these changes tested? Run locally on this branch with JDK 17: - On the default Spark 4.1 profile, `CometTaskMetricsSuite` (25 tests), `CometExecIteratorLifecycleSuite` (10), the "Comet native metrics" tests in `CometExecSuite` (5) and the metrics tests in `CometIcebergNativeSuite` (9) pass. That includes every test the two PRs add. Scalastyle and spotless ran in the same build. - On Spark 3.5 and Scala 2.12 the sources and tests compile, and scalafix passes in CHECK mode. - The native workspace builds, and `cargo fmt --check` and workspace clippy with `--all-targets -D warnings` are clean. My local toolchain is Rust 1.97, and CI uses stable. To check that the tests catch both bugs on this branch, I ran `CometTaskMetricsSuite` and the Iceberg limit test with parts of the change taken out: - Without #6314's change, using a native library rebuilt with `update_metrics` back in the channel branch, only #6314's test fails, with `output_rows was published mid-stream: 4000`. So this branch has #6313, and #5880's tests pass on their own. - With #6314 but with #5880's three listener registrations moved back after `super.compute`, the limit tests over a native scan block, over an Iceberg scan and over a broadcast join fail with `bytesRead should be > 0 at interval -1, got 0`, and the failed-attempt test fails with `List((0,0))`. That is what #6314 alone would do here. - With the registrations moved back and without #6314, only the broadcast join limit test (#5336 itself) and #6314's test fail. The native scan block and Iceberg limit tests pass, because publishing after every batch hides the listener order. By CI's path routing, which I checked with `dev/ci/compute-changes.py`, this PR runs the all-profiles Linux build, macOS, the Spark SQL jobs for 3.5, 4.0 and 4.1, and the four Iceberg versions, so no `run-*` label is needed. -- 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]
