[SPARK-58107][CORE] Detect and repair non-deterministic local checkpoints via RDD-block content checksums#57232
Open
juliuszsompolski wants to merge 24 commits into
Conversation
…ints via RDD-block content checksums ### What changes were proposed in this pull request? Add an opt-in dedup-and-seal mechanism for locally-checkpointed RDDs. When enabled, the BlockManager folds a JDK content checksum (CRC32C by default) over each cache block's serialized+compressed plaintext at store time and reports it per replica to the driver. At the checkpoint commit point, LocalRDDCheckpointData seals the RDD: the master keeps the plurality checksum per partition, evicts divergent replicas, rejects future divergent registrations, and reads self-check against the sealed value. This makes every later read of a non-deterministically-materialized local checkpoint converge on one consistent version. Two internal configs, off by default: - spark.checkpoint.local.verifyChecksum.enabled (default false) - spark.storage.rddBlockChecksum.algorithm (default CRC32C) ### Why are the changes needed? A locally-checkpointed partition materialized by more than one successful task attempt (speculation, or a zombie task of a superseded stage attempt) can produce divergent bytes for the same RDDBlockId; the master appends both locations and a reader picks one at random, so reads disagree. Local checkpointing cuts lineage, so unpersist+recompute is not a recovery. This converges readers on one version. ### Does this PR introduce any user-facing change? No. Both configs are internal and default off. ### How was this patch tested? New unit tests in SerializerManagerSuite (checksum determinism) and BlockManagerSuite (seal plurality/evict, reject+admit, checksumless-count, rdd-removal cleanup, disk-vs-_SER agreement, UpdateBlockInfo Externalizable round-trip, and divergence injection). Added RddBlockChecksumBenchmark for the store-time overhead. Co-authored-by: Isaac
…default Flip spark.checkpoint.local.verifyChecksum.enabled to default true so the dedup-and-seal path is exercised broadly (CI, benchmarks, local checkpoints) while this WIP is under review. Still internal (.internal()); revisit the default before any non-WIP merge. Co-authored-by: Isaac
…point seal Separate the two concerns behind distinct configs: - spark.storage.rddBlockChecksum.enabled (default false): compute a content checksum for every serialized RDD cache block at store time and report it to the driver. On its own it only records checksums (a future consumer can use them, e.g. to log divergent replicas) - no sealing. - spark.checkpoint.local.verifyChecksum.enabled (default true): seal local checkpoints to one consistent version. This implies checksum computation for its RDDs regardless of the global flag. newRddBlockChecksum now computes when (caller requested || global enabled) && isRDD, so the seal path is unaffected and the global switch works on its own. Adds a BlockManagerSuite test that the global flag records a checksum with no seal request. Co-authored-by: Isaac
…having a checksum After splitting compute from seal, spark.storage.rddBlockChecksum.enabled makes every serialized RDD block carry a checksum, so gating the read-side seal self-check on info.checksum.isDefined wrongly opted compute-only blocks into it - each read pulled the master's sealed checksum (uncached for never-sealed blocks). Carry the seal intent to BlockInfo.verifySealed (set = computeChecksum, i.e. the local-checkpoint seal path only) and gate the read-side check on it. Compute-only blocks are now served with no read-path master round-trip. Extends the independence test to read the block before any seal. Co-authored-by: Isaac
…ehavior) Add a BlockManagerSuite test that is the compute-only mirror of the seal divergence test: with spark.storage.rddBlockChecksum.enabled on and no seal request, divergent replicas of a block are recorded but coexist - both stay in the directory, both are served locally, and no sealed checksum exists - proving checksum computation on its own only observes and never converges. A later explicit seal still converges them (observe-then-enforce). Co-authored-by: Isaac
…Sealed test Review follow-ups: - Reorder the compute gate to `rddBlockChecksumEnabled || compute`. - Trim BlockInfo.checksum / verifySealed / sealedChecksum docs and the store-time and read-side comments: state what the field/line is, don't re-explain the code elsewhere, and drop the "local-checkpoint" over-qualification (the mechanism is just sealing). - Replace the weak "computes checksums without a seal request" test with one that directly asserts verifySealed: true for a seal-path store, false for a compute-only store that nonetheless has a checksum. This is the assertion that fails if seal-intent were derived from checksum presence. Co-authored-by: Isaac
… tidy naming/comments Addresses review feedback: - Thread seal intent as its own param: rename computeChecksum -> verifySealedChecksum through getOrElseUpdateRDDBlock -> getOrElseUpdate -> doPutIterator, so info.verifySealedChecksum = verifySealedChecksum is tautological and "compute a checksum" is decided (global-flag || seal-intent) inside newRddBlockChecksum. No overloaded flag. - Move the seal out of LocalRDDCheckpointData to RDD.sealChecksums() (private[rdd]), called guarded from LocalRDDCheckpointData.doCheckpoint; warn+noop if invoked on an unmarked RDD. Doc no longer mentions "checkpoint". - Rename BlockInfo.verifySealed -> verifySealedChecksum, localCopyMatchesSeal -> localCopyMatchesSealedChecksum; RDD.verifySealedChecksum private[spark] -> private[rdd]. - Trim config/field/inline comments (state what it is, don't re-explain elsewhere; drop "local-checkpoint" over-qualification and the false "only on-disk blocks are checksummed" - _SER memory is checksummed too). - Flip newRddBlockChecksum gate to rddBlockChecksumEnabled || verifySealedChecksum. Co-authored-by: Isaac
…rn+noop when unmarked Bring OSS in line with the DBR counterpart (self-review found it lagging): the seal method is RDD.sealChecksums(), called guarded from LocalRDDCheckpointData.doCheckpoint, and warns+noops if invoked on an RDD not marked for verification. Co-authored-by: Isaac
…rialized config; rename mark/seal - Gate the verify mark on a serialized storage level (no warning on the deserialized default; the checkpoint is simply not marked). - Add spark.checkpoint.local.verifyChecksum.forceSerialized (default false) to opt a default checkpoint into a serialized level so it can be verified. - Reword eager/lazy sealing as finalization-time (eager sealed before consumers; lazy sealed after the first materializing job, in-job reads may see unsealed copies). - Rename RDD.sealChecksums -> sealCheckpointChecksums and the RDD mark verifySealedChecksum -> verifyCheckpointChecksums (layers below RDD keep verifySealedChecksum). - Apply review feedback; add LocalCheckpointSuite gating tests. Co-authored-by: Isaac
- verifyCheckpointChecksums doc: note it also informs the executor-local BlockInfo so the read-side self-check enforces the seal. - sealCheckpointChecksums doc: "picks one checksum value" (not "plurality"); explain the doCheckpoint timing as post-runJob-materialization + pre-lineage-cut (no future recompute to reconcile, lost blocks lost forever); require the mark be set before materialization or blocks go unchecksummed. Co-authored-by: Isaac
…cutors Executors need to know in advance - before storing/reading a block - whether to compute its checksum and later apply the read-side sealed-checksum self-check. Co-authored-by: Isaac
- localCopyMatchesSealedChecksum: explain why we re-pull the sealed value while unsealed rather than caching an "unsealed" sentinel or resolving once at getOrElseUpdateRDDBlock entry. - getLocalValues self-check: note localCheckpoint's lineage is cut once sealed, so a skipped copy means the block is genuinely gone (never a recomputed, re-checksummed supersession). - Reject-divergent path: note why it returns true (false would trigger a re-register loop). - MemoryStore: drop the "same as the disk path" aside. Co-authored-by: Isaac
…tic error at readback The failToGetBlockWithLockError throw stays a genuine internal-error indicator: the read-side seal self-check can't fire on this readback because the seal runs at doCheckpoint finalization (after all partitions are stored, before lineage is cut), never between doPutIterator and this get of the copy the task just wrote. Co-authored-by: Isaac
…ocks (checksum on every path) Close the gaps where a sealed local-checkpoint block could be reported to the master without a checksum (and thus wrongly evicted) or replicated unverified: - Heartbeat re-registration (reportAllBlocks) and memory-eviction (dropFromMemory) now forward info.checksum, so a re-report of a sealed block carries it. - dropFromMemory's serialize-to-disk and the store-time putIteratorAsBytes spill now compute and record the block's checksum (they previously wrote to disk without one). - Block replication carries the source's content checksum + seal mark in a new BlockReplicationMetadata (the Java-serialized UploadBlock/UploadBlockStream metadata, an intra-application executor-to-executor RPC). The receiver recomputes the checksum over the received bytes (verifying the transfer; a mismatch logs and stores the recomputed value), records it, and propagates the seal mark so the replica is verified against the seal like a primary and self-checks its own reads. checksum and verifySealedChecksum are separate fields so the checksum stays usable without sealing. - The master reject of a sealed block keeps rejecting a checksum-less (None) report: a sealable block always reports a checksum, so a None is anomalous and must not enter the directory (remote reads trust the directory with no self-check). Tests: None-report of a sealed block is rejected; BlockReplicationMetadata serialization round-trip; end-to-end DISK_ONLY_2 replication of a seal-path block registers a matching checksum and both copies survive the seal. Co-authored-by: Isaac
…for new uploadBlock params uploadBlock/uploadBlockSync gained checksum + verifySealedChecksum, so the Mockito stubs and verifies (7 matchers) now need 9 - otherwise Mockito throws "0 matchers expected, 7 recorded". Add mc.any() for the two new args at every stub/verify site. Co-authored-by: Isaac
… to a divergent checksum At the getOrElseUpdate store-then-readback, getLocalValues can return None because the read-side seal self-check skipped a just-stored copy that diverges from the sealed checksum. Previously that surfaced as the cryptic failToGetBlockWithLockError; now distinguish it and throw a clear sealedBlockDivergedError. Unreachable for eager local checkpoints (lineage is cut, so no post-seal recompute) - worded generically as future-proofing for any sealed-but-recomputable use. Co-authored-by: Isaac
…nsmitted checksum by default) Recomputing the checksum over received replica bytes is only needed to verify the transfer, which the transport layer already covers - so by default trust the transmitted source checksum and record it directly. New config spark.storage.rddBlockChecksum.verifyOnReplication (default false) opts into the recompute-and-verify path. Also correct the dropFromMemory comment: its serialize-to-disk (Left) branch is only reached by deserialized in-memory blocks, which are never sealed, so there the checksum is driven purely by the global compute flag. Co-authored-by: Isaac
…ploadBlockSync params Same as the BlockManagerDecommissionUnitSuite fix: uploadBlockSync gained checksum + verifySealedChecksum (7 -> 9 args), so the stub and the never() verify need two more mc.any() matchers or Mockito throws "0 matchers expected, 7 recorded". Co-authored-by: Isaac
…helpers Follow-up polish from the code review of the dedup-and-seal feature: - Mark the three cross-thread BlockInfo checksum fields @volatile (they are read/written outside the block lock: a lockless reportAllBlocks read and a shared-read-lock write in localCopyMatchesSealedChecksum), matching the class's documented lock discipline. - Fold the inline force-serialized StorageLevel rebuild in localCheckpoint into a new transformStorageLevel(forceSerialized) param, and read the verify-enabled flag once. - Add SerializerManager.wrapForChecksum(Option[Checksum], sink) + BlockManager.recordChecksum helpers, collapsing the duplicated wrap-and-record idiom across the store/spill/evict paths. - Extract BlockManagerMasterEndpoint.evictReplica for the duplicated fire-and-forget RemoveBlock (reject path + sealRddChecksums losers). Co-authored-by: Isaac
… out of updateBlockInfo Follow-up to keep the seal-admission policy off the generic (hot) updateBlockInfo body: move the "is this report rejected by the seal" test into a named private predicate `checksumSealRejectsUpdate`, which also carries the rationale for rejecting a checksum-less report of a sealed block. No behavior change - same predicate, same evictReplica + accepted- but-not-added return. Co-authored-by: Isaac
…ck on the store-time readback Follow-up (#2 from the review) to drop the guaranteed master round-trip the seal self-check adds on every partition's store-time readback: - getLocalValues gains a private checkChecksumSeal overload; the store-time readback in getOrElseUpdate passes checkChecksumSeal = false. The public getLocalValues is unchanged, so callers/mocks are untouched. Genuine (consumer) reads still self-check, and a divergent store is rejected at master registration, so this only removes a wasted RPC on the producing job's own readback (a retry cannot overwrite an existing good copy). - Drop the now-unreachable sealedBlockDivergedError: with the readback no longer returning None on divergence, that diagnostic path can never fire, so restore the original getOrElse comment and remove the error from SparkCoreErrors. Co-authored-by: Isaac
…e-read path The serialized-bytes path skips the read-side sealed-checksum self-check that getLocalValues runs. Record why that is safe (its callers are replication and remote-serve, never a local RDD value read; a sealed RDD block's correctness is enforced at the master, and other blocks reached here are never sealed) and the contract that a seal-observing read must use getLocalValues. Co-authored-by: Isaac
…er the store moved it With spark.storage.rddBlockChecksum.verifyOnReplication on, BlockStoreUpdater.save() recomputed the checksum via blockData() *after* saveToDiskStore(). On the stream-upload path (TempFileBasedBlockStoreUpdater, used when a received block exceeds spark.network.maxRemoteBlockSizeFetchToMem, default 200MB), saveToDiskStore() moves the temp file into the block store, so the recompute reopened a moved-away file -> FileNotFoundException, failing the replica store on that peer. The in-memory ByteBufferBlockStoreUpdater path was unaffected, and the flag is off by default. Recompute before saveToDiskStore() (while blockData() is still readable; the recompute closes its stream before the move) and record it on info.checksum only once the store succeeds, via info.checksum = recomputedChecksum.orElse(sourceChecksum) - recompute wins, else the trusted source checksum. New BlockManagerSuite test drives the stream-upload path with the flag on (no >200MB block needed - that path is taken by putBlockDataAsStream regardless of size); pre-fix it threw. Co-authored-by: Isaac
…up-and-seal follow-ups No behavior change: - transformStorageLevel: one named arg per line (was a named/positional mix). - getLocalValues checkChecksumSeal doc + the store-time readback comment: note the readback is usually pre-seal, and a recompute of an already-sealed block is still caught at master registration and at any later local read. - updateBlockInfo seal-reject comment: drop the superfluous "fire-and-forget, mirroring the seal's eviction" aside (evictReplica's own doc covers fire-and-forget). Co-authored-by: Isaac
Contributor
Author
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
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'sblockLocationssimply appends locations, with no version / attempt id) and a reader picks one viaRandom.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:
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 -> doPutIteratorand recorded on the block'sBlockInfo.Seal at the checkpoint commit point.
LocalRDDCheckpointData.doCheckpoint, after materialization and before the checkpoint is exposed to readers, seals each partition on theBlockManagerMasterEndpoint: 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.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 serializedDISK_ONLYlevel 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. Settingspark.checkpoint.local.verifyChecksum.enabled=falserestores the prior behavior.How was this patch tested?
New unit tests:
SerializerManagerSuite:wrapForChecksumproduces equal checksums for identical bytes, different for divergent bytes.LocalCheckpointSuite: the verify mark is set only for serialized checkpoints;forceSerializedbumps a default checkpoint to a serialized level.BlockManagerSuite:sealRddChecksumskeeps 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_SERreplicas survive the seal; theUpdateBlockInfoExternalizableround-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);BlockReplicationMetadataround-trips thto-endDISK_ONLY_2replication registers a matching checksum on the replica; and theverifyOnReplicationrecompute 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 incore/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)