Repository: dask/dask

Issue:
Slicing an Array with full slices along every dimension should just map the keys
```
ds = xr.open_zarr(
    "gs://weatherbench2/datasets/era5/1959-2023_01_10-wb13-6h-1440x721.zarr",
    #consolidated=True
)
ds = ds[variables].sel(time=slice("2020-01-01", "2022-12-31"))
```

The dataset has one point per chunk along the time axis, so every slice is trivial

<img width="634" alt="Screenshot 2024-10-18 at 17 09 25" src="https://github.com/user-attachments/assets/2c005e78-d97b-4e67-9936-c63ca54a7b66">

this creates a lot of tasks that look like this:

```
 ('getitem-cbd92e9090b2dd038943d6de9184aefe',
  1,
  0,
  0): (<function dask.array.chunk.getitem(obj, index)>, ('open_dataset-sea_surface_temperature-c20638560e986ac27140dafa0e57ac52',
   89121,
   0,
   0), (slice(None, None, None),
   slice(None, None, None),
   slice(None, None, None))),
```

We should optimise this away to something like:

```
 ('getitem-cbd92e9090b2dd038943d6de9184aefe',
  1,
  0,
  0): ('open_dataset-sea_surface_temperature-c20638560e986ac27140dafa0e57ac52',
   89121,
   0,
   0),
```

Discussion:

Patch:
diff --git a/dask/_task_spec.py b/dask/_task_spec.py
index 3b563aced..a319ac865 100644
--- a/dask/_task_spec.py
+++ b/dask/_task_spec.py
@@ -84,7 +84,7 @@ from typing import Any, TypeVar, cast, overload
 
 from dask.sizeof import sizeof
 from dask.typing import Key as KeyType
-from dask.utils import is_namedtuple_instance
+from dask.utils import funcname, is_namedtuple_instance
 
 try:
     from distributed.collections import LRU
@@ -483,7 +483,7 @@ class Alias(GraphNode):
             return self
 
     def __repr__(self):
-        return f"Alias(key={self.key}, target={self.target})"
+        return f"Alias({self.key}->{self.target})"
 
     def __eq__(self, value: object) -> bool:
         if not isinstance(value, Alias):
@@ -587,6 +587,7 @@ class Task(GraphNode):
     packed_func: None | bytes
     _token: str | None
     _is_coro: bool | None
+    _repr: str | None
 
     __slots__ = tuple(__annotations__)
 
@@ -612,6 +613,7 @@ class Task(GraphNode):
             self._dependencies = _no_deps
         self._is_coro = None
         self._token = None
+        self._repr = None
 
     def copy(self):
         self.unpack()
@@ -651,7 +653,38 @@ class Task(GraphNode):
         )
 
     def __repr__(self) -> str:
-        return f"Task({self.key!r})"
+        # When `Task` is deserialized the constructor will not run and
+        # `self._repr` is thus undefined.
+        if not hasattr(self, "_repr") or not self._repr:
+            self.unpack()
+            head = funcname(self.func)
+            tail = ")"
+            label_size = 40
+            args = self.args
+            kwargs = self.kwargs
+            if args or kwargs:
+                label_size2 = int(
+                    (label_size - len(head) - len(tail) - len(str(self.key)))
+                    // (len(args) + len(kwargs))
+                )
+            if args:
+                if label_size2 > 5:
+                    args_repr = ", ".join(repr(t) for t in args)
+                else:
+                    args_repr = "..."
+            else:
+                args_repr = ""
+            if kwargs:
+                if label_size2 > 5:
+                    kwargs_repr = ", " + ", ".join(
+                        f"{k}={repr(v)}" for k, v in sorted(kwargs.items())
+                    )
+                else:
+                    kwargs_repr = ", ..."
+            else:
+                kwargs_repr = ""
+            self._repr = f"<Task {self.key!r} {head}({args_repr}{kwargs_repr}{tail}>"
+        return self._repr
 
     def __call__(self, values=()):
         self._verify_values(values)
@@ -871,19 +904,33 @@ def no_function_cache():
         _func_cache, _func_cache_reverse = cache_before
 
 
-def execute_graph(dsk: list[GraphNode] | dict[KeyType, GraphNode]) -> dict:
+def execute_graph(
+    dsk: Iterable[GraphNode] | Mapping[KeyType, GraphNode],
+    cache: MutableMapping[KeyType, object] | None = None,
+    keys: Container[KeyType] | None = None,
+) -> MutableMapping[KeyType, object]:
+    """Execute a given graph.
+
+    The graph is exceuted in topological order as defined by dask.order until
+    all leaf nodes, i.e. nodes without any dependents, are reached. The returned
+    dictionary contains the results of the leaf nodes.
+
+    If keys are required that are not part of the graph, they can be provided in the `cache` argument.
+
+    If `keys` is provided, the result will contain only values that are part of the `keys` set.
+
+    """
     if isinstance(dsk, (list, tuple, set, frozenset)):
         dsk = {t.key: t for t in dsk}
     else:
         assert isinstance(dsk, dict)
-    dependencies = dict(DependenciesMapping(dsk))
 
     refcount: defaultdict[KeyType, int] = defaultdict(int)
