Reuse identical shuffles in P2P hash join #8306
Conversation
Unit Test ResultsSee test report for an extended history of previous test failures. This is useful for diagnosing flaky tests. 27 files ± 0 27 suites ±0 16h 41m 21s ⏱️ + 1h 13m 39s For more details on these failures, see this check. Results for commit 596adff. ± Comparison against base commit 7d0ba9c. ♻️ This comment has been updated with latest results. |
fjetter
left a comment
fjetter
left a comment
There was a problem hiding this comment.
changes LGTM. I'm just having a little difficulty wrapping my head around what exactly we're now caching
|
|
||
|
|
||
| @gen_cluster(client=True) | ||
| async def test_merge_p2p_shuffle_reused_dataframe_with_different_parameters(c, s, a, b): |
There was a problem hiding this comment.
This test passes for me on main. From what I can tell, this is expected, isn't it?
There was a problem hiding this comment.
Yes, there's no reuse on main. This test makes sure that we don't reuse too much (which is what originally happened in dask-expr).
| # Generate the same shuffle IDs if the input frame is the same and all its parameters match | ||
| assert sum(id_from_key(k) is not None for k in out.dask) == 3 |
There was a problem hiding this comment.
This took me a little while to wrap my head around. Maybe an additional comment is helpful.
What's happening here is
- Shuffle1: shuffle ddf1
- Shuffle2: shuffle ddf2
- Those two shuffle will perform the first merge
- Shuffle 3: ?? Which DF is now shuffled and what can be reused?
There was a problem hiding this comment.
maybe a comment in the test would be helpful
There was a problem hiding this comment.
Shuffle 3 is again ddf2? I'm currently wondering why there is a third shuffle. Shouldn't this all be doable with just two?
There was a problem hiding this comment.
I've adjusted the test a bit, I mixed something up when moving it from dask-expr. The idea with using three shuffles is that I want to guarantee that sharing works between multiple merges irrespective of the side the dataframe is assigned.
2 participants
IFF a shuffle has the exact same input and parameters, P2P hash joins should reuse it.
Identical optimization as in dask/dask-expr#361
pre-commit run --all-files