[SPARK-58107][CORE] Detect and repair non-deterministic local checkpoints via RDD-block content checksums

[juliuszsompolski]

What changes were proposed in this pull request?

A locally-checkpointed RDD partition can be materialized by more than one successful task attempt - from speculation, or a zombie task of a superseded stage attempt during retries. When the computation is non-deterministic, the attempts produce different bytes for the same RDDBlockId(rddId, partIdx). The BlockManager keeps both copies (the master's blockLocations simply appends locations, with no version / attempt id) and a reader picks one via Random.shuffle, so two reads of "the same" local checkpoint can disagree. A generic "unpersist + recompute" recovery is impossible because local checkpointing cuts lineage.

This adds a dedup-and-seal mechanism, enabled by default as a correctness fix. A non-deterministic computation has no single correct value, so it does not try to recompute one - it picks one surviving version per partition and makes every later read agree on it. Three parts:

1. Content checksum at store time. When an RDD marked for verification is materialized, the BlockManager folds a JDK checksum (CRC32C by default) over the serialized + compressed plaintext of each cache block (composed above the encrypting sink, so it is deterministic even under IO encryption) and reports it per replica to the driver alongside the block location. The mark is plumbed through getOrElseUpdateRDDBlock -> getOrElseUpdate -> doPutIterator and recorded on the block's BlockInfo.

2. Seal at the checkpoint commit point. LocalRDDCheckpointData.doCheckpoint, after materialization and before the checkpoint is exposed to readers, seals each partition on the BlockManagerMasterEndpoint: it picks one checksum value as authoritative, evicts the replicas that disagree (dropping them from the directory and asking their executors to drop the local copy, fire-and-forget), records the sealed checksum, and rejects future divergent (or checksum-less) registrations. No recompute loop and no synchronous wait.

3. Read-side self-check. A read of a block on the seal path serves the local copy only if its checksum equals the sealed value; otherwise it skips the local copy and goes to an authoritative remote location - making correctness independent of the asynchronous eviction landing.

The checksum is threaded across every path a block reaches the master (store, eviction/spill, heartbeat re-report, and replication), so a sealed block always reports its checksum and the master can safely reject a checksum-less report. Five .internal() configs gate it, splitting "compute a checksum" (spark.storage.rddBlockChecksum.*) from "seal a local checkpoint" (spark.checkpoint.local.verifyChecksum.*). Only serialized storage levels (e.g. DISK_ONLY) are covered; the uncovered cases (a deserialized level, or blocks materialized before the RDD is marked) are surfaced via a warning, not silently.

Why are the changes needed?

Non-deterministic local checkpoints can silently produce inconsistent reads across a partition, manifesting downstream as data-correctness failures (row-count / invariant violations). Speculation can be disabled for locally-checkpointed task sets, but retries are mandatory for fault tolerance, so "just disable speculation" does not cover the retry / zombie-task case.

This particularly affects Delta Lake's MERGE source materialization, which localCheckpoints the source DataFrame at a serialized DISK_ONLY level to cut lineage before MERGE's multiple jobs. A non-deterministic source materialized inconsistently across those jobs surfaces as MERGE invariant / row-count violations in the written table. This is a general Spark-core correctness gap; Delta MERGE is the motivating downstream consumer.

Does this PR introduce any user-facing change?

Yes, as a correctness fix (all configs are .internal()). With the feature on by default, a non-deterministic locally-checkpointed RDD now yields one consistent version across all reads of a partition, instead of readers potentially disagreeing. Deterministic checkpoints are unaffected. Setting spark.checkpoint.local.verifyChecksum.enabled=false restores the prior behavior.

How was this patch tested?

New unit tests:

• SerializerManagerSuite: wrapForChecksum produces equal checksums for identical bytes, different for divergent bytes.
• LocalCheckpointSuite: the verify mark is set only for serialized checkpoints; forceSerialized bumps a default checkpoint to a serialized level.
• BlockManagerSuite: sealRddChecksums keeps the authoritative checksum and evicts disagreeing replicas; a sealed block rejects a divergent (and a checksum-less) registration and admits a matching one; the checksumless-partition count is reported; sealed checksums are cleared on RDD removal; agreeing disk and _SER replicas survive the seal; the UpdateBlockInfo Externalizable round-trip preserves the checksum; divergence injection - two executors materialize one value and a third a divergent one under the same block id, after which the seal evicts the minority and the diverged executor no longer serves its stale copy (exercising the read-side self-check); the compute-vs-seal split (a compute-only checksum does not opt a block into the read-side check); BlockReplicationMetadata round-trips thto-end DISK_ONLY_2 replication registers a matching checksum on the replica; and the verifyOnReplication recompute works on the stream-upload path.

A microbenchmark, RddBlockChecksumBenchmark, measures the store-time overhead of the checksum layer (a few percent of serialize time uncompressed, largion is on); committed results in core/benchmarks/RddBlockChecksumBenchmark-results.txt.

A deterministic end-to-end test of scheduler-driven divergence (real speculation, or a zombie task) is impractical - Spark kills redundant attempts and aduler orchestration - so divergence is injected at the BlockManager level, which exercises the same feature response (seal -> evict -> read-side skip).

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

Generated-by: Claude Code Opus 4.8 (Anthropic)

[juliuszsompolski]

@cloud-fan

[cloud-fan]

0 blocking, 0 non-blocking, 0 nits. LGTM — a correctness fix of unusually high quality, reusing the established SPARK-41497 master-side tracking pattern with additive, source-compatible signatures and an exhaustive test suite.

Verification

Traced the dedup-and-seal mechanism end-to-end:

• Checksum determinism: wrapForChecksum composes below compression and above the encrypting sink (SerializerManager, MemoryStore.scala:748-750), and DiskStore encrypts below the channel with a random IV — so the checksum covers the representation-independent plaintext and agrees across the disk / _SER / replica paths (confirmed by the disk-vs-_SER survival test).
• Seal atomicity vs. async eviction: sealRddChecksums records the sealed value and drops divergent replicas from the directory synchronously on the master's message-handler thread; only the per-executor RemoveBlock is fire-and-forget. Read correctness never depends on that eviction landing — it rests on the read-side self-check (getLocalValues skips a non-matching local copy and falls through to a remote authoritative location) plus master rejection of divergent re-registration. The isolated "read-side self-check skips a divergent local copy (no eviction)" test pins this.
• Sealed block never wrongly evicted on a checksum-less report: every add-path report site forwards info.checksum (store, reportAllBlocks heartbeat, memory drop, replication); the two None report sites are removals that take the remove branch and never hit checksumSealRejectsUpdate.
• Lockstep invariant: blockChecksums / sealedChecksums are cleared alongside blockLocations at every mutation site (block/rdd/BM removal, per-replica removal on invalid level).

[cloud-fan]

let's make the CI happy