@pytest.mark.parametrize('scheduler', ['threads', 'processes'])
def test_rearrange(shuffle_method, scheduler):
    df = pd.DataFrame({'x': np.random.random(10)})
    ddf = dd.from_pandas(df, npartitions=4)
    ddf2 = ddf.assign(_partitions=(ddf.x % 4))
    result = rearrange_by_column(ddf2, '_partitions', max_branch=32, shuffle=shuffle_method)
    assert (result.npartitions == ddf.npartitions)
    assert set(ddf.dask).issubset(result.dask)
    a = result.compute(scheduler=scheduler)
    get = dask.base.get_scheduler(scheduler=scheduler)
    parts = get(result.dask, result.__dask_keys__())
    for i in a._partitions.drop_duplicates():
        assert (sum(((i in set(part._partitions)) for part in parts)) == 1)
@pytest.mark.parametrize('scheduler', ['threads', 'processes'])
def test_rearrange(shuffle_method, scheduler):
    df = pd.DataFrame({'x': np.random.random(10)})
    ddf = dd.from_pandas(df, npartitions=4)
    ddf2 = ddf.assign(_partitions=(ddf.x % 4))
    result = rearrange_by_column(ddf2, '_partitions', max_branch=32, shuffle=shuffle_method)
    assert (result.npartitions == ddf.npartitions)
    assert set(ddf.dask).issubset(result.dask)
    a = result.compute(scheduler=scheduler)
    get = dask.base.get_scheduler(scheduler=scheduler)
    parts = get(result.dask, result.__dask_keys__())
    for i in a._partitions.drop_duplicates():
        assert (sum(((i in set(part._partitions)) for part in parts)) == 1)