Repository: dask/dask
Pull request #11555 (2024-11-21 19:26:28)

Migrate rechunk and reshape to task spec
- [ ] Closes #xxxx
- [ ] Tests added / passed
- [ ] Passes `pre-commit run --all-files`


github-actions[bot]: ## Unit Test Results
_See [test report](https://dask.github.io/dask/test_report.html) for an extended history of previous test failures. This is useful for diagnosing flaky tests._

     13 files  ±0       13 suites  ±0   3h 27m 27s ⏱️ + 3m 2s
 13 256 tests ±0   12 194 ✅ ±0   1 062 💤 ±0  0 ❌ ±0 
138 328 runs  ±0  119 333 ✅  - 1  18 995 💤 +1  0 ❌ ±0 

Results for commit 759f91d3. ± Comparison against base commit 4814464b.

[test-results]:data:application/gzip;base64,H4sIAHaRP2cC/02NSw7DIAwFrxKx7gJjSHAvUyECEmo+FYFV1buX/EiWM88af5kPg1vYswF8NGzJIV3U52hSmKeVhZRdUWVN+y5Ue/JrydZuR0Dyku/wWSVvRXXehKE4XoWLcY6HiXna0xqFPrimgRCxyiOtidTpbumN72U7j2NIBVinyBP06LkURnJvhXceSWjgAOWDccoidIb9/qkXvgcaAQAA


Patch:
diff --git a/continuous_integration/environment-3.13.yaml b/continuous_integration/environment-3.13.yaml
new file mode 100644
index 000000000..9ac01b8ba
--- /dev/null
+++ b/continuous_integration/environment-3.13.yaml
@@ -0,0 +1,77 @@
+# This job includes coverage
+name: test-environment
+channels:
+  - conda-forge
+dependencies:
+  # required dependencies
+  - python=3.13
+  - packaging
+  - pyyaml
+  - click
+  - cloudpickle
+  - partd
+  - fsspec
+  - toolz
+  # test dependencies
+  - pre-commit
+  - pytest
+  - pytest-cov
+  - pytest-rerunfailures
+  - pytest-timeout
+  - pytest-xdist
+  - moto<5
+  # Optional dependencies
+  - mimesis
+  - numpy >=2  # only tested here
+  - pandas
+  #- numba  # not available for py 13
+  - flask
+  - fastparquet
+  - h5py
+  # Temporarily removing to allow `numpy >=2` to be installed
+  # - pytables
+  - zarr
+  # `tiledb-py=0.17.5` lead to strange seg faults in CI, However 0.18 is needed for 3.11
+  # https://github.com/dask/dask/pull/9569
+  # - tiledb-py # crashes on Python 3.11
+  # - pyspark
+  # - tiledb # crashes on Python 3.11
+  - xarray
+  - sqlalchemy
+  - pyarrow>=14.0.1
+  - coverage
+  - jsonschema
+  # other -- IO
+  - boto3
+  - botocore
+  - bokeh
+  - httpretty
+  - aiohttp
+  - s3fs>=2021.9.0
+  - crick
+  - cytoolz
+  - distributed
+  - ipython
+  - ipycytoscape
+  # until https://github.com/jupyter-widgets/ipywidgets/issues/3731 is fixed
+  - ipywidgets<8.0.5
+  - ipykernel<6.22.0
+  - lz4
+  - psutil
+  - requests
+  #- scikit-image  # not available for py 13
+  - scikit-learn
+  - scipy
+  - python-snappy
+  #- sparse  # not available for py 13
+  - cachey
+  - python-graphviz
+  - python-cityhash
+  - python-xxhash
+  #- mmh3  # only on pypi for py 13
+  - jinja2
+  - pip
+  - pip:
+    - git+https://github.com/dask/distributed
+    - mmh3
+    - git+https://github.com/dask/dask-expr
diff --git a/dask/_task_spec.py b/dask/_task_spec.py
index 82b896b61..d239f8393 100644
--- a/dask/_task_spec.py
+++ b/dask/_task_spec.py
@@ -693,6 +693,8 @@ class NestedContainer(Task):
         _dependencies: set | frozenset | None = None,
         **kwargs: Any,
     ):
