fix: report native shuffle read metrics

[peterxcli]

Which issue does this PR close?

Part of #3996.

Rationale for this change

The direct native shuffle-read path bypasses Spark's normal decode path, so its records-read metric was not updated and temporary shuffle-read metrics were not merged into task metrics. This left the Spark Stage and SQL UIs with missing or zero native shuffle-read values.

What changes are included in this PR?

• Pass the existing SQL shuffle-read metrics callback through CometShuffleBlockIterator.
• Update Spark's records-read metric after each successfully decoded native shuffle batch.
• Merge temporary shuffle-read metrics when a direct native shuffle task completes.
• Extend task-metrics regressions to require the direct native ShuffleScan path, assert SQL/Stages UI aggregates, and preserve metrics on failed attempts.

How are these changes tested?

• cargo build
• cargo fmt --all -- --check
• ./mvnw test -Dtest=none -Dsuites='org.apache.spark.sql.comet.CometTaskMetricsSuite native shuffle read' -Dscalastyle.skip=true (2 tests)
• verified with a built Comet JAR on a fresh Spark 4.1.2 standalone cluster with two workers:
  • executor logs confirmed ShuffleScanExec
  • the Stage UI showed Shuffle Read Size / Records: 2004.6 KiB / 500000
  • the SQL UI showed 500,000 records, 2004.6 KiB read, and 32 local blocks

before: (shuffle read didn't appear)

after:

[comphead]

I'm on it today, thanks @peterxcli

[comphead]

Thanks @peterxcli would you mind to attach a screenshot on what is changed? SparkUI shows currently records from shuffle read stages, just wondering what part of metrics is addressed

[peterxcli]

@comphead updated the screenshot to show the parquet scan result in ui against before/after patch, please take another look, thanks!

[comphead]

Thanks @peterxcli for the test, I believe it might have sense to compare against Spark metrics. The same way is done for the input scan metrics.

[peterxcli]

@comphead
pure spark:

looks like they have different shuffle read byte but I think it's because Spark uses row/serializer shuffle data; Comet uses compressed Arrow IPC, so identical bytes are neither expected nor desirable. Spark itself derives this counter from each fetched buffer’s physical size (Spark source)

[comphead]

rows should be the same so we can check it with Spark metrics, bytes can be checked it is the same ballpark. Please check input scan metrics test, the approach was adopted there

[peterxcli]

@comphead thanks for the direction, added test to check # of rows

[comphead]

Thanks @peterxcli
CI pending