gaoyajun02 opened a new pull request, #58008: URL: https://github.com/apache/spark/pull/58008
### What changes were proposed in this pull request? Follow-up to SPARK-57491 (stale push-based shuffle data detection). SPARK-57491's reducer fallback was too coarse; this PR makes it chunk-granular and opt-in. 1. Chunk-granularity fallback. Replaces checkStaleMapIdInMergedBlock (which fell back the entire merged block if any stale map index was present) with isStaleChunk, which checks a single chunk's bitmap. The meta fetch no longer short-circuits, so in ShuffleBlockFetcherIterator.next() only the chunk that actually contains a stale index falls back to its original blocks; other chunks of the same merged block keep being read from the merged block. 2. No cascade for stale fallback. initiateFallbackFetchForPushMergedBlock gains a fallbackPendingChunks flag (default true). Fetch-failure paths keep cascading to same-host pending chunks; the stale path passes false, since a stale chunk doesn't imply the host is unhealthy, so pending chunks retain their merged-block reads. 3. Opt-in, indeterminate-scoped reducer fallback. Two new configs (both default false): spark.shuffle.push.stale.fallback.enabled gates markStalePushedPartition; spark.shuffle.push.stale.detectAllStages.enabled extends it to all map stages. Duplicate map attempts are now always logged for observability, but marked stale only when fallback is enabled and the stage is in scope. Stage indeterminacy is resolved once at TaskSetManager construction (isIndeterminateShuffleMapStage). ### Why are the changes needed? Speculation is common on large stages in production, so duplicate map attempts that both push are routine. SPARK-57491's fallback had too wide a blast radius: - Whole-block fallback amplifies fetch load across the whole stage. A stale index in one chunk forced fallback of the entire merged block, turning 1 merged fetch into N original-block fetches per affected reducer. On a large shuffle, a single stale partition multiplies network requests by an order of magnitude and slows the whole stage, rather than protecting just the one partition at risk. - Deterministic stages pay for nothing. Duplicate deterministic attempts produce byte-identical output, so the data in the merged block is actually correct — the fallback is pure overhead and a false positive. The real risk is confined to indeterminate stages, where duplicate attempts can diverge. - Safe rollout. SPARK-57491 is unreleased; its always-on fallback could land as a silent perf regression for any job that hits speculation. Opt-in (default: log-only) fallback lets the detection ship safely, enabled only where the correctness gain justifies the extra fetches. Net effect: a stale partition now costs roughly 1 chunk's worth of extra fetches instead of a full merged block, and only indeterminate stages pay it. ### Does this PR introduce _any_ user-facing change? Yes — two new optional configs (both default false): spark.shuffle.push.stale.fallback.enabled and spark.shuffle.push.stale.detectAllStages.enabled. No change vs released Spark versions: SPARK-57491 is itself unreleased (master 5.0.0-SNAPSHOT / branch-4.x 4.4.0-SNAPSHOT). Within the unreleased branches, the default stale-push behavior changes from "always mark + whole-merged-block fallback" to "log-only by default; chunk-granularity fallback when enabled". ### How was this patch tested? Added/updated unit tests: - TaskSetManagerSuite: updated the SPARK-57491 test to enable both fallback switches and assert the partition is marked stale; new test verifies that with fallback disabled (default) the stale attempt is reported but not marked. - ShuffleBlockFetcherIteratorSuite: updated to chunk granularity — a stale index in a chunk's bitmap triggers fallback for that chunk only; new case asserts the non-stale chunk is still read as a ShuffleBlockChunkId. ### Was this patch authored or co-authored using generative AI tooling? Yes. Generated-by: Claude Code (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]
