code: def test_multiple_linear(client):
      df = dd.demo.make_timeseries(freq="15D", partition_freq="30D")
      df["name"] = df["name"].astype("string[python]")
      with dask.config.set({"dataframe.shuffle.method": "p2p"}):
          s1 = df.shuffle("id")
      s1["x"] = s1["x"] + 1
      with dask.config.set({"dataframe.shuffle.method": "p2p"}):
          s2 = s1.shuffle("x")
  
      with dask.config.set({"dataframe.shuffle.method": "tasks"}):
          expected = df.assign(x=lambda df: df.x + 1).shuffle("x")
      # TODO eventually test for fusion between s1's unpacks, the `+1`, and s2's `transfer`s
  
      dd.utils.assert_eq(
          s2,
          expected,
          scheduler=client,
      )
  
file_path: distributed/distributed/shuffle/tests/test_graph.py
imports: from __future__ import annotations
  import asyncio
  import pytest
  import dask
  import dask.dataframe
  from dask.blockwise import Blockwise
  from dask.utils_test import hlg_layer_topological
  from distributed.utils_test import gen_cluster
label_desc: Use image node_id 7 for calling a global function with example usage: test_multiple_linear(client) without return types