code: def test_optimize_blockwise_control_annotations():
      """
      Can we fuse blockwise layers with different, but compatible
      annotations for retries, priority, etc.
      """
  
      a = da.ones(10, chunks=(5,))
      b = a + 1
  
      with dask.annotate(
          retries=5, workers=["a", "b", "c"], allow_other_workers=False
      ):
          c = b + 2
  
      with dask.annotate(
          priority=2, workers=["b", "c", "d"], allow_other_workers=True
      ):
          d = c + 3
  
      with dask.annotate(retries=3, resources={"GPU": 2, "Memory": 10}):
          e = d + 4
  
      with dask.annotate(priority=4, resources={"GPU": 5, "Memory": 4}):
          f = e + 5
  
      # This one will not be fused due to the custom annotation, nor will the one below
      with dask.annotate(foo="bar"):
          g = f + 6
  
      h = g + 6
  
      dsk = da.optimization.optimize_blockwise(h.dask)
  
      # The layers and their annotations should be fusable until the custom one
      assert len(dsk.layers) == 3
      layer = hlg_layer_topological(
          dsk, 0
      )  # First layer is the fused one
      annotations = layer.annotations
  
      assert len(annotations) == 5
      assert annotations["priority"] == 4  # max
      assert annotations["retries"] == 5  # max
      assert (
          annotations["allow_other_workers"] is False
      )  # More restrictive
      assert set(annotations["workers"]) == {"b", "c"}  # intersection
      assert annotations["resources"] == {
          "GPU": 5,
          "Memory": 10,
      }  # Max of resources
  
      # If we disable blockwise annotation fusion, we can only fuse the first two layers.
      with dask.config.set({"optimization.annotations.fuse": False}):
          dsk = da.optimization.optimize_blockwise(h.dask)
          assert len(dsk.layers) == 7
  
imports: from __future__ import annotations
  import collections
  from operator import add
  import numpy
  import pytest
  import dask
  import dask.array
  from dask.array.utils import assert_eq
  from dask.blockwise import _BLOCKWISE_DEFAULT_PREFIX, Blockwise, _unique_dep, index_subs, optimize_blockwise, rewrite_blockwise
  from dask.highlevelgraph import HighLevelGraph
  from dask.utils_test import dec, hlg_layer_topological, inc
label_desc: Use image node_id 4 for calling a global function with example usage: test_optimize_blockwise_control_annotations() without return types