code: def test_blockwise_different_optimization(c):
      # Regression test for incorrect results due to SubgraphCallable.__eq__
      # not correctly handling subgraphs with the same outputs and arity but
      # different internals (GH-7632). The bug is triggered by distributed
      # because it uses a function cache.
      da = pytest.importorskip("dask.array")
      np = pytest.importorskip("numpy")
  
      u = da.from_array(np.arange(3))
      v = da.from_array(np.array([10 + 2j, 7 - 3j, 8 + 1j]))
      cv = v.conj()
      x = u * cv
      (cv,) = dask.optimize(cv)
      y = u * cv
      expected = np.array([0 + 0j, 7 + 3j, 16 - 2j])
      with dask.config.set({"optimization.fuse.active": False}):
          x_value = x.compute()
          y_value = y.compute()
      np.testing.assert_equal(x_value, expected)
      np.testing.assert_equal(y_value, expected)
  
imports: from __future__ import annotations
  import pytest
  import asyncio
  import os
  import sys
  from functools import partial
  from operator import add
  from distributed import Client, SchedulerPlugin, WorkerPlugin
  from distributed.utils_test import cleanup
  from distributed.utils_test import client
  from distributed.utils_test import cluster, cluster_fixture, gen_cluster, loop, loop_in_thread, popen, varying
  import dask
  import dask.bag
  from dask import compute, delayed, persist
  from dask.base import compute_as_if_collection, get_scheduler
  from dask.blockwise import Blockwise
  from dask.delayed import Delayed
  from dask.distributed import futures_of, wait
  from dask.layers import ShuffleLayer, SimpleShuffleLayer
  from dask.utils import get_named_args, get_scheduler_lock, tmpdir, tmpfile
  from dask.utils_test import inc
label_desc: Use image node_id 21 for calling a global function with example usage: test_blockwise_different_optimization(c) without return types