@pytest.mark.slow
def test_gh_2730():
    large = pd.DataFrame({'KEY': np.arange(0, 50000)})
    small = pd.DataFrame({'KEY': np.arange(25, 500)})
    dd_left = dd.from_pandas(small, npartitions=3)
    dd_right = dd.from_pandas(large, npartitions=257)
    with dask.config.set(shuffle='tasks', scheduler='sync'):
        dd_merged = dd_left.merge(dd_right, how='inner', on='KEY')
        result = dd_merged.compute()
    expected = large.merge(small, how='inner', on='KEY')
    tm.assert_frame_equal(result.sort_values('KEY').reset_index(drop=True), expected)
@pytest.mark.slow
def test_gh_2730():
    large = pd.DataFrame({'KEY': np.arange(0, 50000)})
    small = pd.DataFrame({'KEY': np.arange(25, 500)})
    dd_left = dd.from_pandas(small, npartitions=3)
    dd_right = dd.from_pandas(large, npartitions=257)
    with dask.config.set(shuffle='tasks', scheduler='sync'):
        dd_merged = dd_left.merge(dd_right, how='inner', on='KEY')
        result = dd_merged.compute()
    expected = large.merge(small, how='inner', on='KEY')
    tm.assert_frame_equal(result.sort_values('KEY').reset_index(drop=True), expected)