-    for vals in dependencies.values():
+    for vals in DependenciesMapping(dsk).values():
         for val in vals:
             refcount[val] += 1
 
-    cache: dict[KeyType, object] = {}
+    cache = cache or {}
     from dask.order import order
 
     priorities = order(dsk)
@@ -892,7 +939,7 @@ def execute_graph(dsk: list[GraphNode] | dict[KeyType, GraphNode]) -> dict:
         cache[key] = node(cache)
         for dep in node.dependencies:
             refcount[dep] -= 1
-            if refcount[dep] == 0:
+            if refcount[dep] == 0 and keys and dep not in keys:
                 del cache[dep]
 
     return cache
diff --git a/dask/array/slicing.py b/dask/array/slicing.py
index c18e52939..ec12479cf 100644
--- a/dask/array/slicing.py
+++ b/dask/array/slicing.py
@@ -195,7 +195,9 @@ def slice_with_newaxes(out_name, in_name, blockdims, index):
             where_none[i] -= n
 
     # Pass down and do work
-    dsk, blockdims2 = slice_wrap_lists(out_name, in_name, blockdims, index2)
+    dsk, blockdims2 = slice_wrap_lists(
+        out_name, in_name, blockdims, index2, not where_none
+    )
 
     if where_none:
         expand = expander(where_none)
@@ -220,7 +222,7 @@ def slice_with_newaxes(out_name, in_name, blockdims, index):
         return dsk, blockdims2
 
 
-def slice_wrap_lists(out_name, in_name, blockdims, index):
+def slice_wrap_lists(out_name, in_name, blockdims, index, allow_getitem_optimization):
     """
     Fancy indexing along blocked array dasks
 
@@ -251,7 +253,9 @@ def slice_wrap_lists(out_name, in_name, blockdims, index):
 
     # No lists, hooray! just use slice_slices_and_integers
     if not where_list:
-        return slice_slices_and_integers(out_name, in_name, blockdims, index)
+        return slice_slices_and_integers(
+            out_name, in_name, blockdims, index, allow_getitem_optimization
+        )
 
     # Replace all lists with full slices  [3, 1, 0] -> slice(None, None, None)
     index_without_list = tuple(
@@ -269,7 +273,11 @@ def slice_wrap_lists(out_name, in_name, blockdims, index):
         # Do first pass without lists
         tmp = "slice-" + tokenize((out_name, in_name, blockdims, index))
         dsk, blockdims2 = slice_slices_and_integers(
-            tmp, in_name, blockdims, index_without_list
+            tmp,
+            in_name,
+            blockdims,
+            index_without_list,
+            allow_getitem_optimization=False,
         )
 
         # After collapsing some axes due to int indices, adjust axis parameter
@@ -285,7 +293,9 @@ def slice_wrap_lists(out_name, in_name, blockdims, index):
     return dsk3, blockdims2
 
 
-def slice_slices_and_integers(out_name, in_name, blockdims, index):
+def slice_slices_and_integers(
+    out_name, in_name, blockdims, index, allow_getitem_optimization=False
+):
     """
     Dask array indexing with slices and integers
 
@@ -329,7 +339,12 @@ def slice_slices_and_integers(out_name, in_name, blockdims, index):
     all_slices = list(product(*[pluck(1, s) for s in sorted_block_slices]))
 
     dsk_out = {
-        out_name: (getitem, in_name, slices)
+        out_name: (
+            (getitem, in_name, slices)
+            if not allow_getitem_optimization
+            or not all(sl == slice(None, None, None) for sl in slices)
+            else in_name
+        )
         for out_name, in_name, slices in zip(out_names, in_names, all_slices)
     }
 
diff --git a/dask/core.py b/dask/core.py
index afe799207..3cfd3c3a1 100644
--- a/dask/core.py
+++ b/dask/core.py
@@ -1,10 +1,10 @@
 from __future__ import annotations
 
 from collections import defaultdict
-from collections.abc import Collection, Iterable, Mapping
+from collections.abc import Collection, Iterable, Mapping, MutableMapping
 from typing import Any, Literal, TypeVar, cast, overload
 
-from dask._task_spec import DependenciesMapping
+from dask._task_spec import DependenciesMapping, convert_legacy_graph, execute_graph
 from dask.typing import Graph, Key, NoDefault, no_default
 
 
@@ -53,28 +53,6 @@ def istask(x):
     return type(x) is tuple and x and callable(x[0])
 
 
-def has_tasks(dsk, x):
-    """Whether ``x`` has anything to compute.
-
-    Returns True if:
-    - ``x`` is a task
-    - ``x`` is a key in ``dsk``
-    - ``x`` is a list that contains any tasks or keys
-    """
-    if istask(x):
-        return True
-    try:
-        if x in dsk:
-            return True
-    except Exception:
-        pass
-    if isinstance(x, list):
-        for i in x:
-            if has_tasks(dsk, i):
-                return True
-    return False
-
-
 def preorder_traversal(task):
     """A generator to preorder-traverse a task."""
 
@@ -94,53 +72,13 @@ def lists_to_tuples(res, keys):
     return res
 
 
