fix: report native shuffle write metrics accurately

[sunchao]

Why are the changes needed?

When a shuffle becomes too large to keep in memory, Spark spills intermediate data to local disk. Operators use the Spark UI to answer two different questions about that work: how much memory was occupied by the data that spilled, and how much data was actually written to disk. Spark intentionally exposes these as separate task metrics because the on-disk representation is compressed and is often much smaller than the in-memory representation.

Comet's native shuffle writer did not preserve that distinction. It copied the same on-disk byte count into both memoryBytesSpilled and diskBytesSpilled. Worse, its disk counter considered only bytes returned by immediate writes, even though Arrow's batch coalescer can hold those bytes until flush(). In that common case, a real spill could appear as zero memory spilled and zero disk spilled in the Spark UI.

For example, suppose a task spills 256 MiB of in-memory shuffle data and partition indices, which compress to 32 MiB on disk. These numbers are illustrative:

Situation	What actually happened	Spark UI before	Spark UI after
Compressed bytes are emitted during the write	256 MiB in memory, 32 MiB on disk	Memory: 32 MiB; disk: 32 MiB	Memory: 256 MiB; disk: 32 MiB
Compressed bytes are emitted only when the writer flushes	256 MiB in memory, 32 MiB on disk	Memory: 0; disk: 0	Memory: 256 MiB; disk: 32 MiB
The task spills successfully, then fails before the shuffle commit	256 MiB in memory, 32 MiB on disk	Failed-task memory: 0; disk: 0	Failed-task memory: 256 MiB; disk: 32 MiB

The failure case matters especially for debugging: a task that runs out of disk, encounters bad input, or fails after several spills should not look as though it never spilled at all.

Accurate in-memory accounting also has an Arrow-specific complication. One input batch can be split into many zero-copy slices that all reference the same underlying allocation. The regression test uses 16,384 Int64 values, backed by one 128 KiB allocation, and processes them as 16 slices of 1,024 rows. If every slice triggers a spill, simply summing each released memory reservation counts that same 128 KiB allocation 16 times: 2 MiB instead of 128 KiB, before even including the real per-slice partition-index allocations. The underlying allocation remains owned by the original input batch throughout the operation, so it must be counted once for that batch, not once per slice.

Finally, native shuffle already measures the time spent interleaving rows into output partitions, but that work was not visible in the SQL exchange metrics. This made it harder to distinguish repartitioning overhead from encoding and compression overhead when investigating slow shuffle stages.

Part of #3996. This PR addresses native shuffle write observability only; shuffle reads and mixed native/JVM scan-input accounting remain outside its scope.

What changes were proposed in this PR?

The change gives native shuffle spill accounting two independent sources of truth and carries them consistently into both of Spark's observability surfaces.

For disk usage, the source of truth is the physical spill file itself. The native writer observes the file position before a spill and again after buffered Arrow data has been flushed. Their difference is the actual compressed number of bytes written, including data produced only during the final flush and data appended to an existing spill file.

For memory usage, the source of truth is the in-memory shuffle state being spilled: Arrow backing allocations plus the partition-index structures that organize its rows. The native writer records the memory released by the spill and also includes the current batch when the memory pool rejected its reservation after that batch had already been buffered. To avoid turning Arrow's zero-copy slicing into fictional memory growth, it remembers which backing allocations have already been counted for the lifetime of the original input batch. Separate input batches still contribute cumulatively, as Spark's spill metrics require.

Those two measurements are exposed separately in the SQL shuffle exchange and copied into their corresponding Spark task metrics only after the native execution plan has published its final values. The copy runs from a task-completion listener registered before the native iterator; Spark invokes completion listeners in reverse registration order, so native cleanup and final metric publication happen first. Because task-completion listeners also run when an attempt fails or is canceled, the same accounting remains available for the attempts that are most useful to diagnose.

The exchange also surfaces the native partition-interleaving timer, making the main stages of native shuffle work visible alongside the existing repartitioning, encoding, and spill metrics. Existing shuffle output semantics, compression behavior, and scan-input accounting are unchanged.

How was this PR tested?

Focused Rust tests exercise both spill triggers: an explicit maximum buffer size and a memory-pool reservation failure. They compare reported disk spill bytes against the actual spill-file sizes, verify zero spill metrics when no spill occurs, and cover the shared-allocation example above. They also verify that separate outer input batches are counted cumulatively even when they reference the same Arrow backing buffer.

The Spark integration tests validate the complete native-to-Spark reporting path on both Spark 3.5 and Spark 4.0:

• A successful shuffle writes 20,000 rows across four partitions with Zstandard compression and forced spills. The test verifies shuffle records, output bytes, write time, partition-interleaving time, separate memory/disk SQL metrics, and exact agreement between those SQL metrics and Spark's recorded stage/task metrics.
• A failing shuffle processes 8,192 valid rows before a later row triggers an ANSI divide-by-zero inside a native projection. Earlier batches have already spilled, and the test verifies that the failed ShuffleMapTask still reports both memory and disk spill bytes.

cd native
cargo fmt --all -- --check
cargo clippy -p datafusion-comet-shuffle --lib -- -D warnings
cargo test -p datafusion-comet-shuffle --lib spill -- --nocapture
cargo test -p datafusion-comet-shuffle --lib shared_backing_once_per_input_batch -- --nocapture

cd ..
make core
./mvnw -Pspark-3.5 test -Dtest=none \
  '-Dsuites=org.apache.spark.sql.comet.CometTaskMetricsSuite memory and disk spill metrics'
./mvnw -Pspark-4.0 test -Dtest=none \
  '-Dsuites=org.apache.spark.sql.comet.CometTaskMetricsSuite memory and disk spill metrics'