viirya commented on PR #58097: URL: https://github.com/apache/spark/pull/58097#issuecomment-5376850523
Thanks for the thorough pass -- all four plus the minors are addressed. Pushed on top of the branch. **1. Partial-read deadlock through a non-identity chain.** Your invert-the-fallback suggestion is what I went with, plus a bit more. `submitMissingTasks` now maps the live reduce set through `NarrowDependency.getParents` (`liveReduceSet`), threading the read-partition subset through OneToOne (identity), RangeDependency (union's offset), and join fan-in -- so `join`/`union` + LIMIT map correctly instead of falling back to all-live. When the narrow chain reaches the shuffle but the mapping is uncomputable *and* the read is partial, it fails fast (leaves it regular) rather than risking a hang, exactly as you proposed. Your `coalesce(...).limit(n)` case turned out to deadlock regardless of the live set (one reduce task drains several reduce partitions sequentially against a bounded queue), so a `coalesce` reading a shuffle is now left regular altogether. Added union/join + LIMIT and coalesce-fallback tests. **2. `toFlip` matches structurally, not by identity.** Fixed -- `toFlip` is now keyed on `SparkPlan.id`. `transformDown` matches each original node before rebuilding it, so its id is the same instance id the collector recorded; a structural twin the collector left on a blocked path is no longer flipped, regardless of `spark.sql.exchange.reuse`. Added a unit test that builds a free exchange plus a structurally identical blocked twin and asserts only the collected one flips (it fails on the old structural key, passes on the id key). **3. Cleaner arm gated on the manager, not the shuffle.** Fixed -- the channel manager now tracks the shuffle ids it holds, and the tracker-less arm is gated on `!usesStreamingShuffleOutputTracker && holdsShuffle(id)`. A regular shuffle in a feature-on session (including the materialized prefix of a mixed job) is never held, so it stays a no-op there; a channel shuffle's own eager-then-GC re-clean is now a clean no-op too. **4. Per-boundary graph walks.** Folded into the single existing traversal with a `belowRegular` flag, so the per-boundary re-walks are gone. One caveat worth flagging: a single `belowRegular` bool with RDD-only dedup is unsound under DAG sharing -- a node reachable both above and below a regular boundary would take the first-reached context and could miss a pipelined dep only reachable below (a wrongly-accepted job). So the walk keys its visited set on `(RDD, belowRegular)`: a node is visited at most twice, keeping it O(graph) rather than O(K x graph) while staying correct. **Minor notes.** The PR description now lists all three configs and drops the stale `clearAbandoned` reference; the two stale comments in the manager are corrected. -- 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]