-def _execute_task(arg, cache, dsk=None):
-    """Do the actual work of collecting data and executing a function
-
-    Examples
-    --------
-
-    >>> inc = lambda x: x + 1
-    >>> add = lambda x, y: x + y
-    >>> cache = {'x': 1, 'y': 2}
-
-    Compute tasks against a cache
-    >>> _execute_task((add, 'x', 1), cache)  # Compute task in naive manner
-    2
-    >>> _execute_task((add, (inc, 'x'), 1), cache)  # Support nested computation
-    3
-
-    Also grab data from cache
-    >>> _execute_task('x', cache)
-    1
-
-    Support nested lists
-    >>> list(_execute_task(['x', 'y'], cache))
-    [1, 2]
-
-    >>> list(map(list, _execute_task([['x', 'y'], ['y', 'x']], cache)))
-    [[1, 2], [2, 1]]
-
-    >>> _execute_task('foo', cache)  # Passes through on non-keys
-    'foo'
-    """
-    if isinstance(arg, list):
-        return [_execute_task(a, cache) for a in arg]
-    elif istask(arg):
-        func, args = arg[0], arg[1:]
-        # Note: Don't assign the subtask results to a variable. numpy detects
-        # temporaries by their reference count and can execute certain
-        # operations in-place.
-        return func(*(_execute_task(a, cache) for a in args))
-    elif not ishashable(arg):
-        return arg
-    elif arg in cache:
-        return cache[arg]
-    else:
-        return arg
+def _pack_result(result: Mapping, keys: list | Key) -> Any:
+    if isinstance(keys, list):
+        return tuple(_pack_result(result, k) for k in keys)
+    return result[keys]
 
 
-def get(dsk, out, cache=None):
+def get(dsk: Mapping, out: list | Key, cache: MutableMapping | None = None) -> Any:
     """Get value from Dask
 
     Examples
@@ -154,19 +92,15 @@ def get(dsk, out, cache=None):
     >>> get(d, 'y')
     2
     """
-    for k in flatten(out) if isinstance(out, list) else [out]:
+    for k in flatten(out):
         if k not in dsk:
             raise KeyError(f"{k} is not a key in the graph")
     if cache is None:
         cache = {}
-    for key in toposort(dsk):
-        task = dsk[key]
-        result = _execute_task(task, cache)
-        cache[key] = result
-    result = _execute_task(out, cache)
-    if isinstance(out, list):
-        result = lists_to_tuples(result, out)
-    return result
+
+    dsk2 = convert_legacy_graph(dsk, all_keys=set(dsk) | set(cache))
+    result = execute_graph(dsk2, cache, keys=set(flatten([out])))
+    return _pack_result(result, out)
 
 
 def keys_in_tasks(keys: Collection[Key], tasks: Iterable[Any], as_list: bool = False):
diff --git a/dask/dask-schema.yaml b/dask/dask-schema.yaml
index 4be821187..267bbeccf 100644
--- a/dask/dask-schema.yaml
+++ b/dask/dask-schema.yaml
@@ -244,6 +244,13 @@ properties:
               False, then the top-most key will be used. For advanced usage, a function
               to create the new name is also accepted.
 
+          delayed:
+            type: boolean
+            description: |
+              Whether to fuse adjacent delayed calls together. This can be useful to ensure 
+              that an independent chain of delayed calls is executed in a single task to
+              reduce the memory footprint on a single worker.
+
   admin:
     type: object
     properties:
diff --git a/dask/dask.yaml b/dask/dask.yaml
index 5ba27ce89..8da587e7a 100644
--- a/dask/dask.yaml
+++ b/dask/dask.yaml
@@ -42,6 +42,7 @@ optimization:
     max-depth-new-edges: null  # ave_width * 1.5
     subgraphs: null  # true for dask.dataframe, false for everything else
     rename-keys: true
+    delayed: false
 
 admin:
   traceback:
@@ -50,6 +51,7 @@ admin:
       - dask[\\\/](base|core|local|multiprocessing|optimization|threaded|utils)\.py
       - dask[\\\/]array[\\\/]core\.py
       - dask[\\\/]dataframe[\\\/](core|methods)\.py
+      - dask[\\\/]_task_spec\.py
       - distributed[\\\/](client|scheduler|utils|worker)\.py
       - tornado[\\\/]gen\.py
       - pandas[\\\/]core[\\\/]
diff --git a/dask/dataframe/__init__.py b/dask/dataframe/__init__.py
index ca9bf46a8..a64a2fb93 100644
--- a/dask/dataframe/__init__.py
+++ b/dask/dataframe/__init__.py
@@ -28,6 +28,13 @@ def _dask_expr_enabled() -> bool:
         return DASK_EXPR_ENABLED
 
     if use_dask_expr is False:
+        warnings.warn(
+            "The legacy Dask DataFrame implementation is deprecated and will "
+            "be removed in a future version. Set the configuration option "
+            "`dataframe.query-planning` to `True` or None to enable the new Dask "
+            "Dataframe implementation and silence this warning.",
+            FutureWarning,
+        )
         return (DASK_EXPR_ENABLED := False)
     try:
         import dask_expr  # noqa: F401
diff --git a/dask/dataframe/groupby.py b/dask/dataframe/groupby.py
index a1350694f..545c59ecd 100644
--- a/dask/dataframe/groupby.py
+++ b/dask/dataframe/groupby.py
@@ -322,13 +322,14 @@ class Aggregation:
         result will be identified by this name.
     chunk : callable
         a function that will be called with the grouped column of each
