andygrove opened a new issue, #6414:
URL: https://github.com/apache/datafusion-comet/issues/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 `RowBasedChecksum`s, 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]`):
```scala
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.
--
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]
