viirya commented on PR #58097: URL: https://github.com/apache/spark/pull/58097#issuecomment-5455815109
Thanks -- this was a very useful pass. All ten are addressed; four commits pushed. Notes where my fix differs from the suggestion, or where the cost is worth calling out. **1. Cartesian N-to-1 read.** Confirmed and fixed. `UnsafeCartesianRDD` computes each child partition once per right partition, so N reduce tasks mint N readers on one `(shuffleId, epoch, pid)` queue -- rows and markers split, a reader short of `numMaps` hangs, and the first to finish abandons the queue under the others. As you note neither the fan-out check (one consumer RDD, computed many times) nor the width-1 `require` catches it. A shuffle read through a `CartesianProductExec` now leaves the plan regular. **2. coalesce over a binary node.** Confirmed and fixed -- the walk now recurses through every child and stops at the first exchange on each path. Worth recording how the regression test went: my first attempt used two identical `repartition($"k")` branches and passed even with the old guard, because exchange reuse collapsed them and the reuse gate bailed out first, and ~25K rows per reduce partition never filled the 64-batch queue. The committed test uses two structurally different shuffles (`groupBy` vs `repartition`) and enough rows to park the writer; it deadlocks with the old guard and passes with the new one. **3. Hidden shuffles in the limit operators.** Confirmed and fixed by blocking all three in both rules -- but the block is broader than the hazard, and that costs real coverage: `limit(n).collect()` shapes are no longer pipelined even though `executeCollect` never builds the hidden shuffle. I tried to narrow it to "not at the plan root" and measured that this does not work: for `groupBy.count().orderBy(c).limit(5)`, `TakeOrderedAndProject` sits at the root both for `.collect()` (safe) and for `.write.parquet(...)` (hazardous), and the write path re-plans to the same root shape -- the rule sees an identical plan and cannot know the action. So the broad block is deliberate. Your alternative (emit a pipelined-aware dependency) does look feasible: `prepareShuffleDependency`'s `pipelined` branch already installs the detaching write processor, the hidden shuffle is `SinglePartition` so the width-1 read is satisfied trivially, and `doExecute` already holds `child.execute()` so it could inspect the child RDD graph for a `PipelinedShuffleDependency` rather than guessing from the conf. I left it out of this PR because it changes three shared operators in `limit.scala` and introduces execution-time creation of pipelined dependencies -- a mechanism worth its own review. Happy to open a follow-up JIRA if you agree. One consequence I should flag: `PipelinedLimitHangSuite`'s shapes are now regular, so it could no longer exercise the early-stop path and would have passed vacuously. It now asserts the fallback instead, and the live-reduce-set/early-stop machinery moved to three RDD-level partial-read tests in `PipelinedChannelShuffleSuite` (identity, union offsets, zip fan-in). I verified each fails -- hangs to the deadline -- when `liveReduceSet` is deliberately broken. **4. Unmemoized reachability.** Fixed. Reachability is computed once by a memoized, iterative (two-phase post-order) `rddReachesShuffle` that returns its memo, and `liveReduceSet` is now an explicit worklist keyed on `(rdd -> live subset)`. Merging live sets at a shared node is equivalent to the old per-branch recursion plus union because `getParents` distributes over union; a node is re-enqueued only when its set actually grew (set inequality, not size), which terminates in a finite index domain. **5. Cleanup scoped to the manager, not the shuffle.** Fixed, and thank you -- the registry I had was worse than you describe: it is dropped at `unregisterShuffle` and never re-added, so after the unregister-before-run sequence the recreated queues leaked permanently. The arm now keys off `ChannelShuffleRendezvous.holdsShuffle`, i.e. the thing that actually holds the state to free. The manager registry and the `holdsShuffle` trait method are gone. **6. Per-boundary materialization loop.** Fixed -- it runs only when the walk saw a pipelined dependency. I also added the early-out you would probably have asked for next: `classifyJobShuffleShape` returns immediately when no pipelined manager is configured, so a default cluster no longer pays for the graph walk itself on every job submission. **7. Config versions.** Fixed: the three new configs move to 4.4.0. Which also answers your RC1 question -- with branch-4.3 cut this cannot land in 4.3.0, so I have set them to the release it would actually first ship in. **8. Eager batch allocation.** Fixed -- allocated per partition on first record, and a handed-off batch leaves `null` behind rather than eagerly allocating a successor. **9. Duplicated eligibility gate.** Fixed -- extracted to `PipelinedShuffleEligibility`, used by both rules. **10. `MultiShuffleManager` + channel manager.** Confirmed; the `Some` arm now consults `usesStreamingShuffleOutputTracker` rather than mere tracker presence, so the documented invariant and the routing-suite assertion now match the behavior. My earlier "unreachable by construction" comment missed the `blockingIsMulti` clause -- removed. Tests: core 264 and the SQL pipelined suites 26, all green. -- 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]