-        partition. It can either return a single series or a tuple of series.
+        partition, takes a Pandas SeriesGroupBy in input.
+        It can either return a single series or a tuple of series.
         The index has to be equal to the groups.
     agg : callable
         a function that will be called to aggregate the results of each chunk.
-        Again the argument(s) will be grouped series. If ``chunk`` returned a
-        tuple, ``agg`` will be called with all of them as individual positional
-        arguments.
+        Again the argument(s) will be a Pandas SeriesGroupBy. If ``chunk``
+        returned a tuple, ``agg`` will be called with all of them as
+        individual positional arguments.
     finalize : callable
         an optional finalizer that will be called with the results from the
         aggregation.
@@ -1227,10 +1228,23 @@ def _apply_func_to_columns(df_like, prefix, func):
 
 
 def _finalize_mean(df, sum_column, count_column):
-    return df[sum_column] / df[count_column]
+    result = df[sum_column] / df[count_column]
+    return _adjust_for_arrow_na(result, df[count_column])
 
 
-def _finalize_var(df, count_column, sum_column, sum2_column, **kwargs):
+def _adjust_for_arrow_na(result, df, check_for_isna=False):
+    if isinstance(result.dtype, pd.ArrowDtype):
+        # Our mean computation results in np.nan here but pandas doesn't
+        if check_for_isna:
+            result[df.isna()] = pd.NA
+        else:
+            result[df == 0] = pd.NA
+    return result
+
+
+def _finalize_var(
+    df, count_column, sum_column, sum2_column, adjust_arrow=True, **kwargs
+):
     # arguments are being checked when building the finalizer. As of this moment,
     # we're only using ddof, and raising an error on other keyword args.
     ddof = kwargs.get("ddof", 1)
@@ -1243,13 +1257,20 @@ def _finalize_var(df, count_column, sum_column, sum2_column, **kwargs):
     div[div < 0] = 0
     result /= div
     result[(n - ddof) == 0] = np.nan
-
-    return result
+    if adjust_arrow:
+        return _adjust_for_arrow_na(result, div)
+    else:
+        return result
 
 
 def _finalize_std(df, count_column, sum_column, sum2_column, **kwargs):
-    result = _finalize_var(df, count_column, sum_column, sum2_column, **kwargs)
-    return np.sqrt(result)
+    result = _finalize_var(
+        df, count_column, sum_column, sum2_column, adjust_arrow=False, **kwargs
+    )
+    res = np.sqrt(result)
+    if res.dtype != result.dtype:
+        res = res.astype(result.dtype)
+    return _adjust_for_arrow_na(res, result, check_for_isna=True)
 
 
 def _cum_agg_aligned(part, cum_last, index, columns, func, initial):
diff --git a/dask/delayed.py b/dask/delayed.py
index 1ff6a2dba..e43fab1da 100644
--- a/dask/delayed.py
+++ b/dask/delayed.py
@@ -22,10 +22,12 @@ from dask.base import tokenize as _tokenize
 from dask.context import globalmethod
 from dask.core import flatten, quote
 from dask.highlevelgraph import HighLevelGraph
