fix: report native child spill metrics in shuffle tasks

[sunchao]

Why are the changes needed?

Closes #5382.

When a Spark task spills data, the task-level spill metrics are the natural place to understand how much memory pressure that task experienced. They populate the Spark UI's stage and task views and are commonly used when investigating slow shuffles, tuning memory, or comparing execution engines. For a native Comet shuffle, however, those totals currently describe only part of the work that actually happened.

Consider a query that sorts rows within each input partition before repartitioning them:

spark.table("orders")
  .sortWithinPartitions($"amount".desc)
  .repartition(4, $"customer_id")
  .count()

Comet can execute the sort and shuffle writer together within the same native shuffle task:

CometShuffleExchangeExec
  CometSortExec
    Scan

Both the shuffle writer and the sort can spill independently. Their individual spill values are already available on their corresponding Spark SQL operators, but the existing task-metric bridge reads only the shuffle writer's metrics. Consequently, the SQL operator view and the task/stage view can tell conflicting stories about the same execution.

Suppose the shuffle writer spills 32 MiB to disk and the child sort spills another 128 MiB. Suppose also that the writer reports 96 MiB of in-memory spill, while the child sort exposes no in-memory spill counter:

Spill metric	Shuffle writer	Child sort	Task total before	Task total after
Disk bytes spilled	32 MiB	128 MiB	32 MiB	160 MiB
Memory bytes spilled	96 MiB	Not reported	96 MiB	96 MiB

Before this change, the task appears to have spilled only 32 MiB to disk even though its own operators report five times that amount. This is not a difference in how much data the query actually spilled: it is missing accounting at the boundary between Comet's operator metrics and Spark's task metrics.

The unchanged memory total in this example is equally important. Disk spill bytes can represent compressed data written to storage, while memory spill bytes represent the corresponding in-memory volume. Treating the child's 128 MiB of disk spill as though it were also 128 MiB of memory spill would merely replace one misleading metric with another. A child should contribute to task-level memory spill only when it actually exposes a trustworthy memory-spill measurement.

What changes were proposed in this PR?

The change makes task-level spill reporting represent the complete native shuffle task rather than only its outer shuffle writer.

Comet already maintains an operator-shaped tree of SQL metrics for the native plan. Instead of introducing a second accounting system, this PR connects the shuffle writer's existing metrics to that existing child-operator tree and reports the combined tree when the task completes. Disk spill is aggregated from the disk counters that operators actually expose; memory spill is aggregated independently from genuine memory counters. The traversal also recognizes shared metric objects, so a reused accumulator contributes once even if it appears through multiple paths in the tree.

Returning to the example above, the completed task now reports 32 MiB + 128 MiB = 160 MiB of disk spill. Its memory-spill total remains 96 MiB because the child sort has no memory counter to contribute. If a different child operator exposes, for example, 384 MiB of memory spill, that value is included naturally and the task reports 96 MiB + 384 MiB = 480 MiB instead. No memory value is inferred from compressed disk bytes.

The other subtlety is timing. Native operators can publish their final metrics only while their iterators are being closed at task completion, and a shuffle task can initialize nested input producers before its writer is constructed. Registering the reporting callback too late can therefore observe incomplete child metrics, particularly when an attempt fails. This PR establishes one task-level reporting callback before those producers are initialized, so Spark's completion-listener ordering first finalizes the native operators and then reads the complete metric tree. That single callback replaces the previous writer-only callback rather than layering additional reporting on top of it.

The result is intentionally narrow: it changes how already-existing metrics are collected into Spark's task totals, without changing query execution, spill behavior, native memory management, the planner, or JNI.

How was this PR tested?

A Spark integration regression runs a real native sort beneath a native shuffle, applies enough memory pressure to make the sort spill, and then compares the Spark stage metrics with the SQL metrics of the actual executed operators. It verifies that task-level disk spill equals the writer's disk spill plus the child sort's disk spill exactly once. It also verifies that the child sort has no memory-spill counter and that the task's memory-spill total therefore remains equal to the writer's real memory metric.

A separate deterministic task-completion test models a nested input producer and a shuffle writer publishing their metrics at completion time. It verifies that both disk and memory values are collected after those final updates, for both successful and failed task attempts. Additional coverage constructs a nested metric tree with shared accumulators to ensure reused metrics are not counted multiple times. Existing shuffle spill, failed-attempt, scan-input, output, and task-binary-size tests continue to run alongside the new regressions.

Both focused suites passed on Spark 3.5, Spark 4.0, and Spark 4.1, with 12 passing tests per version.

Focused validation commands

SUITES=org.apache.spark.sql.comet.CometTaskMetricsSuite,org.apache.spark.sql.comet.execution.shuffle.CometNativeShuffleInputRDDSuite

./mvnw -Pspark-3.5 test -Dtest=none -DfailIfNoTests=false -Dsuites="$SUITES"
./mvnw -Pspark-4.0 test -Dtest=none -DfailIfNoTests=false -Dsuites="$SUITES"
./mvnw -Pspark-4.1 test -Dtest=none -DfailIfNoTests=false -Dsuites="$SUITES"

[peterxcli]

other lgtm

[peterxcli]

@sunchao #5447 I filed a related issue, please feel free to take it, but if you dont have bandwidth to do it then I'm happy to help :)

[comphead]

Thanks @sunchao

[sunchao]

Merged, thanks @comphead and @peterxcli for the review!