juliuszsompolski opened a new pull request, #57232: URL: https://github.com/apache/spark/pull/57232
### 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 `localCheckpoint`s 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) -- 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]
