NoahKusaba opened a new issue, #2403: URL: https://github.com/apache/datafusion-ballista/issues/2403
**Is your feature request related to a problem or challenge? Please describe what you are trying to do.** ShuffleReaderExec::execute orders its fetch list in two steps with separate goals: // Sort partitions for evenly send fetching partition requests to avoid hot executors within one task ... .sorted_by(|(p1_idx, _), (p2_idx, _)| Ord::cmp(p1_idx, p2_idx)) ... // Shuffle partitions for evenly send fetching partition requests to avoid hot executors within multiple tasks partition_locations.shuffle(&mut rng()); The sort builds a round-robin so one task spreads its in-flight window across producers. The shuffle then discards it: Fisher-Yates yields a uniform permutation regardless of input order, so nothing of the sort survives. The intra-task goal is not met today. This was raised in review on #256: ▎ thinkharderdev: Why do we sort by index before shuffling here? ▎ ▎ yahoNanJing: Sorting first may be helpful for reducing the bias of random chosen. Maybe it's not necessary. Sorting does not reduce bias in a uniform shuffle, but it does serve the purpose in its own comment. That purpose is what gets thrown away. The inter-task goal is left to independent RNGs, so peers can still aim their first fetches at the same producer. Random spread leaves collisions. There is also nothing available to do better with. restrict_plan_to_partitions re-indexes each task's slice from zero, so every task's reader looks like partition 0 and cannot tell which peer it is. **Describe the solution you'd like** Stamp each local output partition with its global index among the stage's peers, carry it to the executor, and use it to rotate a deterministic round-robin over producers. Round-robin gives the intra-task spread the sort was meant to provide. The rotation staggers peers exactly rather than probabilistically. **Describe alternatives you've considered** - Drop the sort, keep the RNG. Matches what the code effectively does today and answers the #256 question, but leaves peers colliding. - Seed the RNG deterministically. Nothing stable to seed with. After restriction every reader is partition 0, so peers seed identically. - Assign order in the scheduler at dispatch. Needs runtime state for a result a peer index gives locally. **Additional context** Needs one new field on ShuffleReaderExecNode. An absent field decodes to the identity mapping, so plans from an older scheduler keep working. -- 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]
