viirya commented on PR #58097: URL: https://github.com/apache/spark/pull/58097#issuecomment-5513411546
Thanks for the thorough pass -- all ten are addressed. Pushed as `9c40252`, `52b560d`, `51e5396`. **1. Streaming plans.** `PipelinedShuffleEligibility` now bails out when any node's logical link `isStreaming`, using the same idiom `InsertAdaptiveSparkPlan` uses to keep AQE off streaming plans, so the blanket rule never reaches an `IncrementalExecution` plan and `markStreamingPath` keeps deciding the static side. Pinned by a test that a streaming query in a feature-on session comes out with no pipelined exchange. **2. Capability gate for the RTM path.** Added `PipelinedShuffleManager.supportsLiveReducePartitionHints` (default `false`), overridden `true` only by the channel manager. It gates both the epoch stamping and the live-set computation, so an RTM job on the streaming manager takes the master path with no cloned `Properties`, no `liveReduceSet`/`rddReachesShuffle`, and no abort. A `DAGSchedulerSuite` test asserts a manager without the capability gets nothing stamped. **3. The early-out.** You are right that the branch is unreachable -- `initializeShuffleManager` always instantiates it -- so I restored `classifyJobShuffleKinds` as the cheap pre-pass and run the two-context walk only when a job has both kinds. The misleading comment is gone. One correction to my own claim, since it affects how you read the fix: for all-regular and all-pipelined jobs this is a pure cost restoration with no observable behavioral difference, so no behavior test can prove it. I pinned what is testable instead -- the pre-pass contract for all three non-mixed shapes, plus mix detection, where breaking the fix does make the test fail. **4. Transport-agnostic cleanup.** Restored `holdsShuffle` on the trait with the channel manager delegating to the rendezvous; `ContextCleaner` no longer names a transport and reads `sc.env.pipelinedShuffleManager`. Answering from the rendezvous rather than a registry is deliberate -- a shuffle unregistered before its job runs recreates its queues -- and that case has a regression test. `outputTrackerMaster` now uses `sc.env` too (`51e5396`). **5. Rendezvous keying.** Nested by `shuffleId` first, so `holdsShuffle` and `removeShuffle` are O(1) map lookups and `removeShuffle` is one atomic `remove` per map. **6. The identity assumption.** Replaced with a `ShuffleReducePartitionMapping` trait that the reader RDD implements: `ShuffledRDD` reports identity, `ShuffledRowRDD` answers from the partition spec (width-1 `CoalescedPartitionSpec`, `PartialReducerPartitionSpec`), and anything else returns `None`, which marks the read unmappable and falls back to all-partitions-live. So the scheduler no longer infers a mapping from a count, and the invariant no longer rests on a SQL rule's behavior. Also added the dropped-record debug counter you suggested, as the second line of defense. **7. Fetch-wait.** Now times the whole wait once and converts at the end, as `withFetchWaitTimeTracked` does. **8. Benchmark.** Shapes are now derived from the core count so the workflow's 4-core runner can regenerate the file, and the `local[48]` case is dropped. Worth flagging what this exposed: the demand is not `inputParts + shufflePartitions + 1`. `pipelinedJobConcurrentTaskDemand` adds the map-side width of every distinct pipelined shuffle, and in the two-exchange `groupBy(k).count+orderBy(k)` chain the middle stage's width counts twice, so that shape needs `inputParts + 2*shufflePartitions + 1`. My first attempt at shrinking used the wrong formula and failed admission at 20 slots vs 16 free. Sizing off that worst case, 4 cores fit `1+1+1+1`. A side effect is that the regenerated file now contains the AQE-on chain, which the old shape had been crashing on. The committed file is a local run, so I have kicked off the benchmark workflow on this branch for JDK 21 and 25 with `create-commit=true`; it will overwrite that file with runner-generated numbers and add the missing `-jdk25-` variant. **9. Config naming and surface.** Renamed to `spark.shuffle.channel.batchSize` / `spark.shuffle.channel.queueCapacity`. Kept the SQL flag public and made it self-sufficient in the sense you described: it now warns once when the manager does not match, instead of only `logDebug`. **10. Small things.** The scaladoc sentence is replaced with the actual reason the rule is blanket (the scheduler requires an all-pipelined job, so a per-exchange choice would just create rejected mixes); both rules are `object`s; the `stringArgs` comment is corrected; and the three suites' harness is now one `PipelinedShuffleTestSession` trait. On tearing down a session the suite did not create: the feature is gated on configs that must be present at session creation, so `SharedSparkSession` + `withSQLConf` cannot express it -- I documented that in the trait rather than leaving it implicit, but if you know a precedent I would rather follow it. Comment density is trimmed and the review-history narration moved out. One item I looked at and did not do, so it does not look like an oversight: making the limit operators' hidden shuffle pipelined-aware (option A) is viable -- the reads are width-1 and `prepareShuffleDependency` already installs the detaching write processor -- but it touches shared `limit.scala` and adds an execution-time mechanism, so I would rather do it as a follow-up JIRA than widen this PR. The current change blocks those operators instead. -- 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]
