With these changes, everything in test_shuffle.py passes under dask/distributed#5520 except test_shuffle_sort:

@fjetter ready for final review?

I ran test_set_index_overlap 1,000 times locally with 8 concurrent pytest processes and it never failed, which makes me think this will be hard to reproduce. However, the fact that it failed twice in the same way—and that the failure was that a row was simply missing from the shuffle—makes me concerned. Not that these changes have actually broken anything (that test doesn't even use the parametrized shuffle fixture; so it's always using the default shuffle), but that using scheduler=threads instead of scheduler=sync in assert_eq has inadvertently revealed an existing bug.

The process pool tests that failed on Windows both explicitly created a ProcessPool within the test, and only one was even a DataFrame test; neither were related to the sync changes and I think can be ignored. The macOS failures were some weird CI thing with the TileDB conda package. So the only test I'm concerned about is test_set_index_overlap. We'll see what happens with this merge of main.

Welp, test_groupby.py::test_cumulative[disk-cumsum-key1-d] , test_groupby.py::test_cumulative[disk-cumprod-key1-d] and test_shuffle.py::test_set_index_overlap failed in CI. Though they all were set to use the disk-based shuffle, the groupby ones don't even do a shuffle, so the shuffle mode is irrelevant.

I have gotten test_cumulative to fail with the same error locally, but still not test_set_index_overlap.

The tests are vaguely similar, in that they both involve columns with repeated values like [1, 2, 3, 4, 4, 5, 6, 7] (repeated 4, ends up in both partitions). In test_set_index_overlap, the repeated value is getting dropped from the shuffle output. In test_cumulative, the repeated value is getting repeated in groupby output (group 4 shows up twice), or something like that?

Still feeling like these are revealing actual bugs, possibly thread-unsafe code in pandas and/or partd.

As a workaround, if we don't want to deal with these bugs right now, I suppose I could make assert_eq take a scheduler="sync" kwarg, so the default will stay the same for all these tests, but in the P2P shuffle tests we can explicitly pass scheduler=None or scheduler=client.

This is no longer needed to merge dask/distributed#5520. I've pulled out #8559, #8557, and #8396 into separate PRs, and removed them from this branch.

This branch now represents most of the tweaks needed to make all the dask/dask shuffling tests pass using the p2p shuffle, with one glaring exception: every test would have to specify scheduler=None to assert_eq. We'll have to think about how to handle that when we actually decide to run all these tests in distributed's CI.

I'm closing this, but it'll be good to have for posterity whenever we do want to get these tests running with the p2p shuffle.