+from dask.optimization import fuse
 from dask.typing import Graph, NestedKeys
 from dask.utils import (
     OperatorMethodMixin,
     apply,
+    ensure_dict,
     funcname,
     is_namedtuple_instance,
     methodcaller,
@@ -522,9 +524,29 @@ def right(method):
     return partial(_swap, method)
 
 
+def _fuse_delayed(dsk, keys, **kwargs):
+    dependencies = dsk.get_all_dependencies()
+    dsk = ensure_dict(dsk)
+
+    fuse_subgraphs = config.get("optimization.fuse.subgraphs")
+    if fuse_subgraphs is None:
+        fuse_subgraphs = True
+    dsk, _ = fuse(
+        dsk,
+        keys,
+        dependencies=dependencies,
+        fuse_subgraphs=fuse_subgraphs,
+    )
+    return dsk
+
+
 def optimize(dsk, keys, **kwargs):
     if not isinstance(keys, (list, set)):
         keys = [keys]
+
+    if config.get("optimization.fuse.delayed"):
+        dsk = _fuse_delayed(dsk, keys, **kwargs)
+
     if not isinstance(dsk, HighLevelGraph):
         dsk = HighLevelGraph.from_collections(id(dsk), dsk, dependencies=())
     dsk = dsk.cull(set(flatten(keys)))
diff --git a/dask/diagnostics/profile_visualize.py b/dask/diagnostics/profile_visualize.py
index 32d3f37cc..98c56e25e 100644
--- a/dask/diagnostics/profile_visualize.py
+++ b/dask/diagnostics/profile_visualize.py
@@ -10,7 +10,7 @@ from tlz import accumulate, groupby, pluck, unique
 
 from dask._compatibility import import_optional_dependency
 from dask.core import istask
-from dask.utils import apply, funcname, import_required
+from dask.utils import import_required
 
 
 def BOKEH_VERSION():
@@ -36,98 +36,6 @@ def unquote(expr):
     return expr
 
 
-def pprint_task(task, keys, label_size=60):
-    """Return a nicely formatted string for a task.
-
-    Parameters
-    ----------
-    task:
-        Value within dask graph to render as text
-    keys: iterable
-        List of keys within dask graph
-    label_size: int (optional)
-        Maximum size of output label, defaults to 60
-
-    Examples
-    --------
-    >>> from operator import add, mul
-    >>> dsk = {'a': 1,
-    ...        'b': 2,
-    ...        'c': (add, 'a', 'b'),
-    ...        'd': (add, (mul, 'a', 'b'), 'c'),
-    ...        'e': (sum, ['a', 'b', 5]),
-    ...        'f': (add,),
-    ...        'g': []}
-
-    >>> pprint_task(dsk['c'], dsk)
-    'add(_, _)'
-    >>> pprint_task(dsk['d'], dsk)
-    'add(mul(_, _), _)'
-    >>> pprint_task(dsk['e'], dsk)
-    'sum([_, _, *])'
-    >>> pprint_task(dsk['f'], dsk)
-    'add()'
-    >>> pprint_task(dsk['g'], dsk)
-    '[]'
-    """
-    if istask(task):
-        func = task[0]
-        if func is apply:
-            head = funcname(task[1])
-            tail = ")"
-            args = unquote(task[2]) if len(task) > 2 else ()
-            kwargs = unquote(task[3]) if len(task) > 3 else {}
-        else:
-            if hasattr(func, "funcs"):
-                head = "(".join(funcname(f) for f in func.funcs)
-                tail = ")" * len(func.funcs)
-            else:
-                head = funcname(task[0])
-                tail = ")"
-            args = task[1:]
-            kwargs = {}
-        if args or kwargs:
-            label_size2 = int(
-                (label_size - len(head) - len(tail)) // (len(args) + len(kwargs))
-            )
-            pprint = lambda t: pprint_task(t, keys, label_size2)
-        if args:
-            if label_size2 > 5:
-                args = ", ".join(pprint(t) for t in args)
-            else:
-                args = "..."
-        else:
-            args = ""
-        if kwargs:
-            if label_size2 > 5:
-                kwargs = ", " + ", ".join(
-                    f"{k}={pprint(v)}" for k, v in sorted(kwargs.items())
-                )
-            else:
-                kwargs = ", ..."
-        else:
-            kwargs = ""
-        return f"{head}({args}{kwargs}{tail}"
-    elif isinstance(task, list):
-        if not task:
-            return "[]"
-        elif len(task) > 3:
-            result = pprint_task(task[:3], keys, label_size)
-            return result[:-1] + ", ...]"
-        else:
-            label_size2 = int((label_size - 2 - 2 * len(task)) // len(task))
-            args = ", ".join(pprint_task(t, keys, label_size2) for t in task)
-            return f"[{args}]"
-    else:
-        try:
-            if task in keys:
-                return "_"
-            else:
-                return "*"
-        except TypeError:
-            return "*"
-
-
 def get_colors(palette, funcs):
     """Get a dict mapping funcs to colors from palette.
 
@@ -297,7 +205,7 @@ def plot_tasks(
         data["width"] = width = [e - s for (s, e) in zip(starts, ends)]
         data["x"] = [w / 2 + s - start_time for (w, s) in zip(width, starts)]
         data["y"] = [id_lk[i] + 1 for i in ids]
-        data["function"] = funcs = [pprint_task(i, dsk, label_size) for i in tasks]
+        data["function"] = funcs = [repr(i) for i in tasks]
         data["color"] = get_colors(palette, funcs)
         data["key"] = [str(i) for i in keys]
 
@@ -491,7 +399,7 @@ def plot_cache(
     if results:
         starts, ends = list(zip(*results))[3:]
         tics = sorted(unique(starts + ends))
-        groups = groupby(lambda d: pprint_task(d[1], dsk, label_size), results)
+        groups = groupby(lambda d: repr(d), results)
         data = {}
         for k, vals in groups.items():
             cnts = dict.fromkeys(tics, 0)
diff --git a/dask/local.py b/dask/local.py
index b00af3ae7..8a8ecaa42 100644
--- a/dask/local.py
+++ b/dask/local.py
@@ -116,8 +116,9 @@ from functools import partial
 from queue import Empty, Queue
 
 from dask import config
+from dask._task_spec import DataNode, DependenciesMapping, convert_legacy_graph
 from dask.callbacks import local_callbacks, unpack_callbacks
-from dask.core import _execute_task, flatten, get_dependencies, has_tasks, reverse_dict
+from dask.core import flatten, get_dependencies, reverse_dict
 from dask.order import order
 from dask.typing import Key
 
@@ -165,17 +166,19 @@ def start_state_from_dask(dsk, cache=None, sortkey=None):
         cache = config.get("cache", None)
     if cache is None:
         cache = dict()
+
+    dsk = convert_legacy_graph(dsk, all_keys=set(dsk) | set(cache))
     data_keys = set()
     for k, v in dsk.items():
-        if not has_tasks(dsk, v):
-            cache[k] = v
+        if isinstance(v, DataNode):
+            cache[k] = v()
             data_keys.add(k)
 
     dsk2 = dsk.copy()
     dsk2.update(cache)
 
-    dependencies = {k: get_dependencies(dsk2, k) for k in dsk}
-    waiting = {k: v.copy() for k, v in dependencies.items() if k not in data_keys}
+    dependencies = DependenciesMapping(dsk)
+    waiting = {k: set(v) for k, v in dependencies.items() if k not in data_keys}
 
     dependents = reverse_dict(dependencies)
     for a in cache:
@@ -223,7 +226,7 @@ def execute_task(key, task_info, dumps, loads, get_id, pack_exception):
     """
     try:
         task, data = loads(task_info)
-        result = _execute_task(task, data)
+        result = task(data)
         id = get_id()
         result = dumps((result, id))
         failed = False
@@ -423,7 +426,7 @@ def get_async(
         result_flat = {result}
     results = set(result_flat)
 
-    dsk = dict(dsk)
+    dsk = dict(convert_legacy_graph(dsk))
     with local_callbacks(callbacks) as callbacks:
         _, _, pretask_cbs, posttask_cbs, _ = unpack_callbacks(callbacks)
         started_cbs = []
@@ -508,7 +511,7 @@ def get_async(
                                 for dep in get_dependencies(dsk, key)
                             }
                             task = dsk[key]
-                            _execute_task(task, data)  # Re-execute locally
+                            task(data)  # Re-execute locally
                         else:
                             raise_exception(exc, tb)
                     res, worker_id = loads(res_info)
diff --git a/pyproject.toml b/pyproject.toml
index 73d761cc8..8d3f0ac6d 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -142,6 +142,7 @@ filterwarnings = [
     'ignore:The current Dask DataFrame implementation is deprecated.*:DeprecationWarning',
     "ignore:Minimal version of pyarrow will soon be increased to 14.0.1",
     "ignore:You are using pyarrow version .* pyarrow-hotfix",
+    "ignore:The legacy Dask DataFrame implementation is deprecated.*:FutureWarning",
 ]
 xfail_strict = true
 # pytest-timeout settings


Test patch:
diff --git a/dask/array/tests/test_linalg.py b/dask/array/tests/test_linalg.py
index 60eda63cb..91364b035 100644
--- a/dask/array/tests/test_linalg.py
+++ b/dask/array/tests/test_linalg.py
@@ -640,7 +640,7 @@ def test_solve_triangular_vector(shape, chunk):
     db = da.from_array(b, chunk)
     res = da.linalg.solve_triangular(dAu, db)
     assert_eq(res, scipy.linalg.solve_triangular(Au, b))
-    assert_eq(dAu.dot(res), b.astype(float))
+    assert_eq(dAu.dot(res), b.astype(float), rtol=1e-4)
 
     # lower
     Al = np.tril(A)
diff --git a/dask/array/tests/test_slicing.py b/dask/array/tests/test_slicing.py
index a5ea27370..2d1ab3006 100644
--- a/dask/array/tests/test_slicing.py
+++ b/dask/array/tests/test_slicing.py
@@ -1036,3 +1036,13 @@ def test_take_sorted_indexer():
             )
         },
     } == dict(result.dask)
