def test_shuffle_hlg_layer():
    ddf = dd.from_pandas(pd.DataFrame({'a': np.random.randint(0, 10, 100)}), npartitions=10)
    ddf_shuffled = ddf.shuffle('a', max_branch=3, shuffle='tasks')
    keys = [(ddf_shuffled._name, i) for i in range(ddf_shuffled.npartitions)]
    dsk = ddf_shuffled.__dask_graph__()
    dsk_culled = dsk.cull(set(keys))
    assert isinstance(dsk_culled, dask.highlevelgraph.HighLevelGraph)
    assert any((isinstance(layer, dd.shuffle.ShuffleLayer) for layer in dsk.layers.values()))
    for layer in dsk.layers.values():
        if isinstance(layer, dd.shuffle.ShuffleLayer):
            assert (not hasattr(layer, '_cached_dict'))
    assert (len(dsk_culled) < len(dsk))
    for (name, layer) in dsk.layers.items():
        if isinstance(layer, dd.shuffle.ShuffleLayer):
            assert name.startswith('shuffle-')
    dsk_dict = dict(dsk_culled)
    (dsk_dict_culled, _) = cull(dsk_dict, keys)
    assert (dsk_dict_culled == dsk_dict)
def test_shuffle_hlg_layer():
    ddf = dd.from_pandas(pd.DataFrame({'a': np.random.randint(0, 10, 100)}), npartitions=10)
    ddf_shuffled = ddf.shuffle('a', max_branch=3, shuffle='tasks')
    keys = [(ddf_shuffled._name, i) for i in range(ddf_shuffled.npartitions)]
    dsk = ddf_shuffled.__dask_graph__()
    dsk_culled = dsk.cull(set(keys))
    assert isinstance(dsk_culled, dask.highlevelgraph.HighLevelGraph)
    assert any((isinstance(layer, dd.shuffle.ShuffleLayer) for layer in dsk.layers.values()))
    for layer in dsk.layers.values():
        if isinstance(layer, dd.shuffle.ShuffleLayer):
            assert (not hasattr(layer, '_cached_dict'))
    assert (len(dsk_culled) < len(dsk))
    for (name, layer) in dsk.layers.items():
        if isinstance(layer, dd.shuffle.ShuffleLayer):
            assert name.startswith('shuffle-')
    dsk_dict = dict(dsk_culled)
    (dsk_dict_culled, _) = cull(dsk_dict, keys)
    assert (dsk_dict_culled == dsk_dict)