Comet shuffle does not report Spark 4.1's order-independent shuffle checksum #6414

Describe the bug

Spark 4.1 added spark.sql.shuffle.orderIndependentChecksum.enabled and spark.sql.shuffle.orderIndependentChecksum.enableFullRetryOnMismatch, both off by default. Spark 4.2 adds enableQueryLevelRollbackOnMismatch alongside them. With them on, ShuffleExchangeExec gives the ShuffleDependency a set of RowBasedChecksums, Spark's shuffle writers fold every record into an order-independent checksum, and the value reaches the driver in MapStatus.checksumValue. MapOutputTracker.addMapOutput compares it with the previous attempt of the same map task. A retried task that produced different output (an indeterminate stage) is detected, and Spark retries the consumer stages instead of mixing old and new output.

Comet's shuffle writers never compute that checksum. CometNativeShuffleWriter, CometUnsafeShuffleWriter and CometBypassMergeSortShuffleWriter all build their MapStatus without one; MapStatusHelper relies on the parameter's default of 0. So every Comet map output reports checksumValue == 0, a mismatch can never be detected for a Comet shuffle, and the protection is silently off when a user enables it. Comet doesn't check these configs, so it doesn't fall back to Spark's shuffle either.

This came up in #6406. Once Comet actually runs in Spark's MapStatusEndToEndSuite, "Propagate checksum from executor to driver" fails with mapStatuses.forall(_.checksumValue != 0) was false.

Steps to reproduce

On Spark 4.1 with Comet and CometShuffleManager, from a test in an org.apache.spark package (as MapStatusEndToEndSuite does, since MapOutputTrackerMaster is private[spark]):

spark.conf.set("spark.sql.shuffle.orderIndependentChecksum.enabled", "true")
spark.range(1000).repartition(10).write.mode("overwrite").saveAsTable("t")
val tracker = spark.sparkContext.env.mapOutputTracker.asInstanceOf[MapOutputTrackerMaster]
tracker.shuffleStatuses(0).mapStatuses.map(_.checksumValue) // all 0 with Comet

Expected behavior

When spark.sql.shuffle.orderIndependentChecksum.enabled is true, Comet should either report an equivalent order-independent checksum from its shuffle writers, or fall back to Spark's shuffle for the exchange so that the protection stays in place.

Additional context

The configs are off by default, so only users who opt in are affected. The #6406 fix marks the MapStatusEndToEndSuite test IgnoreComet in the Spark 4.1.3 and 4.2.0 diffs, pointing at this issue.