+
+
+def test_all_none_slices_just_mappings():
+    arr = da.ones((10, 10), chunks=(1, 5))
+    result = arr[slice(None, 6), slice(None)]
+    dsk = dict(result.dask)
+    assert len([k for k in dsk if "getitem" in k[0]]) == 12
+    # check that we are just mapping the keys
+    assert all(v in dsk for k, v in dsk.items() if "getitem" in k[0])
+    assert_eq(result, np.ones((6, 10)))
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index ea9326b52..b39ba333b 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import contextlib
 import decimal
+import importlib
 import sys
 import warnings
 import weakref
@@ -6358,3 +6359,22 @@ def test_dataframe_into_delayed():
     result = delayed(delayed_func)(df)
     assert sum(map(len, result.dask.layers.values())) == 6
     result.compute()
+
+
+@pytest.mark.filterwarnings(
+    "error:The legacy Dask DataFrame implementation is deprecated.*:FutureWarning"
+)
+def test_import_raises_warning():
+    import dask
+
+    try:
+        with dask.config.set({"dataframe.query-planning": False}):
+            with pytest.raises(FutureWarning, match="The legacy"):
+                importlib.reload(dask.dataframe)
+    finally:
+        if dask.config.get("dataframe.query-planning") is False:
+            # Build without dask-expr and config is False
+            with pytest.raises(FutureWarning, match="The legacy"):
+                importlib.reload(dask.dataframe)
+        else:
+            importlib.reload(dask.dataframe)
diff --git a/dask/dataframe/tests/test_groupby.py b/dask/dataframe/tests/test_groupby.py
index 11aaeecff..df5c0b45e 100644
--- a/dask/dataframe/tests/test_groupby.py
+++ b/dask/dataframe/tests/test_groupby.py
@@ -3755,3 +3755,29 @@ def test_groupby_value_counts_all_na_partitions():
         ddf.groupby("A")["B"].value_counts(),
         df.groupby("A")["B"].value_counts(),
     )
