viirya commented on PR #57128: URL: https://github.com/apache/spark/pull/57128#issuecomment-4921311633
Thanks @dongjoon-hyun for the detailed analysis. I agree with the concern, and I think it can be pushed one step further, which changes what the right fix looks like. The local read rewrite is per-task block-count preserving by construction. It reuses the coalesced read's parallelism `P`, so each rewritten task covers `numMappers * numReducers / P` (mapper, reducer) cells — exactly the block count of the coalesced task it replaces. Only the shape changes: from "all mappers x narrow reducer range, fetched over the network" to "one mapper x wide reducer range, read locally". Since the overhead SPARK-58004 targets (per-task metadata and fetch fan-in) scales with block count, the rewrite never actually worsens it; only the literal reducer span grows, by a factor of ~`numMappers`. A concrete example with `numMappers = 10`, `numReducers = 1000`, `maxReducerPartitionsPerTask = 100`: | | tasks | span/task | blocks/task | read | |---|---|---|---|---| | coalesced read (kept by this PR) | 10 | 100 | 1000 | remote | | local read it rejects | 10 | 1000 | 1000 | local | The rejected plan is strictly better, which is why the check as written effectively disables local reads whenever the bound was binding. This also means comparing `PartialMapperPartitionSpec` spans against `max * numMappers` would (modulo integer-division rounding) never trigger — it degenerates to no check at all. The same holds for the `CoalescedMapperPartitionSpec` branch, which is likewise block-neutral. So I think the real decision is what contract the config makes, and the two candidate fixes follow from it: - If the config is a proxy for per-task fan-in overhead (which matches the SPARK-58004 motivation), the cleanest fix is to drop the guard entirely and instead document in the config doc that the limit constrains coalesced specs, and that a later local read rewrite may replace them with mapper-oriented specs whose literal reducer span exceeds the limit while keeping per-task block counts unchanged. Otherwise users who inspect the final plan will reasonably report the config as broken. - If the literal span bound must hold for every task, then regenerating the specs with higher parallelism (your second suggestion) is the way to go: `P' ~= numMappers * ceil(numReducers / max)` gives tasks that each read at most `max` blocks from one mapper, locally. In the example above that is 100 tasks x 100 blocks, satisfying the bound while keeping the local read. The cost is a task count multiplied by `numMappers`, so it should probably fall back to keeping the coalesced read when `P'` becomes excessive (e.g. `numReducers = 100k`, `max = 10`, `numMappers = 100` would yield 1M tasks), making the current PR's revert the last resort rather than the only behavior. My preference is the first option, since I could not find a cost in the local read path that scales with the literal span rather than the block count — but either way the current all-or-nothing revert seems to be the least favorable choice for users who set this config. -- 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]
