[SPARK-58830][CORE] Add MapStatus checksum to detect metadata corruption between mapper and reducer

[wang-haihua]

What changes were proposed in this pull request?

This PR adds an integrity checksum to MapStatus that covers the set of non-empty
partition indices, and verifies it wherever a MapStatus is consumed to plan shuffle
block fetches.

• A new MapStatusChecksum utility computes a checksum (CRC32 or Adler32) over the
  indices of non-empty partitions in a partitionLengths array. It returns None when
  there are no non-empty partitions.
• MapStatus (both CompressedMapStatus and HighlyCompressedMapStatus) gains a new
  nonEmptyChecksum: Option[Int] field, computed at construction time when enabled, and
  carried through Java (Externalizable) serialization via a presence flag followed by
  the value, matching the existing pattern used for other optional fields.
• MapOutputTracker.verifyChecksumOrFail recomputes the checksum from the MapStatus
  actually consumed by a reducer and compares it against the stored value. A mismatch
  throws a FetchFailedException, reusing Spark's existing fetch-failure / stage-retry
  recovery path instead of introducing a new failure mode.
• Two new internal, off-by-default configs:
  • spark.shuffle.mapStatus.checksum.enabled (default false)
  • spark.shuffle.mapStatus.checksum.algorithm (default CRC32, also supports ADLER32)

The checksum is computed over the non-empty/empty boundary of each partition rather
than the sizes themselves, because sizes are already lossily compressed/estimated by
CompressedMapStatus / HighlyCompressedMapStatus and can legitimately differ in
representation while still being correct. Only the non-empty/empty boundary matters for
the silent-data-loss failure mode this PR targets.

Why are the changes needed?

MapStatus is the most frequently accessed piece of shuffle metadata: every mapper
produces one, the driver holds one per map task, and every reducer consults it to decide
which blocks to fetch. If a MapStatus's partitionLengths (or its compressed
representation) is corrupted in memory or in transit between the mapper and a reducer --
e.g. due to a transient hardware fault, a JVM/memory corruption, or a serialization bug
-- a partition that actually has data on disk can appear to have size 0. Because block
fetchers are intentionally designed to skip zero-size blocks
(MapStatus.getSizeForBlock(reduceId) == 0), the reducer silently skips fetching that
block. The on-disk data is never read and is effectively lost, with no exception, no
failed stage, and no log signal anywhere in the pipeline.

We hit this in production: a job produced fewer output rows than an otherwise-equivalent
rerun, with no errors in the driver or executor logs. Comparing Spark UI SQL metrics for
the two runs showed shuffle "records written" and "records read" diverging by a large
margin at an Exchange node in the faulty run. Instrumenting MapStatus at several points
along its mapper -> driver -> reducer path showed a handful of non-empty partition
entries had been flipped to zero somewhere along that path, while the corresponding
shuffle data files on disk were complete and correct. The root cause traced back to a
hardware fault on one worker node that silently corrupted the MapStatus object held in
the driver's memory. Full writeup: SPARK-58830.

Existing safeguards don't catch this class of failure:

• SPARK-35276 added a checksum over shuffle data files, so disk/IO-level corruption of
  shuffle bytes is detectable when a reducer actually reads them. But when the corruption
  is in the metadata (a non-empty partition reported as empty), the reducer never
  attempts to read that block, so the data checksum path is never exercised.
• SPARK-40872 handles a related but distinct push-based-shuffle case (a merged chunk that
  is empty because it was merged by a bad node) via a fallback to fetching the original
  blocks. It doesn't cover non-empty-to-empty corruption of a mapper's own MapStatus
  outside push-based merging.
• SPARK-57491 handles two attempts of the same partition both completing a push under
  speculation/non-deterministic shuffle keys, where only one attempt's MapStatus
  should survive. It doesn't address transport/memory corruption of a single
  MapStatus.

None of the above detects the case where a MapStatus that was correct when the mapper
produced it gets corrupted before a reducer consumes it. This PR closes that gap.

Does this PR introduce any user-facing change?

No behavior change by default: both new configs are internal and spark.shuffle.mapStatus.checksum.enabled defaults to false.

When explicitly enabled, a MapStatus metadata checksum mismatch now surfaces as a FetchFailedException, which triggers Spark's standard stage-retry recovery, instead of silently under-reading shuffle data.

How was this patch tested?

• New unit tests in MapStatusChecksumSuite cover: compute/recompute returning None for all-empty inputs, invariance to size values (only non-empty indices matter), sensitivity to a single-bit change in the non-empty set, agreement with an independent CRC32 reference implementation, ADLER32 vs CRC32 producing different values, rejection of unsupported algorithm names, recompute agreeing with compute and detecting a changed non-empty set, and Java serialization round-trips of the new field on both CompressedMapStatus and HighlyCompressedMapStatus.
• New test in MapOutputTrackerSuite, "MapStatus checksum mismatch triggers FetchFailedException", exercises verifyChecksumOrFail against a real MapOutputTrackerMaster with a corrupted checksum and asserts a FetchFailedException is thrown.
• Ran both suites locally: build/sbt 'core/testOnly *MapStatusChecksumSuite' and build/sbt 'core/testOnly *MapOutputTrackerSuite' -- all tests pass.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Claude Sonnet 5)