+
+
+def test_agg_pyarrow_casts():
+    pytest.importorskip("pyarrow")
+    df = pd.DataFrame(
+        {
+            "x": range(15),
+            "y": pd.Series([pd.NA] * 10 + [1.0] * 5, dtype="double[pyarrow]"),
+            "group": ["a"] * 5 + ["b"] * 5 + ["c"] * 5,
+        }
+    )
+
+    ddf = dd.from_pandas(df, npartitions=2)
+    if PANDAS_GE_220:
+        # np.sqrt doesn't work before 2.2
+        additional_aggs = {"z": ("y", "std")}
+    else:
+        additional_aggs = {}
+
+    result = ddf.groupby("group").agg(
+        x=("y", "var"), y=("y", "mean"), **additional_aggs
+    )
+    expected = df.groupby("group").agg(
+        x=("y", "var"), y=("y", "mean"), **additional_aggs
+    )
+    assert_eq(result, expected)
diff --git a/dask/diagnostics/tests/test_profiler.py b/dask/diagnostics/tests/test_profiler.py
index b57ebf082..80b83bbc3 100644
--- a/dask/diagnostics/tests/test_profiler.py
+++ b/dask/diagnostics/tests/test_profiler.py
@@ -8,10 +8,11 @@ from timeit import default_timer
 
 import pytest
 
+from dask._task_spec import Task
 from dask.diagnostics import CacheProfiler, Profiler, ResourceProfiler
 from dask.diagnostics.profile_visualize import BOKEH_VERSION
 from dask.threaded import get
-from dask.utils import apply, tmpfile
+from dask.utils import tmpfile
 from dask.utils_test import slowadd
 
 try:
@@ -40,7 +41,8 @@ def test_profiler():
     keys = [i.key for i in prof_data]
     assert keys == ["c", "d", "e"]
     tasks = [i.task for i in prof_data]
-    assert tasks == [(add, "a", "b"), (mul, "a", "b"), (mul, "c", "d")]
+    assert len(tasks) == 3
+    assert all(isinstance(t, Task) for t in tasks)
     prof.clear()
     assert prof.results == []
 
@@ -192,38 +194,6 @@ def test_unquote():
     assert unquote(task_dask) == t
 
 
-@pytest.mark.skipif("not bokeh")
-def test_pprint_task():
-    from dask.diagnostics.profile_visualize import pprint_task
-
-    keys = {"a", "b", "c", "d", "e"}
-    assert pprint_task((add, "a", 1), keys) == "add(_, *)"
-    assert pprint_task((add, (add, "a", 1)), keys) == "add(add(_, *))"
-    res = "sum([*, _, add(_, *)])"
-    assert pprint_task((sum, [1, "b", (add, "a", 1)]), keys) == res
-    assert pprint_task((sum, (1, 2, 3, 4, 5, 6, 7)), keys) == "sum(*)"
-
-    assert len(pprint_task((sum, list(keys) * 100), keys)) < 100
-    assert pprint_task((sum, list(keys) * 100), keys) == "sum([_, _, _, ...])"
-    assert (
-        pprint_task((sum, [1, 2, (sum, ["a", 4]), 5, 6] * 100), keys)
-        == "sum([*, *, sum([_, *]), ...])"
-    )
-    assert (
-        pprint_task((sum, [1, 2, (sum, ["a", (sum, [1, 2, 3])]), 5, 6]), keys)
-        == "sum([*, *, sum([_, sum(...)]), ...])"
-    )
-
-    # With kwargs
-    def foo(w, x, y=(), z=3):
-        return w + x + sum(y) + z
-
-    task = (apply, foo, (tuple, ["a", "b"]), (dict, [["y", ["a", "b"]], ["z", "c"]]))
-    assert pprint_task(task, keys) == "foo(_, _, y=[_, _], z=_)"
-    task = (apply, foo, (tuple, ["a", "b"]), (dict, [["y", ["a", 1]], ["z", 1]]))
-    assert pprint_task(task, keys) == "foo(_, _, y=[_, *], z=*)"
-
-
 @pytest.mark.skipif("not bokeh")
 def test_profiler_plot():
     with prof:
diff --git a/dask/tests/test_callbacks.py b/dask/tests/test_callbacks.py
index 08f5a20c8..dd29121b8 100644
--- a/dask/tests/test_callbacks.py
+++ b/dask/tests/test_callbacks.py
@@ -27,7 +27,7 @@ def test_start_state_callback():
     class MyCallback(Callback):
         def _start_state(self, dsk, state):
             flag[0] = True
-            assert dsk["x"] == 1
+            assert dsk["x"]() == 1
             assert len(state["cache"]) == 1
 
     with MyCallback():
@@ -98,8 +98,8 @@ def test_nested_schedulers():
         get_threaded(outer_dsk, "b")
 
     assert not Callback.active
-    assert outer_callback.dsk == outer_dsk
-    assert inner_callback.dsk == inner_dsk
+    assert outer_callback.dsk.keys() == outer_dsk.keys()
+    assert inner_callback.dsk.keys() == inner_dsk.keys()
     assert not Callback.active
 
 
