Flaky ZeroDivisionError in cugraph's `test_dask_mg_pagerank`

https://github.com/rapidsai/dask-upstream-testing/actions/runs/28849096601/job/85559855975#step:13:2162

_________________________ test_dask_mg_pagerank[False] _________________________
Traceback (most recent call last):
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/_pytest/runner.py", line 353, in from_call
    result: TResult | None = func()
                             ^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/_pytest/runner.py", line 245, in <lambda>
    lambda: runtest_hook(item=item, **kwds),
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_hooks.py", line 512, in __call__
    return self._hookexec(self.name, self._hookimpls.copy(), kwargs, firstresult)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_manager.py", line 120, in _hookexec
    return self._inner_hookexec(hook_name, methods, kwargs, firstresult)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 167, in _multicall
    raise exception
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 139, in _multicall
    teardown.throw(exception)
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/_pytest/logging.py", line 850, in pytest_runtest_call
    yield
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 139, in _multicall
    teardown.throw(exception)
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/_pytest/capture.py", line 900, in pytest_runtest_call
    return (yield)
            ^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 139, in _multicall
    teardown.throw(exception)
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 53, in run_old_style_hookwrapper
    return result.get_result()
           ^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_result.py", line 103, in get_result
    raise exc.with_traceback(tb)
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 38, in run_old_style_hookwrapper
    res = yield
          ^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 139, in _multicall
    teardown.throw(exception)
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 53, in run_old_style_hookwrapper
    return result.get_result()
           ^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_result.py", line 103, in get_result
    raise exc.with_traceback(tb)
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 38, in run_old_style_hookwrapper
    res = yield
          ^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 139, in _multicall
    teardown.throw(exception)
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/_pytest/skipping.py", line 268, in pytest_runtest_call
    return (yield)
            ^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 121, in _multicall
    res = hook_impl.function(*args)
          ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/_pytest/runner.py", line 179, in pytest_runtest_call
    item.runtest()
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/_pytest/python.py", line 1720, in runtest
    self.ihook.pytest_pyfunc_call(pyfuncitem=self)
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_hooks.py", line 512, in __call__
    return self._hookexec(self.name, self._hookimpls.copy(), kwargs, firstresult)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_manager.py", line 120, in _hookexec
    return self._inner_hookexec(hook_name, methods, kwargs, firstresult)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 167, in _multicall
    raise exception
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 121, in _multicall
    res = hook_impl.function(*args)
          ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/_pytest/python.py", line 166, in pytest_pyfunc_call
    result = testfunction(**testargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/packages/cugraph/python/cugraph/cugraph/tests/comms/test_comms_mg.py", line 60, in test_dask_mg_pagerank
    result_pr1 = get_pagerank_result(karate, is_mg=True)
                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/packages/cugraph/python/cugraph/cugraph/tests/comms/test_comms_mg.py", line 40, in get_pagerank_result
    dg = dataset.get_dask_graph(store_transposed=True)
         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/cugraph/datasets/dataset.py", line 383, in get_dask_graph
    G.from_dask_cudf_edgelist(
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/cugraph/structure/graph_classes.py", line 360, in from_dask_cudf_edgelist
    self._Impl._simpleDistributedGraphImpl__from_edgelist(
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/cugraph/structure/graph_implementation/simpleDistributedGraph.py", line 359, in __from_edgelist
    persisted_keys_d = persist_dask_df_equal_parts_per_worker(
                       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/cugraph/dask/common/part_utils.py", line 106, in persist_dask_df_equal_parts_per_worker
    ddf_keys = dask_df.to_delayed()
               ^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_collection.py", line 2521, in to_delayed
    frame = self.optimize()
            ^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_collection.py", line 580, in optimize
    return new_collection(self.expr.optimize(fuse=fuse))
                          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_expr.py", line 97, in optimize
    return optimize(self, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/_expr.py", line 925, in optimize
    return optimize_until(expr, stage)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/_expr.py", line 945, in optimize_until
    expr = expr.lower_completely()
           ^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/_expr.py", line 523, in lower_completely
    new = expr.lower_once(lowered)
          ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/_expr.py", line 478, in lower_once
    out = expr._lower()
          ^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_repartition.py", line 89, in _lower
    if self.new_partitions < self.frame.npartitions:
                             ^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_expr.py", line 446, in npartitions
    return len(self.divisions) - 1
               ^^^^^^^^^^^^^^
  File "/github/home/.local/share/uv/python/cpython-3.12.13-linux-x86_64-gnu/lib/python3.12/functools.py", line 998, in __get__
    val = self.func(instance)
          ^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_expr.py", line 431, in divisions
    return tuple(self._divisions())
                 ^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_concat.py", line 85, in _divisions
    if {df.npartitions for df in self._frames} == {1}:
        ^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_expr.py", line 446, in npartitions
    return len(self.divisions) - 1
               ^^^^^^^^^^^^^^
  File "/github/home/.local/share/uv/python/cpython-3.12.13-linux-x86_64-gnu/lib/python3.12/functools.py", line 998, in __get__
    val = self.func(instance)
          ^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_expr.py", line 431, in divisions
    return tuple(self._divisions())
                 ^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_expr.py", line 594, in _divisions
    if not self._broadcast_dep(arg):
           ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_expr.py", line 585, in _broadcast_dep
    return dep.npartitions == 1 and dep.ndim < self.ndim
           ^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_expr.py", line 446, in npartitions
    return len(self.divisions) - 1
               ^^^^^^^^^^^^^^
  File "/github/home/.local/share/uv/python/cpython-3.12.13-linux-x86_64-gnu/lib/python3.12/functools.py", line 998, in __get__
    val = self.func(instance)
          ^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_expr.py", line 431, in divisions
    return tuple(self._divisions())
                 ^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_expr.py", line 2223, in _divisions
    return super()._divisions()
           ^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_expr.py", line 594, in _divisions
    if not self._broadcast_dep(arg):
           ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_expr.py", line 585, in _broadcast_dep
    return dep.npartitions == 1 and dep.ndim < self.ndim
           ^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_expr.py", line 446, in npartitions
    return len(self.divisions) - 1
               ^^^^^^^^^^^^^^
  File "/github/home/.local/share/uv/python/cpython-3.12.13-linux-x86_64-gnu/lib/python3.12/functools.py", line 998, in __get__
    val = self.func(instance)
          ^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_expr.py", line 431, in divisions
    return tuple(self._divisions())
                 ^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_expr.py", line 595, in _divisions
    assert arg.divisions == dependencies[0].divisions, (
           ^^^^^^^^^^^^^
  File "/github/home/.local/share/uv/python/cpython-3.12.13-linux-x86_64-gnu/lib/python3.12/functools.py", line 998, in __get__
    val = self.func(instance)
          ^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_expr.py", line 431, in divisions
    return tuple(self._divisions())
                 ^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_repartition.py", line 58, in _divisions
    return x._divisions()
           ^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_repartition.py", line 175, in _divisions
    return tuple(self.frame.divisions[i] for i in self._partitions_boundaries)
                                                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/github/home/.local/share/uv/python/cpython-3.12.13-linux-x86_64-gnu/lib/python3.12/functools.py", line 998, in __get__
    val = self.func(instance)
          ^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_repartition.py", line 193, in _partitions_boundaries
    return self._compute_partition_boundaries(npartitions, npartitions_input)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_repartition.py", line 179, in _compute_partition_boundaries
    npartitions_ratio = n_old_partitions / n_new_partitions
                        ~~~~~~~~~~~~~~~~~^~~~~~~~~~~~~~~~~~
ZeroDivisionError: division by zero
----------------------------- Captured stdout call -----------------------------
dataset1=/__w/dask-upstream-testing/dask-upstream-testing/packages/cugraph/datasets/karate.csv
--------------------------- Captured stdout teardown ---------------------------

Dask client closed.
_ test_mg_symmetrize_df[graph_file:packages/cugraph/datasets/karate-asymmetric.csv-edgevals:True-multi_columns:False] _
Traceback (most recent call last):
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/_pytest/runner.py", line 353, in from_call
    result: TResult | None = func()
                             ^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/_pytest/runner.py", line 245, in <lambda>
    lambda: runtest_hook(item=item, **kwds),
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_hooks.py", line 512, in __call__
    return self._hookexec(self.name, self._hookimpls.copy(), kwargs, firstresult)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_manager.py", line 120, in _hookexec
    return self._inner_hookexec(hook_name, methods, kwargs, firstresult)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 167, in _multicall
    raise exception
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 139, in _multicall
    teardown.throw(exception)
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/_pytest/logging.py", line 850, in pytest_runtest_call
    yield
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 139, in _multicall
    teardown.throw(exception)
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/_pytest/capture.py", line 900, in pytest_runtest_call
    return (yield)
            ^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 139, in _multicall
    teardown.throw(exception)
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 53, in run_old_style_hookwrapper
    return result.get_result()
           ^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_result.py", line 103, in get_result
    raise exc.with_traceback(tb)
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 38, in run_old_style_hookwrapper
    res = yield
          ^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 139, in _multicall
    teardown.throw(exception)
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 53, in run_old_style_hookwrapper
    return result.get_result()
           ^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_result.py", line 103, in get_result
    raise exc.with_traceback(tb)
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 38, in run_old_style_hookwrapper
    res = yield
          ^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 139, in _multicall
    teardown.throw(exception)
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/_pytest/skipping.py", line 268, in pytest_runtest_call
    return (yield)
            ^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 121, in _multicall
    res = hook_impl.function(*args)
          ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/_pytest/runner.py", line 179, in pytest_runtest_call
    item.runtest()
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/_pytest/python.py", line 1720, in runtest
    self.ihook.pytest_pyfunc_call(pyfuncitem=self)
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_hooks.py", line 512, in __call__
    return self._hookexec(self.name, self._hookimpls.copy(), kwargs, firstresult)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_manager.py", line 120, in _hookexec
    return self._inner_hookexec(hook_name, methods, kwargs, firstresult)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 167, in _multicall
    raise exception
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/pluggy/_callers.py", line 121, in _multicall
    res = hook_impl.function(*args)
          ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/_pytest/python.py", line 166, in pytest_pyfunc_call
    result = testfunction(**testargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/packages/cugraph/python/cugraph/cugraph/tests/internals/test_symmetrize_mg.py", line 251, in test_mg_symmetrize_df
    compare(ddf, sym_ddf, src_col_name, dst_col_name, val_col_name)
  File "/__w/dask-upstream-testing/dask-upstream-testing/packages/cugraph/python/cugraph/cugraph/tests/internals/test_symmetrize_mg.py", line 74, in compare
    if len(ddf1) != len(join):
                    ^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_collection.py", line 397, in __len__
    return new_collection(Len(self)).compute()
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/base.py", line 392, in compute
    (result,) = compute(self, traverse=False, **kwargs)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/base.py", line 766, in compute
    expr = expr.optimize()
           ^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/_expr.py", line 440, in optimize
    return optimize_until(self, stage)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/_expr.py", line 945, in optimize_until
    expr = expr.lower_completely()
           ^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/_expr.py", line 523, in lower_completely
    new = expr.lower_once(lowered)
          ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/_expr.py", line 489, in lower_once
    new = operand.lower_once(lowered)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/_expr.py", line 489, in lower_once
    new = operand.lower_once(lowered)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/_expr.py", line 489, in lower_once
    new = operand.lower_once(lowered)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  [Previous line repeated 3 more times]
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/_expr.py", line 478, in lower_once
    out = expr._lower()
          ^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_reductions.py", line 226, in _lower
    split_every = getattr(self, "split_every", 0) or chunked.npartitions
                                                     ^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_expr.py", line 446, in npartitions
    return len(self.divisions) - 1
               ^^^^^^^^^^^^^^
  File "/github/home/.local/share/uv/python/cpython-3.12.13-linux-x86_64-gnu/lib/python3.12/functools.py", line 998, in __get__
    val = self.func(instance)
          ^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_expr.py", line 431, in divisions
    return tuple(self._divisions())
                 ^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_expr.py", line 595, in _divisions
    assert arg.divisions == dependencies[0].divisions, (
           ^^^^^^^^^^^^^
  File "/github/home/.local/share/uv/python/cpython-3.12.13-linux-x86_64-gnu/lib/python3.12/functools.py", line 998, in __get__
    val = self.func(instance)
          ^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_expr.py", line 431, in divisions
    return tuple(self._divisions())
                 ^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_repartition.py", line 175, in _divisions
    return tuple(self.frame.divisions[i] for i in self._partitions_boundaries)
                                                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/github/home/.local/share/uv/python/cpython-3.12.13-linux-x86_64-gnu/lib/python3.12/functools.py", line 998, in __get__
    val = self.func(instance)
          ^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_repartition.py", line 193, in _partitions_boundaries
    return self._compute_partition_boundaries(npartitions, npartitions_input)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/__w/dask-upstream-testing/dask-upstream-testing/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_repartition.py", line 179, in _compute_partition_boundaries
    npartitions_ratio = n_old_partitions / n_new_partitions
                        ~~~~~~~~~~~~~~~~~^~~~~~~~~~~~~~~~~~
ZeroDivisionError: division by zero

Start with packages/cugraph/python/cugraph/cugraph/tests/comms/test_comms_mg.py, especially test_dask_mg_pagerank and get_pagerank_result, and reproduce the linked failure. Trace the ZeroDivisionError through dataset.get_dask_graph, from_dask_cudf_edgelist, and persist_dask_df_equal_parts_per_worker. Done means the test no longer intermittently fails during Dask graph construction.