+        if len(args) == 1 and isinstance(args[0], self.klass):
+            args = args[0]  # type: ignore
         super().__init__(
             None,
             self.to_container,
@@ -728,8 +730,6 @@ class Dict(NestedContainer):
             if len(args) > 1:
                 raise ValueError("Dict can only take one positional argument")
             kwargs = args[0]
-        elif not kwargs:
-            raise ValueError("Dict needs at least one argument")
         super().__init__(
             *(Tuple(*it) for it in kwargs.items()), _dependencies=_dependencies
         )
@@ -915,3 +915,22 @@ def fuse_linear_task_spec(dsk, keys):
                 # Having the same prefixes can result in the same key, i.e. getitem-hash -> getitem-hash
                 result[top_key] = Alias(top_key, target=renamed_key)
     return result
+
+
+def cull(
+    dsk: dict[KeyType, GraphNode], keys: Iterable[KeyType]
+) -> dict[KeyType, GraphNode]:
+    work = set(keys)
+    seen: set[KeyType] = set()
+    dsk2 = {}
+    wpop = work.pop
+    wupdate = work.update
+    sadd = seen.add
+    while work:
+        k = wpop()
+        if k in seen or k not in dsk:
+            continue
+        sadd(k)
+        dsk2[k] = v = dsk[k]
+        wupdate(v.dependencies)
+    return dsk2
diff --git a/dask/array/core.py b/dask/array/core.py
index 6a255c76d..34c559c9f 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -63,6 +63,7 @@ from dask.utils import (
     IndexCallable,
     SerializableLock,
     cached_cumsum,
+    cached_max,
     cached_property,
     concrete,
     derived_from,
@@ -82,6 +83,11 @@ from dask.utils import (
 )
 from dask.widgets import get_template
 
+try:
+    ARRAY_TEMPLATE = get_template("array.html.j2")
+except ImportError:
+    ARRAY_TEMPLATE = None
+
 T_IntOrNaN = Union[int, float]  # Should be Union[int, Literal[np.nan]]
 
 DEFAULT_GET = named_schedulers.get("threads", named_schedulers["sync"])
@@ -1516,7 +1522,7 @@ class Array(DaskMethodsMixin):
 
     @property
     def chunksize(self) -> tuple[T_IntOrNaN, ...]:
-        return tuple(max(c) for c in self.chunks)
+        return tuple(cached_max(c) for c in self.chunks)
 
     @property
     def dtype(self):
@@ -1641,7 +1647,7 @@ class Array(DaskMethodsMixin):
             nbytes = "unknown"
             cbytes = "unknown"
 
-        return get_template("array.html.j2").render(
+        return ARRAY_TEMPLATE.render(
             array=self,
             grid=grid,
             nbytes=nbytes,
diff --git a/dask/array/rechunk.py b/dask/array/rechunk.py
index 946265118..81fd680ef 100644
--- a/dask/array/rechunk.py
+++ b/dask/array/rechunk.py
@@ -20,6 +20,7 @@ import tlz as toolz
 from tlz import accumulate
 
 from dask import config
+from dask._task_spec import Alias, Task, TaskRef, parse_input
 from dask.array.chunk import getitem
 from dask.array.core import Array, concatenate3, normalize_chunks
 from dask.array.utils import validate_axis
@@ -722,18 +723,20 @@ def _compute_rechunk(x, chunks):
                 slc.start == 0 and slc.stop == x.chunks[i][ind]
                 for i, (slc, ind) in enumerate(zip(slices, old_index))
             ):
-                rec_cat_arg_flat[rec_cat_index] = old_blocks[old_block_index]
+                rec_cat_arg_flat[rec_cat_index] = TaskRef(old_blocks[old_block_index])
             else:
-                intermediates[name] = (getitem, old_blocks[old_block_index], slices)
-                rec_cat_arg_flat[rec_cat_index] = name
+                intermediates[name] = Task(
+                    name, getitem, TaskRef(old_blocks[old_block_index]), slices
+                )
+                rec_cat_arg_flat[rec_cat_index] = TaskRef(name)
 
         assert rec_cat_index == rec_cat_arg.size - 1
 
         # New block is formed by concatenation of sliced old blocks
         if all(d == 1 for d in rec_cat_arg.shape):
-            x2[key] = rec_cat_arg.flat[0]
+            x2[key] = Alias(key, rec_cat_arg.flat[0])
         else:
-            x2[key] = (concatenate3, rec_cat_arg.tolist())
+            x2[key] = Task(key, concatenate3, parse_input(rec_cat_arg.tolist()))
 
     del old_blocks, new_index
 
diff --git a/dask/array/reshape.py b/dask/array/reshape.py
index b824110c7..bbe903391 100644
--- a/dask/array/reshape.py
+++ b/dask/array/reshape.py
@@ -7,6 +7,7 @@ from operator import mul
 
 import numpy as np
 
+from dask._task_spec import Task, TaskRef
 from dask.array.core import Array
 from dask.array.utils import meta_from_array
 from dask.base import tokenize
@@ -346,7 +347,8 @@ def reshape(x, shape, merge_chunks=True, limit=None):
 
     if x.npartitions == 1:
         key = next(flatten(x.__dask_keys__()))
-        dsk = {(name,) + (0,) * len(shape): (M.reshape, key, shape)}
+        new_key = (name,) + (0,) * len(shape)
+        dsk = {new_key: Task(new_key, M.reshape, TaskRef(key), shape)}
         chunks = tuple((d,) for d in shape)
         graph = HighLevelGraph.from_collections(name, dsk, dependencies=[x])
         return Array(graph, name, chunks, meta=meta)
@@ -364,7 +366,10 @@ def reshape(x, shape, merge_chunks=True, limit=None):
     in_keys = list(product([x2.name], *[range(len(c)) for c in inchunks]))
     out_keys = list(product([name], *[range(len(c)) for c in outchunks]))
     shapes = list(product(*outchunks))
-    dsk = {a: (M.reshape, b, shape) for a, b, shape in zip(out_keys, in_keys, shapes)}
+    dsk = {
+        a: Task(a, M.reshape, TaskRef(b), shape)
+        for a, b, shape in zip(out_keys, in_keys, shapes)
+    }
 
     graph = HighLevelGraph.from_collections(name, dsk, dependencies=[x2])
     return Array(graph, name, outchunks, meta=meta)
@@ -500,7 +505,12 @@ def reshape_blockwise(
 
         dsk = {
             (outname,)
-            + tuple(chunk_out): (_reshape_blockwise, (x.name,) + tuple(chunk_in), shape)
+            + tuple(chunk_out): Task(
+                (outname,) + tuple(chunk_out),
+                _reshape_blockwise,
+                TaskRef((x.name,) + tuple(chunk_in)),
+                shape,
+            )
             for chunk_in, chunk_out, shape in zip(
                 chunk_tuples, out_chunk_tuples, out_shapes
             )
@@ -532,7 +542,12 @@ def reshape_blockwise(
 
     dsk = {
         (outname,)
-        + tuple(chunk_out): (_reshape_blockwise, (x.name,) + tuple(chunk_in), shape)
+        + tuple(chunk_out): Task(
+            (outname,) + tuple(chunk_out),
+            _reshape_blockwise,
+            TaskRef((x.name,) + tuple(chunk_in)),
+            shape,
+        )
         for chunk_in, chunk_out, shape in zip(
             chunk_tuples, out_chunk_tuples, out_shapes
         )
diff --git a/dask/array/svg.py b/dask/array/svg.py
index 0a0d74cbf..29da53546 100644
--- a/dask/array/svg.py
+++ b/dask/array/svg.py
@@ -2,10 +2,14 @@ from __future__ import annotations
 
 import math
 import re
+from functools import lru_cache
 
 import numpy as np
 
+from dask.utils import cached_cumsum
 
+
+@lru_cache(maxsize=512)
 def svg(chunks, size=200, **kwargs):
     """Convert chunks from Dask Array into an SVG Image
 
@@ -253,7 +257,7 @@ def svg_1d(chunks, sizes=None, **kwargs):
 
 
 def grid_points(chunks, sizes):
-    cumchunks = [np.cumsum((0,) + c) for c in chunks]
+    cumchunks = [np.array(cached_cumsum(c, initial_zero=True)) for c in chunks]
     points = [x * size / x[-1] for x, size in zip(cumchunks, sizes)]
     return points
 
diff --git a/dask/bag/core.py b/dask/bag/core.py
index 8ee3efafe..4bd31a258 100644
--- a/dask/bag/core.py
+++ b/dask/bag/core.py
@@ -38,7 +38,15 @@ from tlz import (
 )
 
 from dask import config
-from dask._task_spec import GraphNode
+from dask._task_spec import (
+    GraphNode,
+    List,
+    Task,
+    _execute_subgraph,
+    convert_legacy_graph,
+    cull,
+    fuse_linear_task_spec,
+)
 from dask.bag import chunk
 from dask.bag.avro import to_avro
 from dask.base import (
@@ -50,17 +58,15 @@ from dask.base import (
 )
 from dask.blockwise import blockwise
 from dask.context import globalmethod
-from dask.core import flatten, get_dependencies, istask, quote, reverse_dict
+from dask.core import flatten, istask, quote
 from dask.delayed import Delayed, unpack_collections
 from dask.highlevelgraph import HighLevelGraph
-from dask.optimization import cull, fuse, inline
 from dask.sizeof import sizeof
 from dask.typing import Graph, NestedKeys, no_default
 from dask.utils import (
     apply,
     digit,
     ensure_bytes,
-    ensure_dict,
     ensure_unicode,
     funcname,
     get_default_shuffle_method,
@@ -96,17 +102,47 @@ def lazify_task(task, start=True):
     (<built-in function sum>, (<class 'map'>, <function inc at ...>, [1, 2, 3]))
     """
     if isinstance(task, GraphNode):
-        return task
-    if type(task) is list and len(task) < 50:
-        return [lazify_task(arg, False) for arg in task]
-    if not istask(task):
-        return task
-    head, tail = task[0], task[1:]
-    if not start and head in (list, reify):
-        task = task[1]
-        return lazify_task(*tail, start=False)
+        if isinstance(task, List) and len(task.args) < 50:
+            return List(*[lazify_task(arg, False) for arg in task.args])
+        if not isinstance(task, Task):
+            return task
+        if not start and task.func in (list, reify) and isinstance(task.args[0], Task):
+            assert len(task.args) == 1
+            task = task.args[0]
+        if task.func is _execute_subgraph:
+            subgraph = task.args[0]
+            outkey = task.args[1]
+            # If there is a reify at the output of the subgraph we don't want to act
+            final_task = lazify_task(subgraph[outkey], True)
+            subgraph = {
+                k: lazify_task(v, False) for k, v in subgraph.items() if k != outkey
+            }
+            subgraph[outkey] = final_task
+            return Task(
+                task.key,
+                _execute_subgraph,
+                subgraph,
+                outkey,
+                *task.args[2:],
+                **task.kwargs,
+            )
+        return Task(
+            task.key,
+            task.func,
+            *[lazify_task(arg, False) for arg in task.args],
+            **task.kwargs,
+        )
     else:
-        return (head,) + tuple(lazify_task(arg, False) for arg in tail)
+        if type(task) is list and len(task) < 50:
+            return [lazify_task(arg, False) for arg in task]
+        if not istask(task):
+            return task
+        head, tail = task[0], task[1:]
+        if not start and head in (list, reify):
+            task = task[1]
+            return lazify_task(*tail, start=False)
+        else:
+            return (head,) + tuple(lazify_task(arg, False) for arg in tail)
 
 
 def lazify(dsk):
@@ -120,47 +156,14 @@ def lazify(dsk):
     return valmap(lazify_task, dsk)
 
 
-def inline_singleton_lists(dsk, keys, dependencies=None):
-    """Inline lists that are only used once.
-
-    >>> d = {'b': (list, 'a'),
-    ...      'c': (sum, 'b', 1)}
-    >>> inline_singleton_lists(d, 'c')
-    {'c': (<built-in function sum>, (<class 'list'>, 'a'), 1)}
-
-    Pairs nicely with lazify afterwards.
-    """
-    if dependencies is None:
-        dependencies = {k: get_dependencies(dsk, task=v) for k, v in dsk.items()}
-    dependents = reverse_dict(dependencies)
-
-    inline_keys = {
-        k
-        for k, v in dsk.items()
-        if istask(v)
-        and not isinstance(v, GraphNode)
-        and v
-        and v[0] is list
-        and len(dependents[k]) == 1
-    }
-    inline_keys.difference_update(flatten(keys))
-    dsk = inline(dsk, inline_keys, inline_constants=False)
-    for k in inline_keys:
-        del dsk[k]
-    return dsk
-
-
-def optimize(dsk, keys, fuse_keys=None, rename_fused_keys=None, **kwargs):
+def optimize(dsk, keys, fuse_keys=None, **kwargs):
     """Optimize a dask from a dask Bag."""
-    dsk = ensure_dict(dsk)
-    dsk2, dependencies = cull(dsk, keys)
-    kwargs = {}
-    if rename_fused_keys is not None:
-        kwargs["rename_keys"] = rename_fused_keys
-    dsk3, dependencies = fuse(dsk2, keys + (fuse_keys or []), dependencies, **kwargs)
-    dsk4 = inline_singleton_lists(dsk3, keys, dependencies)
-    dsk5 = lazify(dsk4)
-    return dsk5
+    dsk = convert_legacy_graph(dsk)
+    keys = list(flatten(keys))
+    dsk2 = cull(dsk, keys)
+    dsk3 = fuse_linear_task_spec(dsk2, keys + (fuse_keys or []))
+    dsk4 = lazify(dsk3)
+    return dsk4
 
 
 def _to_textfiles_chunk(data, lazy_file, last_endline):
diff --git a/dask/utils.py b/dask/utils.py
index 00e4f09ca..3f42bd6c0 100644
--- a/dask/utils.py
+++ b/dask/utils.py
@@ -2102,6 +2102,37 @@ def _cumsum(seq, initial_zero):
         return tuple(toolz.accumulate(add, seq))
 
 
+@functools.lru_cache
+def _max(seq):
+    if isinstance(seq, _HashIdWrapper):
+        seq = seq.wrapped
+    return max(seq)
+
+
+def cached_max(seq):
+    """Compute max with caching.
+
+    Caching is by the identity of `seq` rather than the value. It is thus
+    important that `seq` is a tuple of immutable objects, and this function
+    is intended for use where `seq` is a value that will persist (generally
+    block sizes).
+
+    Parameters
+    ----------
+    seq : tuple
+        Values to reduce
+
+    Returns
+    -------
+    tuple
+    """
+    assert isinstance(seq, tuple)
+    # Look up by identity first, to avoid a linear-time __hash__
+    # if we've seen this tuple object before.
+    result = _max(_HashIdWrapper(seq))
+    return result
+
+
 def cached_cumsum(seq, initial_zero=False):
     """Compute :meth:`toolz.accumulate` with caching.
 
diff --git a/docs/source/_static/custom.js b/docs/source/_static/custom.js
new file mode 100644
index 000000000..8a81d6cad
--- /dev/null
+++ b/docs/source/_static/custom.js
@@ -0,0 +1,26 @@
+document.addEventListener("DOMContentLoaded", function () {
+  var script = document.createElement("script");
+  script.type = "module";
+  script.id = "runllm-widget-script"
+
+  script.src = "https://widget.runllm.com";
+
+  script.setAttribute("version", "stable");
+  script.setAttribute("runllm-keyboard-shortcut", "Mod+j"); // cmd-j or ctrl-j to open the widget.
+  script.setAttribute("runllm-name", "DaskBot");
+  script.setAttribute("runllm-position", "BOTTOM_RIGHT"); // put above ethical ads
+  script.setAttribute("runllm-position-x", "20px");
+  script.setAttribute("runllm-position-y", "50%");
+  script.setAttribute("runllm-assistant-id", "273");
+  script.setAttribute("runllm-theme-color", "#FFC11E");
+  script.setAttribute("runllm-slack-community-url", "https://dask.slack.com/");
+  script.setAttribute("runllm-per-user-usage-limit", 2);
+  script.setAttribute("runllm-usage-limit-effective-days", 30);
+  script.setAttribute("runllm-usage-limit-message", `Hi! You've hit the limit for anonymous questions, but you can join us on Slack in #DaskBot and ask as many questions as you'd like.`);
+  script.setAttribute("runllm-brand-logo", "_images/dask_icon.svg");
+  script.setAttribute("runllm-floating-button-text", "Ask DaskBot");
+  script.setAttribute("runllm-join-community-text", "Chat with DaskBot in Slack");
+
+  script.async = true;
+  document.head.appendChild(script);
+});
diff --git a/docs/source/conf.py b/docs/source/conf.py
index 1badbb40c..737cfae6a 100644
--- a/docs/source/conf.py
+++ b/docs/source/conf.py
@@ -154,6 +154,8 @@ html_theme_options = {"logo_only": True}
 # so a file named "default.css" will overwrite the builtin "default.css".
 html_static_path = ["_static"]
 
+html_js_files = ["custom.js"]
+
 # If not '', a 'Last updated on:' timestamp is inserted at every page bottom,
 # using the given strftime format.
 # html_last_updated_fmt = '%b %d, %Y'
diff --git a/pyproject.toml b/pyproject.toml
index 04bce8699..1d3f70f69 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -20,6 +20,7 @@ classifiers = [
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
+    "Programming Language :: Python :: 3.13",
     "Topic :: Scientific/Engineering",
     "Topic :: System :: Distributed Computing",
 ]


Test patch:
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
index fcf9b7f8c..5ec24560d 100644
--- a/.github/workflows/tests.yml
+++ b/.github/workflows/tests.yml
@@ -26,13 +26,15 @@ jobs:
       fail-fast: false
       matrix:
         os: ["windows-latest", "ubuntu-latest", "macos-14"]
-        environment: ["3.10", "3.11", "3.12"]
+        environment: ["3.10", "3.11", "3.12", "3.13"]
         extra: [null]
         exclude:
           - os: "macos-14"
             environment: "3.10"
           - os: "macos-14"
             environment: "3.11"
+          - os: "macos-14"
+            environment: "3.13"
         include:
           # Minimum dependencies
           - os: "ubuntu-latest"
diff --git a/dask/array/tests/test_rechunk.py b/dask/array/tests/test_rechunk.py
index ee5774a4f..2e3e5e41c 100644
--- a/dask/array/tests/test_rechunk.py
+++ b/dask/array/tests/test_rechunk.py
@@ -5,6 +5,8 @@ from itertools import product
 
 import pytest
 
+from dask._task_spec import Task
+
 np = pytest.importorskip("numpy")
 import math
 
@@ -545,9 +547,9 @@ def test_dont_concatenate_single_chunks(shape, chunks):
     y = x.rechunk(chunks)
     dsk = dict(y.dask)
     assert not any(
-        funcname(task[0]).startswith("concat")
+        funcname(task.func).startswith("concat")
         for task in dsk.values()
-        if dask.istask(task)
+        if isinstance(task, Task)
     )
 
 
diff --git a/dask/bag/tests/test_bag.py b/dask/bag/tests/test_bag.py
index fe1b5ffc9..20f43b89f 100644
--- a/dask/bag/tests/test_bag.py
+++ b/dask/bag/tests/test_bag.py
@@ -19,14 +19,13 @@ from tlz import groupby, identity, join, merge, pluck, unique, valmap
 
 import dask
 import dask.bag as db
+from dask._task_spec import List, Task
 from dask.bag.core import (
     Bag,
     collect,
     from_delayed,
-    inline_singleton_lists,
     lazify,
     lazify_task,
-    optimize,
     partition,
     reduceby,
     reify,
@@ -574,6 +573,47 @@ def test_random_sample_random_state():
 
 
 def test_lazify_task():
+    task = Task(None, sum, Task(None, reify, Task(None, map, inc, [1, 2, 3])))
+    assert lazify_task(task) == Task(None, sum, Task(None, map, inc, [1, 2, 3]))
+
+    task = Task(None, reify, Task(None, map, inc, [1, 2, 3]))
+    assert lazify_task(task) == task
+
+    task = Task(
+        None,
+        reify,
+        Task(None, map, inc, Task(None, reify, Task(None, filter, iseven, "y"))),
+    )
+    expected = Task(None, reify, Task(None, map, inc, Task(None, filter, iseven, "y")))
+    assert lazify_task(task) == expected
+
+    task = Task(
+        None,
+        sum,
+        List(
+            Task(
+                None,
+                reify,
+                Task(None, map, inc, Task(None, reify, Task(None, filter, iseven, 1))),
+            ),
+            Task(
+                None,
+                reify,
+                Task(None, map, inc, Task(None, reify, Task(None, filter, iseven, 2))),
+            ),
+        ),
+    )
+    assert lazify_task(task) == Task(
+        None,
+        sum,
+        List(
+            Task(None, map, inc, Task(None, filter, iseven, 1)),
+            Task(None, map, inc, Task(None, filter, iseven, 2)),
+        ),
+    )
+
+
+def test_lazify_task_legacy():
     task = (sum, (reify, (map, inc, [1, 2, 3])))
     assert lazify_task(task) == (sum, (map, inc, [1, 2, 3]))
 
@@ -598,40 +638,6 @@ def test_lazify():
     assert lazify(a) == b
 
 
-def test_inline_singleton_lists():
-    inp = {"b": (list, "a"), "c": (f, "b", 1)}
-    out = {"c": (f, (list, "a"), 1)}
-    assert inline_singleton_lists(inp, ["c"]) == out
-
-    out = {"c": (f, "a", 1)}
-    assert optimize(inp, ["c"], rename_fused_keys=False) == out
-
-    # If list is an output key, don't fuse it
-    assert inline_singleton_lists(inp, ["b", "c"]) == inp
-    assert optimize(inp, ["b", "c"], rename_fused_keys=False) == inp
-
-    inp = {"b": (list, "a"), "c": (f, "b", 1), "d": (f, "b", 2)}
-    assert inline_singleton_lists(inp, ["c", "d"]) == inp
-
-    # Doesn't inline constants
-    inp = {"b": (4, 5), "c": (f, "b")}
-    assert inline_singleton_lists(inp, ["c"]) == inp
-
-
-def test_rename_fused_keys_bag():
-    inp = {"b": (list, "a"), "c": (f, "b", 1)}
-
-    outp = optimize(inp, ["c"], rename_fused_keys=False)
-    assert outp.keys() == {"c"}
-    assert outp["c"][1:] == ("a", 1)
-
-    with dask.config.set({"optimization.fuse.rename-keys": False}):
-        assert optimize(inp, ["c"]) == outp
-
-    # By default, fused keys are renamed
-    assert optimize(inp, ["c"]) != outp
-
-
 def test_take():
     assert list(b.take(2)) == [0, 1]
     assert b.take(2) == (0, 1)
@@ -1156,32 +1162,55 @@ def test_to_delayed():
     assert t.compute() == 21
 
 
-def test_to_delayed_optimize_graph(tmpdir):
+def test_to_delayed_optimize_graph(tmpdir, monkeypatch):
+    calls = 0
+    from dask.bag.core import reify
+
+    def count_reify(*args, **kwargs):
+        nonlocal calls
+        calls += 1
+        return reify(*args, **kwargs)
+
+    monkeypatch.setattr(dask.bag.core, "reify", count_reify)
     b = db.from_sequence([1, 2, 3, 4, 5, 6], npartitions=1)
     b2 = b.map(inc).map(inc).map(inc)
 
     [d] = b2.to_delayed()
-    text = str(dict(d.dask))
-    assert text.count("reify") == 1
+
+    calls = 0
+    d.compute()
+    assert calls == 1
+
+    calls = 0
+    b = db.from_sequence([1, 2, 3, 4, 5, 6], npartitions=1)
+    b2 = b.map(inc).map(inc).map(inc)
+    [d] = b2.to_delayed()
     assert d.__dask_layers__() != b2.__dask_layers__()
     [d2] = b2.to_delayed(optimize_graph=False)
     assert dict(d2.dask) == dict(b2.dask)
     assert d2.__dask_layers__() == b2.__dask_layers__()
-    assert d.compute() == d2.compute()
+    calls = 0
+    result = d.compute()
+    assert calls == 1
+    calls = 0
+    result2 = d2.compute()
+    assert calls == 3
+    assert result == result2
 
     x = b2.sum()
     d = x.to_delayed()
-    text = str(dict(d.dask))
     assert d.__dask_layers__() == x.__dask_layers__()
-    assert text.count("reify") == 0
     d2 = x.to_delayed(optimize_graph=False)
     assert dict(d2.dask) == dict(x.dask)
     assert d2.__dask_layers__() == x.__dask_layers__()
+
+    calls = 0
     assert d.compute() == d2.compute()
+    assert calls == 3
 
+    calls = 0
     [d] = b2.to_textfiles(str(tmpdir), compute=False)
-    text = str(dict(d.dask))
-    assert text.count("reify") <= 0
+    assert calls == 0
 
 
 def test_from_delayed():
diff --git a/dask/tests/test_base.py b/dask/tests/test_base.py
index f03a30042..752288805 100644
--- a/dask/tests/test_base.py
+++ b/dask/tests/test_base.py
@@ -371,7 +371,7 @@ def test_compute_no_opt():
     # Otherwise, the lengths below would be 4 and 0.
     assert len([k for k in keys if "mul" in k[0]]) == 8
     assert len([k for k in keys if "add" in k[0]]) == 4
-    assert len([k for k in keys if "add-mul" in k[0]]) == 4  # See? Renamed
+    assert len([k for k in keys if "add-from_sequence-mul" in k[0]]) == 4
 
 
 @pytest.mark.skipif("not da")
diff --git a/dask/tests/test_task_spec.py b/dask/tests/test_task_spec.py
index 0b41e673b..1b6cecd5a 100644
--- a/dask/tests/test_task_spec.py
+++ b/dask/tests/test_task_spec.py
@@ -1033,3 +1033,8 @@ def test_nested_containers_different_types(task_type, python_type):
     t = task_type(Task("key-1", func, "a", TaskRef("b")), Task("key-2", func, "c", "d"))
     assert t.dependencies == {"b"}
     assert t({"b": "b"}) == python_type(("a-b", "c-d"))
+
+
+@pytest.mark.parametrize("Container", [Dict, List, Set, Tuple])
+def test_nested_containers_empty(Container):
+    assert Container(Container.klass())() == Container.klass()
diff --git a/dask/tests/test_tokenize.py b/dask/tests/test_tokenize.py
index b4435fbb0..a2a5f625d 100644
--- a/dask/tests/test_tokenize.py
+++ b/dask/tests/test_tokenize.py
@@ -15,9 +15,11 @@ from typing import Union
 
 import cloudpickle
 import pytest
+from packaging.version import Version
 from tlz import compose, curry, partial
 
 import dask
+from dask._compatibility import PY_VERSION
 from dask.core import flatten, literal
 from dask.tokenize import TokenizationError, normalize_token, tokenize
 from dask.utils import tmpfile
@@ -954,6 +956,9 @@ def test_local_objects():
     assert check_tokenize(LocalDaskTokenize()) != check_tokenize(LocalReducible())
 
 
+@pytest.mark.skipif(
+    PY_VERSION >= Version("3.13"), reason="https://github.com/dask/dask/issues/11457"
+)
 def test_tokenize_dataclass():
     a1 = ADataClass(1)
     a2 = ADataClass(2)