diff --git a/dask/tests/test_core.py b/dask/tests/test_core.py
index 41b42b44d..67023bd66 100644
--- a/dask/tests/test_core.py
+++ b/dask/tests/test_core.py
@@ -11,7 +11,6 @@ from dask.core import (
     get_dependencies,
     get_deps,
     getcycle,
-    has_tasks,
     ishashable,
     iskey,
     istask,
@@ -88,23 +87,6 @@ def test_istask():
     assert not istask(f(sum, 2))
 
 
-def test_has_tasks():
-    dsk = {
-        "a": [1, 2, 3],
-        "b": "a",
-        "c": [1, (inc, 1)],
-        "d": [(sum, "a")],
-        "e": ["a", "b"],
-        "f": [["a", "b"], 2, 3],
-    }
-    assert not has_tasks(dsk, dsk["a"])
-    assert has_tasks(dsk, dsk["b"])
-    assert has_tasks(dsk, dsk["c"])
-    assert has_tasks(dsk, dsk["d"])
-    assert has_tasks(dsk, dsk["e"])
-    assert has_tasks(dsk, dsk["f"])
-
-
 def test_preorder_traversal():
     t = (add, 1, 2)
     assert list(preorder_traversal(t)) == [add, 1, 2]
diff --git a/dask/tests/test_delayed.py b/dask/tests/test_delayed.py
index b9ae588cf..4ce1dbc8e 100644
--- a/dask/tests/test_delayed.py
+++ b/dask/tests/test_delayed.py
@@ -16,6 +16,7 @@ from tlz import merge
 import dask
 import dask.bag as db
 from dask import compute
+from dask.base import collections_to_dsk
 from dask.delayed import Delayed, delayed, to_task_dask
 from dask.highlevelgraph import HighLevelGraph
 from dask.utils_test import inc
@@ -855,3 +856,28 @@ def test_delayed_function_attributes_forwarded():
     assert add.__name__ == "add"
     assert add.__doc__ == "This is a docstring"
     assert add.__wrapped__(1, 2) == 3
+
+
+def test_delayed_fusion():
+    @delayed
+    def test(i):
+        return i + 1
+
+    @delayed
+    def test2(i):
+        return i + 2
+
+    @delayed
+    def test3(i):
+        return i + 3
+
+    obj = test3(test2(test(10)))
+    dsk = dict(collections_to_dsk([obj]))
+    assert len(dsk) == 3
+
+    obj2 = test3(test2(test(10)))
+    with dask.config.set({"optimization.fuse.delayed": True}):
+        dsk2 = dict(collections_to_dsk([obj]))
+        result = dask.compute(obj2)
+    assert len(dsk2) == 2
+    assert dask.compute(obj) == result
diff --git a/dask/tests/test_task_spec.py b/dask/tests/test_task_spec.py
index 46c3e6974..05f92d0cf 100644
--- a/dask/tests/test_task_spec.py
+++ b/dask/tests/test_task_spec.py
@@ -62,6 +62,54 @@ def test_convert_legacy_dsk_skip_new():
     assert converted == dsk
 
 
+def test_repr():
+    t = Task("key", func, "a", "b")
+    assert repr(t) == "<Task 'key' func('a', 'b')>"
+
+    t = Task("nested", func2, t, t.ref())
+    assert (
+        repr(t) == "<Task 'nested' func2(<Task 'key' func('a', 'b')>, Alias(key->key))>"
+    )
+
+    def long_function_name_longer_even_longer(a, b):
+        return a + b
+
+    t = Task("long", long_function_name_longer_even_longer, t, t.ref())
+    assert repr(t) == "<Task 'long' long_function_name_longer_even_longer(...)>"
+
+    def use_kwargs(a, kwarg=None):
+        return a + kwarg
+
+    t = Task("kwarg", use_kwargs, "foo", kwarg="kwarg_value")
+    assert repr(t) == "<Task 'kwarg' use_kwargs('foo', kwarg='kwarg_value')>"
+
+
+def test_unpickled_repr():
+    t = pickle.loads(pickle.dumps(Task("key", func, "a", "b")))
+    assert repr(t) == "<Task 'key' func('a', 'b')>"
+
+    t = pickle.loads(pickle.dumps(Task("nested", func2, t, t.ref())))
+    assert (
+        repr(t) == "<Task 'nested' func2(<Task 'key' func('a', 'b')>, Alias(key->key))>"
+    )
+
+    def long_function_name_longer_even_longer(a, b):
+        return a + b
+
+    t = pickle.loads(
+        pickle.dumps(Task("long", long_function_name_longer_even_longer, t, t.ref()))
+    )
+    assert repr(t) == "<Task 'long' long_function_name_longer_even_longer(...)>"
+
+    def use_kwargs(a, kwarg=None):
+        return a + kwarg
+
+    t = pickle.loads(
+        pickle.dumps(Task("kwarg", use_kwargs, "foo", kwarg="kwarg_value"))
+    )
+    assert repr(t) == "<Task 'kwarg' use_kwargs('foo', kwarg='kwarg_value')>"
+
+
 def _assert_dsk_conversion(new_dsk):
     vals = [(k, v.key) for k, v in new_dsk.items()]
     assert all(v[0] == v[1] for v in vals), vals
@@ -754,7 +802,7 @@ def test_execute_tasks_in_graph():
         t3 := Task("key-3", func, "foo", "bar"),
         Task("key-4", func, t3.ref(), t2.ref()),
     ]
-    res = execute_graph(dsk)
+    res = execute_graph(dsk, keys=["key-4"])
     assert len(res) == 1
     assert res["key-4"] == "foo-bar-a-b=c"