Repository: dask/dask
Pull request #11657 (2025-01-09 22:06:20)

Avoid Dict in fused tasks
- [ ] Partially addresses https://github.com/dask/distributed/issues/8958 and reduces the graph size.
- [ ] 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._

     15 files  ±0       15 suites  ±0   4h 34m 20s ⏱️ +30s
 17 151 tests ±0   15 954 ✅  - 2   1 195 💤 ±0  2 ❌ +2 
211 304 runs  ±0  194 123 ✅  - 6  17 177 💤 +2  4 ❌ +4 

For more details on these failures, see [this check](https://github.com/dask/dask/runs/35401026988).

Results for commit fa5071e5. ± Comparison against base commit ecb0cdd9.

[test-results]:data:application/gzip;base64,H4sIAGhRgGcC/02Myw7CIBQFf8WwdsEtXB7+jAEKCbEPQ2Fl/HehaanLmZMzHxLi5DfyuAHeb2QrMV80lmRyXJfGggtaVV3zvktAOPm5Fef2k0Z+yVd8NwkauwsmTtUNXfiU1lRNa6eytPQAwCg/uKc1h4F1eaQlSHm6I92P/2W3znPMFUgwSCV4FJwrVMJZVM5YFmCkilHltQc2GmvJ9wcVkomiGgEAAA==

phofl: The errors seem legit?
hendrikmakait: This appears to be a stochastic error. I couldn't reproduce it reliably on my machine (even in 1000 attempts), but I now ensure deterministic ordering, which _should_ avoid the error.
hendrikmakait: > TypeError: '<' not supported between instances of 'tuple' and 'str'

Oh well, I'll have to think about this for a minute.
fjetter: Can we test this? The entire "use few objects (that are GC tracked)" seems like a brittle property and a unit test would be very helpful
hendrikmakait: We could test that fusion results in n_subtasks + 1 task objects, but everything else seems like testing implementation.

Patch:
diff --git a/dask/_task_spec.py b/dask/_task_spec.py
index d0fe635b4..80df3cfbd 100644
--- a/dask/_task_spec.py
+++ b/dask/_task_spec.py
@@ -195,10 +195,10 @@ class _MultiContainer(Container):
 SubgraphType = None
 
 
-def _execute_subgraph(inner_dsk, outkey, inkeys):
+def _execute_subgraph(inner_dsk, outkey, inkeys, *dependencies):
     final = {}
     final.update(inner_dsk)
-    for k, v in inkeys.items():
+    for k, v in zip(inkeys, dependencies):
         final[k] = DataNode(None, v)
     res = execute_graph(final, keys=[outkey])
     return res[outkey]
@@ -459,19 +459,19 @@ class GraphNode:
         for t in tasks:
             all_deps.update(t.dependencies)
             all_keys.add(t.key)
-        external_deps = all_deps - all_keys
+        external_deps = tuple(sorted(all_deps - all_keys, key=hash))
         leafs = all_keys - all_deps
         if len(leafs) > 1:
             raise ValueError(f"Cannot fuse tasks with multiple outputs {leafs}")
 
         outkey = leafs.pop()
-
         return Task(
             key or outkey,
             _execute_subgraph,
             {t.key: t for t in tasks},
             outkey,
-            (Dict({k: Alias(k) for k in external_deps}) if external_deps else {}),
+            external_deps,
+            *(TaskRef(k) for k in external_deps),
             _data_producer=any(t.data_producer for t in tasks),
         )
 
diff --git a/dask/array/core.py b/dask/array/core.py
index 85fe6b1b3..f3185b48c 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -3881,7 +3881,7 @@ def to_zarr(
 
     if storage_options:
         if _zarr_v3():
-            store = zarr.storage.RemoteStore(
+            store = zarr.storage.FsspecStore(
                 url, mode=kwargs.pop("mode", "a"), **storage_options
             )
         else:
diff --git a/dask/array/optimization.py b/dask/array/optimization.py
index 20893e30a..6cf9ae70a 100644
--- a/dask/array/optimization.py
+++ b/dask/array/optimization.py
@@ -1,15 +1,38 @@
 from __future__ import annotations
 
+from collections.abc import Callable
 from itertools import zip_longest
 from numbers import Integral
+from typing import Any
+
+import numpy as np
 
 from dask import config
-from dask._task_spec import convert_legacy_graph, fuse_linear_task_spec
+from dask._task_spec import (
+    Alias,
+    DataNode,
+    GraphNode,
+    Task,
+    _execute_subgraph,
+    convert_legacy_graph,
+    fuse_linear_task_spec,
+)
+from dask.array.chunk import getitem
+from dask.array.core import getter, getter_inline, getter_nofancy
 from dask.blockwise import fuse_roots, optimize_blockwise
 from dask.core import flatten
 from dask.highlevelgraph import HighLevelGraph
 from dask.utils import ensure_dict
 
+# All get* functions the optimizations know about
+
+
+GETTERS = (getter, getter_nofancy, getter_inline, getitem)
+# These get* functions aren't ever completely removed from the graph,
+# even if the index should be a no-op by numpy semantics. Some array-like's
+# don't completely follow semantics, making indexing always necessary.
+GETNOREMOVE = (getter, getter_nofancy)
+
 
 def optimize(dsk, keys, **kwargs):
     """Optimize dask for array computation
@@ -37,6 +60,129 @@ def optimize(dsk, keys, **kwargs):
 
     dsk = convert_legacy_graph(dsk)
     dsk = fuse_linear_task_spec(dsk, keys=keys)
+    dsk = optimize_slices(dsk)
+
+    return dsk
+
+
+def _is_getter_task(
+    value: GraphNode,
+) -> tuple[Callable, Any, Any, bool, bool | None] | None:
+    """Check if a value in a Dask graph looks like a getter.
+    1. Is it a tuple with the first element a known getter.
+    If a getter is found, it returns a tuple with (getter, array, index, asarray, lock).
+    Otherwise it returns ``None``.
+    """
+    if not isinstance(value, Task):
+        return None
+    func = value.func
+    if func in GETTERS:
+        get = func
+    else:
+        return None
+
+    length = len(value.args)
+    if length == 2:
+        # getter defaults to asarray=True, getitem is semantically False
+        return get, value.args[0], value.args[1], get is not getitem, None
+    elif length == 4:
+        return get, *list(value.args)
+
+    return None
+
+
+def optimize_slices(dsk):
+    """Optimize slices
+    1.  Fuse repeated slices, like x[5:][2:6] -> x[7:11]
+
+    This is generally not very important since we are fusing those tasks anyway. There
+    is one specific exception to how xarray implements opening netcdf files and subsequent
+    slices. Not merging them together can cause reading the whole netcdf file before
+    we drop the unnecessary data. Fusing slices avoids that pattern.
+
+    See https://github.com/pydata/xarray/issues/9926
+    """
+    fancy_ind_types = (list, np.ndarray)
+    dsk = dsk.copy()
+    for _, v in dsk.items():
+        if not (isinstance(v, Task) and v.func is _execute_subgraph):
+            continue
+
+        inner_graph: dict = v.args[0]
+
+        seen = set()
+
+        for inner_key, inner_value in inner_graph.items():
+            if inner_key in seen:
+                continue
+            else:
+                seen.add(inner_key)
+
+            if a_task := _is_getter_task(inner_value):
+                temp_key = inner_key
+                get, a, a_index, a_asarray, a_lock = a_task
+                fused = False
+
+                while a.key in inner_graph and (
+                    b_task := _is_getter_task(inner_graph[a.key])
+                ):
+                    seen.add(a.key)
+                    f2, b, b_index, b_asarray, b_lock = b_task
+
+                    if isinstance(a_index, DataNode):
+                        a_index = a_index.value
+                    if isinstance(b_index, DataNode):
+                        b_index = b_index.value
+
+                    if a_lock and a_lock is not b_lock:
+                        break
+                    if (type(a_index) is tuple) != (type(b_index) is tuple):
+                        break
+                    if type(a_index) is tuple:
+                        indices = b_index + a_index
+                        if len(a_index) != len(b_index) and any(
+                            i is None for i in indices
+                        ):
+                            break
+                        if f2 is getter_nofancy and not any(
+                            isinstance(i, fancy_ind_types) for i in indices
+                        ):
+                            break
+                    elif f2 is getter_nofancy and (
+                        type(a_index) in fancy_ind_types
+                        or type(b_index) in fancy_ind_types
+                    ):
+                        break
+                    try:
+                        c_index = fuse_slice(b_index, a_index)
+                        # rely on fact that nested gets never decrease in
+                        # strictness e.g. `(getter_nofancy, (getter, ...))` never
+                        # happens
+                        get = getter if f2 is getter_inline else f2
+
+                        inner_graph[temp_key] = Alias(temp_key, a.key)
+                        temp_key = a.key
+                        fused = True
+                    except NotImplementedError:
+                        break
+                    a, a_index, a_lock = b, c_index, b_lock
+                    a_asarray |= b_asarray
+
+                    if not isinstance(a, Task):
+                        break
+
+                if not fused:
+                    pass
+                elif get is getitem or (a_asarray and not a_lock):
+                    # default settings are fine, drop the extra parameters Since we
+                    # always fallback to inner `get` functions, `get is getitem`
+                    # can only occur if all gets are getitem, meaning all
+                    # parameters must be getitem defaults.
+                    inner_graph[temp_key] = Task(temp_key, get, a, a_index)
+                else:
+                    inner_graph[temp_key] = Task(
+                        temp_key, get, a, a_index, a_asarray, a_lock
+                    )
 
     return dsk
 
diff --git a/dask/bag/core.py b/dask/bag/core.py
index 6e334b441..4e9278a00 100644
--- a/dask/bag/core.py
+++ b/dask/bag/core.py
@@ -112,7 +112,7 @@ def lazify_task(task, start=True):
             assert len(task.args) == 1
             task = task.args[0]
         if task.func is _execute_subgraph:
-            subgraph, outkey, inkeys = task.args
+            subgraph, outkey, inkeys, *dependencies = task.args
             # 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 = {
@@ -125,6 +125,7 @@ def lazify_task(task, start=True):
                 subgraph,
                 outkey,
                 inkeys,
+                *dependencies,
                 _data_producer=task.data_producer,
             )
         return Task(
diff --git a/dask/dataframe/__init__.py b/dask/dataframe/__init__.py
index 1536c1cb8..468c72e1b 100644
--- a/dask/dataframe/__init__.py
+++ b/dask/dataframe/__init__.py
@@ -41,29 +41,82 @@ from dask.dataframe.dask_expr import (
     merge,
     merge_asof,
     pivot_table,
+    read_parquet,
+    repartition,
+    to_bag,
+    to_datetime,
+    to_numeric,
+    to_parquet,
+    to_records,
+    to_timedelta,
+)
+from dask.dataframe.groupby import Aggregation
+from dask.dataframe.io import (
+    demo,
     read_csv,
     read_fwf,
     read_hdf,
     read_json,
     read_orc,
-    read_parquet,
     read_sql,
     read_sql_query,
     read_sql_table,
     read_table,
-    repartition,
-    to_bag,
     to_csv,
-    to_datetime,
     to_hdf,
     to_json,
-    to_numeric,
     to_orc,
-    to_parquet,
-    to_records,
     to_sql,
-    to_timedelta,
 )
-from dask.dataframe.groupby import Aggregation
-from dask.dataframe.io import demo
 from dask.dataframe.utils import assert_eq
+
+__all__ = [
+    "DataFrame",
+    "Index",
+    "Scalar",
+    "Series",
+    "Aggregation",
+    "backends",
+    "compute",
+    "concat",
+    "dispatch",
+    "from_array",
+    "from_dask_array",
+    "from_delayed",
+    "from_dict",
+    "from_graph",
+    "from_map",
+    "from_pandas",
+    "get_collection_type",
+    "get_dummies",
+    "isna",
+    "map_overlap",
+    "map_partitions",
+    "melt",
+    "merge",
+    "merge_asof",
+    "pivot_table",
+    "read_csv",
+    "read_fwf",
+    "read_hdf",
+    "read_json",
+    "read_orc",
+    "read_parquet",
+    "read_sql",
+    "read_sql_query",
+    "read_sql_table",
+    "read_table",
+    "repartition",
+    "to_bag",
+    "to_csv",
+    "to_datetime",
+    "to_hdf",
+    "to_json",
+    "to_numeric",
+    "to_orc",
+    "to_parquet",
+    "to_records",
+    "to_sql",
+    "to_timedelta",
+    "assert_eq",
+]
diff --git a/dask/dataframe/api.py b/dask/dataframe/api.py
new file mode 100644
index 000000000..315eb00c6
--- /dev/null
+++ b/dask/dataframe/api.py
@@ -0,0 +1,10 @@
+from __future__ import annotations
+
+from dask.dataframe.core import get_parallel_type  # noqa: F401
+from dask.dataframe.dask_expr import FrameBase, elemwise, new_collection  # noqa: F401
+from dask.dataframe.dask_expr._expr import are_co_aligned, emulate  # noqa: F401
+from dask.dataframe.dask_expr._groupby import GroupBy, SeriesGroupBy  # noqa: F401
+from dask.dataframe.dask_expr._reductions import ApplyConcatApply  # noqa: F401
+from dask.dataframe.dask_expr._rolling import Rolling  # noqa: F401
+from dask.dataframe.extensions import make_array_nonempty, make_scalar  # noqa: F401
+from dask.dataframe.utils import meta_nonempty  # noqa: F401
diff --git a/dask/dataframe/dask_expr/__init__.py b/dask/dataframe/dask_expr/__init__.py
index 1be35e241..8d85c18c0 100644
--- a/dask/dataframe/dask_expr/__init__.py
+++ b/dask/dataframe/dask_expr/__init__.py
@@ -7,15 +7,5 @@ from dask.dataframe.dask_expr._dummies import get_dummies
 from dask.dataframe.dask_expr._groupby import Aggregation
 from dask.dataframe.dask_expr.io._delayed import from_delayed
 from dask.dataframe.dask_expr.io.bag import to_bag
-from dask.dataframe.dask_expr.io.csv import to_csv
-from dask.dataframe.dask_expr.io.hdf import read_hdf, to_hdf
-from dask.dataframe.dask_expr.io.json import read_json, to_json
-from dask.dataframe.dask_expr.io.orc import read_orc, to_orc
 from dask.dataframe.dask_expr.io.parquet import to_parquet
 from dask.dataframe.dask_expr.io.records import to_records
-from dask.dataframe.dask_expr.io.sql import (
-    read_sql,
-    read_sql_query,
-    read_sql_table,
-    to_sql,
-)
diff --git a/dask/dataframe/dask_expr/_collection.py b/dask/dataframe/dask_expr/_collection.py
index 6ecddfd2f..3d4994825 100644
--- a/dask/dataframe/dask_expr/_collection.py
+++ b/dask/dataframe/dask_expr/_collection.py
@@ -949,7 +949,7 @@ Expr={expr}"""
 
     @derived_from(pd.DataFrame)
     def resample(self, rule, closed=None, label=None):
-        from dask.dataframe.dask_expr._resample import Resampler
+        from dask.dataframe.tseries.resample import Resampler
 
         return Resampler(self, rule, **{"closed": closed, "label": label})
 
@@ -2406,7 +2406,7 @@ Expr={expr}"""
         parallel=False,
         engine_kwargs=None,
     ):
-        from dask.dataframe.dask_expr.io.sql import to_sql
+        from dask.dataframe.io.sql import to_sql
 
         return to_sql(
             self,
@@ -2426,13 +2426,13 @@ Expr={expr}"""
 
     def to_orc(self, path, *args, **kwargs):
         """See dd.to_orc docstring for more information"""
-        from dask.dataframe.dask_expr.io.orc import to_orc
+        from dask.dataframe.io.orc import to_orc
 
         return to_orc(self, path, *args, **kwargs)
 
     def to_csv(self, filename, **kwargs):
         """See dd.to_csv docstring for more information"""
-        from dask.dataframe.dask_expr.io.csv import to_csv
+        from dask.dataframe.io.csv import to_csv
 
         return to_csv(self, filename, **kwargs)
 
@@ -2481,7 +2481,7 @@ Expr={expr}"""
 
     def to_hdf(self, path_or_buf, key, mode="a", append=False, **kwargs):
         """See dd.to_hdf docstring for more information"""
-        from dask.dataframe.dask_expr.io.hdf import to_hdf
+        from dask.dataframe.io.hdf import to_hdf
 
         return to_hdf(self, path_or_buf, key, mode, append, **kwargs)
 
@@ -3101,7 +3101,7 @@ class DataFrame(FrameBase):
                 "installed."
             )
         if meta is None:
-            meta = expr._emulate(M.map, self, func, na_action=na_action, udf=True)
+            meta = expr.emulate(M.map, self, func, na_action=na_action, udf=True)
             warnings.warn(meta_warning(meta))
         return new_collection(expr.Map(self, arg=func, na_action=na_action, meta=meta))
 
@@ -3233,7 +3233,7 @@ class DataFrame(FrameBase):
             )
             raise NotImplementedError(msg)
         if meta is no_default:
-            meta = expr._emulate(
+            meta = expr.emulate(
                 M.apply, self, function, args=args, udf=True, axis=axis, **kwargs
             )
             warnings.warn(meta_warning(meta))
@@ -4200,7 +4200,7 @@ class Series(FrameBase):
                     expr.MapAlign(self, arg, op=None, na_action=na_action, meta=meta)
                 )
         if meta is None:
-            meta = expr._emulate(M.map, self, arg, na_action=na_action, udf=True)
+            meta = expr.emulate(M.map, self, arg, na_action=na_action, udf=True)
             warnings.warn(meta_warning(meta))
         return new_collection(expr.Map(self, arg=arg, na_action=na_action, meta=meta))
 
@@ -4384,7 +4384,7 @@ class Series(FrameBase):
         """
         self._validate_axis(axis)
         if meta is no_default:
-            meta = expr._emulate(M.apply, self, function, args=args, udf=True, **kwargs)
+            meta = expr.emulate(M.apply, self, function, args=args, udf=True, **kwargs)
             warnings.warn(meta_warning(meta))
         return new_collection(self.expr.apply(function, *args, meta=meta, **kwargs))
 
@@ -4727,7 +4727,7 @@ class Index(Series):
                     expr.MapIndexAlign(self, arg, na_action, meta, is_monotonic)
                 )
         if meta is None:
-            meta = expr._emulate(M.map, self, arg, na_action=na_action, udf=True)
+            meta = expr.emulate(M.map, self, arg, na_action=na_action, udf=True)
             warnings.warn(meta_warning(meta))
         return new_collection(
             expr.Map(
@@ -5083,65 +5083,6 @@ def from_dask_array(x, columns=None, index=None, meta=None):
     return from_dask_array(x, columns=columns, index=index, meta=meta)
 
 
-@dataframe_creation_dispatch.register_inplace("pandas")
-def read_csv(
-    path,
-    *args,
-    header="infer",
-    storage_options=None,
-    **kwargs,
-):
-    from dask.dataframe.io.csv import read_csv as _read_csv
-
-    return _read_csv(
-        path,
-        *args,
-        header=header,
-        storage_options=storage_options,
-        **kwargs,
-    )
-
-
-def read_table(
-    path,
-    *args,
-    header="infer",
-    usecols=None,
-    storage_options=None,
-    **kwargs,
-):
-    from dask.dataframe.io.csv import read_table as _read_table
-
-    return _read_table(
-        path,
-        *args,
-        header=header,
-        storage_options=storage_options,
-        usecols=usecols,
-        **kwargs,
-    )
-
-
-def read_fwf(
-    path,
-    *args,
-    header="infer",
-    usecols=None,
-    storage_options=None,
-    **kwargs,
-):
-    from dask.dataframe.io.csv import read_fwf as _read_fwf
-
-    return _read_fwf(
-        path,
-        *args,
-        header=header,
-        storage_options=storage_options,
-        usecols=usecols,
-        **kwargs,
-    )
-
-
 @dataframe_creation_dispatch.register_inplace("pandas")
 def read_parquet(
     path=None,
diff --git a/dask/dataframe/dask_expr/_expr.py b/dask/dataframe/dask_expr/_expr.py
index a3dcaa55f..dac7d0615 100644
--- a/dask/dataframe/dask_expr/_expr.py
+++ b/dask/dataframe/dask_expr/_expr.py
@@ -2893,7 +2893,7 @@ class Partitions(Expr):
 
     def _simplify_down(self):
         from dask.dataframe.dask_expr import SetIndexBlockwise
-        from dask.dataframe.dask_expr._resample import ResampleAggregation
+        from dask.dataframe.tseries.resample import ResampleAggregation
 
         if isinstance(self.frame, Blockwise) and not isinstance(
             self.frame, (BlockwiseIO, Fused, SetIndexBlockwise, ResampleAggregation)
@@ -4033,7 +4033,7 @@ def _extract_meta(x, nonempty=False):
         return x
 
 
-def _emulate(func, *args, udf=False, **kwargs):
+def emulate(func, *args, udf=False, **kwargs):
     """
     Apply a function using args / kwargs. If arguments contain dd.DataFrame /
     dd.Series, using internal cache (``_meta``) for calculation
@@ -4053,7 +4053,7 @@ def _get_meta_map_partitions(args, dfs, func, kwargs, meta, parent_meta):
         # Use non-normalized kwargs here, as we want the real values (not
         # delayed values)
         a = [meta_nonempty(arg._meta) if isinstance(arg, Expr) else arg for arg in args]
-        meta = _emulate(func, *a, udf=True, **kwargs)
+        meta = emulate(func, *a, udf=True, **kwargs)
         meta_is_emulated = True
     else:
         meta = make_meta(meta, index=meta_index, parent_meta=parent_meta)
diff --git a/dask/dataframe/dask_expr/_merge_asof.py b/dask/dataframe/dask_expr/_merge_asof.py
index 26ed238d8..436eb229c 100644
--- a/dask/dataframe/dask_expr/_merge_asof.py
+++ b/dask/dataframe/dask_expr/_merge_asof.py
@@ -199,6 +199,16 @@ class MergeAsofIndexed(MergeAsof):
     def _divisions(self):
         return self.left.divisions
 
+    @functools.cached_property
+    def left_on(self):
+        # For optimisation
+        return self.left_by
+
+    @functools.cached_property
+    def right_on(self):
+        # For optimisation
+        return self.right_by
+
     @functools.cached_property
     def _kwargs(self):
         return {
diff --git a/dask/dataframe/dask_expr/_resample.py b/dask/dataframe/dask_expr/_resample.py
deleted file mode 100644
index 0c4ed7843..000000000
--- a/dask/dataframe/dask_expr/_resample.py
+++ /dev/null
@@ -1,293 +0,0 @@
-from __future__ import annotations
-
-import functools
-from collections import namedtuple
-
-import numpy as np
-from pandas.core.resample import Resampler as pd_Resampler
-
-from dask.dataframe.dask_expr._collection import new_collection
-from dask.dataframe.dask_expr._expr import (
-    Blockwise,
-    Expr,
-    Projection,
-    make_meta,
-    plain_column_projection,
-)
-from dask.dataframe.dask_expr._repartition import Repartition
-from dask.dataframe.dispatch import meta_nonempty
-from dask.dataframe.tseries.resample import _resample_bin_and_out_divs, _resample_series
-from dask.utils import derived_from
-
-BlockwiseDep = namedtuple(typename="BlockwiseDep", field_names=["iterable"])  # type: ignore
-
-
-class ResampleReduction(Expr):
-    _parameters = [
-        "frame",
-        "rule",
-        "kwargs",
-        "how_args",
-        "how_kwargs",
-    ]
-    _defaults = {
-        "closed": None,
-        "label": None,
-        "kwargs": None,
-        "how_args": (),
-        "how_kwargs": None,
-    }
-    how: str | None = None
-    fill_value = np.nan
-
-    @functools.cached_property
-    def npartitions(self):
-        return self.frame.npartitions
-
-    def _divisions(self):
-        return self._resample_divisions[1]
-
-    @functools.cached_property
-    def _meta(self):
-        resample = meta_nonempty(self.frame._meta).resample(self.rule, **self.kwargs)
-        meta = getattr(resample, self.how)(*self.how_args, **self.how_kwargs or {})
-        return make_meta(meta)
-
-    @functools.cached_property
-    def kwargs(self):
-        return {} if self.operand("kwargs") is None else self.operand("kwargs")
-
-    @functools.cached_property
-    def how_kwargs(self):
-        return {} if self.operand("how_kwargs") is None else self.operand("how_kwargs")
-
-    @functools.cached_property
-    def _resample_divisions(self):
-        return _resample_bin_and_out_divs(
-            self.frame.divisions, self.rule, **self.kwargs or {}
-        )
-
-    def _simplify_up(self, parent, dependents):
-        if isinstance(parent, Projection):
-            return plain_column_projection(self, parent, dependents)
-
-    def _lower(self):
-        partitioned = Repartition(
-            self.frame, new_divisions=self._resample_divisions[0], force=True
-        )
-        output_divisions = self._resample_divisions[1]
-        return ResampleAggregation(
-            partitioned,
-            BlockwiseDep(output_divisions[:-1]),
-            BlockwiseDep(output_divisions[1:]),
-            BlockwiseDep(["left"] * (len(output_divisions[1:]) - 1) + [None]),
-            self.rule,
-            self.kwargs,
-            self.how,
-            self.fill_value,
-            list(self.how_args),
-            self.how_kwargs,
-        )
-
-
-class ResampleAggregation(Blockwise):
-    _parameters = [
-        "frame",
-        "divisions_left",
-        "divisions_right",
-        "closed",
-        "rule",
-        "kwargs",
-        "how",
-        "fill_value",
-        "how_args",
-        "how_kwargs",
-    ]
-    operation = staticmethod(_resample_series)
-
-    @functools.cached_property
-    def _meta(self):
-        return self.frame._meta
-
-    def _divisions(self):
-        return list(self.divisions_left.iterable) + [self.divisions_right.iterable[-1]]
-
-    def _blockwise_arg(self, arg, i):
-        if isinstance(arg, BlockwiseDep):
-            return arg.iterable[i]
-        return super()._blockwise_arg(arg, i)
-
-
-class ResampleCount(ResampleReduction):
-    how = "count"
-    fill_value = 0
-
-
-class ResampleSum(ResampleReduction):
-    how = "sum"
-    fill_value = 0
-
-
-class ResampleProd(ResampleReduction):
-    how = "prod"
-
-
-class ResampleMean(ResampleReduction):
-    how = "mean"
-
-
-class ResampleMin(ResampleReduction):
-    how = "min"
-
-
-class ResampleMax(ResampleReduction):
-    how = "max"
-
-
-class ResampleFirst(ResampleReduction):
-    how = "first"
-
-
-class ResampleLast(ResampleReduction):
-    how = "last"
-
-
-class ResampleVar(ResampleReduction):
-    how = "var"
-
-
-class ResampleStd(ResampleReduction):
-    how = "std"
-
-
-class ResampleSize(ResampleReduction):
-    how = "size"
-    fill_value = 0
-
-
-class ResampleNUnique(ResampleReduction):
-    how = "nunique"
-    fill_value = 0
-
-
-class ResampleMedian(ResampleReduction):
-    how = "median"
-
-
-class ResampleQuantile(ResampleReduction):
-    how = "quantile"
-
-
-class ResampleOhlc(ResampleReduction):
-    how = "ohlc"
-
-
-class ResampleSem(ResampleReduction):
-    how = "sem"
-
-
-class ResampleAgg(ResampleReduction):
-    how = "agg"
-
-    def _simplify_up(self, parent, dependents):
-        # Disable optimization in `agg`; function may access other columns
-        return
-
-
-class Resampler:
-    """Aggregate using one or more operations
-
-    The purpose of this class is to expose an API similar
-    to Pandas' `Resampler` for dask-expr
-    """
-
-    def __init__(self, obj, rule, **kwargs):
-        if obj.divisions[0] is None:
-            msg = (
-                "Can only resample dataframes with known divisions\n"
-                "See https://docs.dask.org/en/latest/dataframe-design.html#partitions\n"
-                "for more information."
-            )
-            raise ValueError(msg)
-        self.obj = obj
-        self.rule = rule
-        self.kwargs = kwargs
-
-    def _single_agg(self, expr_cls, how_args=(), how_kwargs=None):
-        return new_collection(
-            expr_cls(
-                self.obj,
-                self.rule,
-                self.kwargs,
-                how_args=how_args,
-                how_kwargs=how_kwargs,
-            )
-        )
-
-    @derived_from(pd_Resampler)
-    def count(self):
-        return self._single_agg(ResampleCount)
-
-    @derived_from(pd_Resampler)
-    def sum(self):
-        return self._single_agg(ResampleSum)
-
-    @derived_from(pd_Resampler)
-    def prod(self):
-        return self._single_agg(ResampleProd)
-
-    @derived_from(pd_Resampler)
-    def mean(self):
-        return self._single_agg(ResampleMean)
-
-    @derived_from(pd_Resampler)
-    def min(self):
-        return self._single_agg(ResampleMin)
-
-    @derived_from(pd_Resampler)
-    def max(self):
-        return self._single_agg(ResampleMax)
-
-    @derived_from(pd_Resampler)
-    def first(self):
-        return self._single_agg(ResampleFirst)
-
-    @derived_from(pd_Resampler)
-    def last(self):
-        return self._single_agg(ResampleLast)
-
-    @derived_from(pd_Resampler)
-    def var(self):
-        return self._single_agg(ResampleVar)
-
-    @derived_from(pd_Resampler)
-    def std(self):
-        return self._single_agg(ResampleStd)
-
-    @derived_from(pd_Resampler)
-    def size(self):
-        return self._single_agg(ResampleSize)
-
-    @derived_from(pd_Resampler)
-    def nunique(self):
-        return self._single_agg(ResampleNUnique)
-
-    @derived_from(pd_Resampler)
-    def median(self):
-        return self._single_agg(ResampleMedian)
-
-    @derived_from(pd_Resampler)
-    def quantile(self):
-        return self._single_agg(ResampleQuantile)
-
-    @derived_from(pd_Resampler)
-    def ohlc(self):
-        return self._single_agg(ResampleOhlc)
-
-    @derived_from(pd_Resampler)
-    def sem(self):
-        return self._single_agg(ResampleSem)
-
-    @derived_from(pd_Resampler)
-    def agg(self, func, *args, **kwargs):
-        return self._single_agg(ResampleAgg, how_args=(func, *args), how_kwargs=kwargs)
diff --git a/dask/dataframe/dask_expr/io/__init__.py b/dask/dataframe/dask_expr/io/__init__.py
index b86fbd5f5..7fdf8035f 100644
--- a/dask/dataframe/dask_expr/io/__init__.py
+++ b/dask/dataframe/dask_expr/io/__init__.py
@@ -1,5 +1,4 @@
 from __future__ import annotations
 
-from dask.dataframe.dask_expr.io.csv import *
 from dask.dataframe.dask_expr.io.io import *
 from dask.dataframe.dask_expr.io.parquet import *
diff --git a/dask/dataframe/dask_expr/io/csv.py b/dask/dataframe/dask_expr/io/csv.py
deleted file mode 100644
index 0cb9d32bd..000000000
--- a/dask/dataframe/dask_expr/io/csv.py
+++ /dev/null
@@ -1,148 +0,0 @@
-from __future__ import annotations
-
-
-def to_csv(
-    df,
-    filename,
-    single_file=False,
-    encoding="utf-8",
-    mode="wt",
-    name_function=None,
-    compression=None,
-    compute=True,
-    scheduler=None,
-    storage_options=None,
-    header_first_partition_only=None,
-    compute_kwargs=None,
-    **kwargs,
-):
-    """
-    Store Dask DataFrame to CSV files
-
-    One filename per partition will be created. You can specify the
-    filenames in a variety of ways.
-
-    Use a globstring::
-
-    >>> df.to_csv('/path/to/data/export-*.csv')  # doctest: +SKIP
-
-    The * will be replaced by the increasing sequence 0, 1, 2, ...
-
-    ::
-
-        /path/to/data/export-0.csv
-        /path/to/data/export-1.csv
-
-    Use a globstring and a ``name_function=`` keyword argument.  The
-    name_function function should expect an integer and produce a string.
-    Strings produced by name_function must preserve the order of their
-    respective partition indices.
-
-    >>> from datetime import date, timedelta
-    >>> def name(i):
-    ...     return str(date(2015, 1, 1) + i * timedelta(days=1))
-
-    >>> name(0)
-    '2015-01-01'
-    >>> name(15)
-    '2015-01-16'
-
-    >>> df.to_csv('/path/to/data/export-*.csv', name_function=name)  # doctest: +SKIP
-
-    ::
-
-        /path/to/data/export-2015-01-01.csv
-        /path/to/data/export-2015-01-02.csv
-        ...
-
-    You can also provide an explicit list of paths::
-
-    >>> paths = ['/path/to/data/alice.csv', '/path/to/data/bob.csv', ...]  # doctest: +SKIP
-    >>> df.to_csv(paths) # doctest: +SKIP
-
-    You can also provide a directory name:
-
-    >>> df.to_csv('/path/to/data') # doctest: +SKIP
-
-    The files will be numbered 0, 1, 2, (and so on) suffixed with '.part':
-
-    ::
-
-        /path/to/data/0.part
-        /path/to/data/1.part
-
-    Parameters
-    ----------
-    df : dask.DataFrame
-        Data to save
-    filename : string or list
-        Absolute or relative filepath(s). Prefix with a protocol like ``s3://``
-        to save to remote filesystems.
-    single_file : bool, default False
-        Whether to save everything into a single CSV file. Under the
-        single file mode, each partition is appended at the end of the
-        specified CSV file.
-    encoding : string, default 'utf-8'
-        A string representing the encoding to use in the output file.
-    mode : str, default 'w'
-        Python file mode. The default is 'w' (or 'wt'), for writing
-        a new file or overwriting an existing file in text mode. 'a'
-        (or 'at') will append to an existing file in text mode or
-        create a new file if it does not already exist. See :py:func:`open`.
-    name_function : callable, default None
-        Function accepting an integer (partition index) and producing a
-        string to replace the asterisk in the given filename globstring.
-        Should preserve the lexicographic order of partitions. Not
-        supported when ``single_file`` is True.
-    compression : string, optional
-        A string representing the compression to use in the output file,
-        allowed values are 'gzip', 'bz2', 'xz',
-        only used when the first argument is a filename.
-    compute : bool, default True
-        If True, immediately executes. If False, returns a set of delayed
-        objects, which can be computed at a later time.
-    storage_options : dict
-        Parameters passed on to the backend filesystem class.
-    header_first_partition_only : bool, default None
-        If set to True, only write the header row in the first output
-        file. By default, headers are written to all partitions under
-        the multiple file mode (``single_file`` is False) and written
-        only once under the single file mode (``single_file`` is True).
-        It must be True under the single file mode.
-    compute_kwargs : dict, optional
-        Options to be passed in to the compute method
-    kwargs : dict, optional
-        Additional parameters to pass to :meth:`pandas.DataFrame.to_csv`.
-
-    Returns
-    -------
-    The names of the file written if they were computed right away.
-    If not, the delayed tasks associated with writing the files.
-
-    Raises
-    ------
-    ValueError
-        If ``header_first_partition_only`` is set to False or
-        ``name_function`` is specified when ``single_file`` is True.
-
-    See Also
-    --------
-    fsspec.open_files
-    """
-    from dask.dataframe.io.csv import to_csv as _to_csv
-
-    return _to_csv(
-        df.optimize(),
-        filename,
-        single_file=single_file,
-        encoding=encoding,
-        mode=mode,
-        name_function=name_function,
-        compression=compression,
-        compute=compute,
-        scheduler=scheduler,
-        storage_options=storage_options,
-        header_first_partition_only=header_first_partition_only,
-        compute_kwargs=compute_kwargs,
-        **kwargs,
-    )
diff --git a/dask/dataframe/dask_expr/io/hdf.py b/dask/dataframe/dask_expr/io/hdf.py
deleted file mode 100644
index 01af77e85..000000000
--- a/dask/dataframe/dask_expr/io/hdf.py
+++ /dev/null
@@ -1,143 +0,0 @@
-from __future__ import annotations
-
-
-def read_hdf(
-    pattern,
-    key,
-    start=0,
-    stop=None,
-    columns=None,
-    chunksize=1000000,
-    sorted_index=False,
-    lock=True,
-    mode="r",
-):
-    from dask.dataframe.io import read_hdf as _read_hdf
-
-    return _read_hdf(
-        pattern,
-        key,
-        start=start,
-        stop=stop,
-        columns=columns,
-        chunksize=chunksize,
-        sorted_index=sorted_index,
-        lock=lock,
-        mode=mode,
-    )
-
-
-def to_hdf(
-    df,
-    path,
-    key,
-    mode="a",
-    append=False,
-    scheduler=None,
-    name_function=None,
-    compute=True,
-    lock=None,
-    dask_kwargs=None,
-    **kwargs,
-):
-    """Store Dask Dataframe to Hierarchical Data Format (HDF) files
-
-    This is a parallel version of the Pandas function of the same name.  Please
-    see the Pandas docstring for more detailed information about shared keyword
-    arguments.
-
-    This function differs from the Pandas version by saving the many partitions
-    of a Dask DataFrame in parallel, either to many files, or to many datasets
-    within the same file.  You may specify this parallelism with an asterix
-    ``*`` within the filename or datapath, and an optional ``name_function``.
-    The asterix will be replaced with an increasing sequence of integers
-    starting from ``0`` or with the result of calling ``name_function`` on each
-    of those integers.
-
-    This function only supports the Pandas ``'table'`` format, not the more
-    specialized ``'fixed'`` format.
-
-    Parameters
-    ----------
-    path : string, pathlib.Path
-        Path to a target filename. Supports strings, ``pathlib.Path``, or any
-        object implementing the ``__fspath__`` protocol. May contain a ``*`` to
-        denote many filenames.
-    key : string
-        Datapath within the files.  May contain a ``*`` to denote many locations
-    name_function : function
-        A function to convert the ``*`` in the above options to a string.
-        Should take in a number from 0 to the number of partitions and return a
-        string. (see examples below)
-    compute : bool
-        Whether or not to execute immediately.  If False then this returns a
-        ``dask.Delayed`` value.
-    lock : bool, Lock, optional
-        Lock to use to prevent concurrency issues.  By default a
-        ``threading.Lock``, ``multiprocessing.Lock`` or ``SerializableLock``
-        will be used depending on your scheduler if a lock is required. See
-        dask.utils.get_scheduler_lock for more information about lock
-        selection.
-    scheduler : string
-        The scheduler to use, like "threads" or "processes"
-    **other:
-        See pandas.to_hdf for more information
-
-    Examples
-    --------
-    Save Data to a single file
-
-    >>> df.to_hdf('output.hdf', '/data')            # doctest: +SKIP
-
-    Save data to multiple datapaths within the same file:
-
-    >>> df.to_hdf('output.hdf', '/data-*')          # doctest: +SKIP
-
-    Save data to multiple files:
-
-    >>> df.to_hdf('output-*.hdf', '/data')          # doctest: +SKIP
-
-    Save data to multiple files, using the multiprocessing scheduler:
-
-    >>> df.to_hdf('output-*.hdf', '/data', scheduler='processes') # doctest: +SKIP
-
-    Specify custom naming scheme.  This writes files as
-    '2000-01-01.hdf', '2000-01-02.hdf', '2000-01-03.hdf', etc..
-
-    >>> from datetime import date, timedelta
-    >>> base = date(year=2000, month=1, day=1)
-    >>> def name_function(i):
-    ...     ''' Convert integer 0 to n to a string '''
-    ...     return base + timedelta(days=i)
-
-    >>> df.to_hdf('*.hdf', '/data', name_function=name_function) # doctest: +SKIP
-
-    Returns
-    -------
-    filenames : list
-        Returned if ``compute`` is True. List of file names that each partition
-        is saved to.
-    delayed : dask.Delayed
-        Returned if ``compute`` is False. Delayed object to execute ``to_hdf``
-        when computed.
-
-    See Also
-    --------
-    read_hdf:
-    to_parquet:
-    """
-    from dask.dataframe.io import to_hdf as _to_hdf
-
-    return _to_hdf(
-        df.optimize(),
-        path,
-        key,
-        mode=mode,
-        append=append,
-        scheduler=scheduler,
-        name_function=name_function,
-        compute=compute,
-        lock=lock,
-        dask_kwargs=dask_kwargs,
-        **kwargs,
-    )
diff --git a/dask/dataframe/dask_expr/io/json.py b/dask/dataframe/dask_expr/io/json.py
deleted file mode 100644
index f5ace1154..000000000
--- a/dask/dataframe/dask_expr/io/json.py
+++ /dev/null
@@ -1,187 +0,0 @@
-from __future__ import annotations
-
-import pandas as pd
-
-from dask.dataframe.dask_expr._backends import dataframe_creation_dispatch
-from dask.dataframe.utils import insert_meta_param_description
-
-
-@dataframe_creation_dispatch.register_inplace("pandas")
-@insert_meta_param_description
-def read_json(
-    url_path,
-    orient="records",
-    lines=None,
-    storage_options=None,
-    blocksize=None,
-    sample=2**20,
-    encoding="utf-8",
-    errors="strict",
-    compression="infer",
-    meta=None,
-    engine=pd.read_json,
-    include_path_column=False,
-    path_converter=None,
-    **kwargs,
-):
-    """Create a dataframe from a set of JSON files
-
-    This utilises ``pandas.read_json()``, and most parameters are
-    passed through - see its docstring.
-
-    Differences: orient is 'records' by default, with lines=True; this
-    is appropriate for line-delimited "JSON-lines" data, the kind of JSON output
-    that is most common in big-data scenarios, and which can be chunked when
-    reading (see ``read_json()``). All other options require blocksize=None,
-    i.e., one partition per input file.
-
-    Parameters
-    ----------
-    url_path: str, list of str
-        Location to read from. If a string, can include a glob character to
-        find a set of file names.
-        Supports protocol specifications such as ``"s3://"``.
-    encoding, errors:
-        The text encoding to implement, e.g., "utf-8" and how to respond
-        to errors in the conversion (see ``str.encode()``).
-    orient, lines, kwargs
-        passed to pandas; if not specified, lines=True when orient='records',
-        False otherwise.
-    storage_options: dict
-        Passed to backend file-system implementation
-    blocksize: None or int
-        If None, files are not blocked, and you get one partition per input
-        file. If int, which can only be used for line-delimited JSON files,
-        each partition will be approximately this size in bytes, to the nearest
-        newline character.
-    sample: int
-        Number of bytes to pre-load, to provide an empty dataframe structure
-        to any blocks without data. Only relevant when using blocksize.
-    encoding, errors:
-        Text conversion, ``see bytes.decode()``
-    compression : string or None
-        String like 'gzip' or 'xz'.
-    engine : callable or str, default ``pd.read_json``
-        The underlying function that dask will use to read JSON files. By
-        default, this will be the pandas JSON reader (``pd.read_json``).
-        If a string is specified, this value will be passed under the ``engine``
-        key-word argument to ``pd.read_json`` (only supported for pandas>=2.0).
-    include_path_column : bool or str, optional
-        Include a column with the file path where each row in the dataframe
-        originated. If ``True``, a new column is added to the dataframe called
-        ``path``. If ``str``, sets new column name. Default is ``False``.
-    path_converter : function or None, optional
-        A function that takes one argument and returns a string. Used to convert
-        paths in the ``path`` column, for instance, to strip a common prefix from
-        all the paths.
-    $META
-
-    Returns
-    -------
-    dask.DataFrame
-
-    Examples
-    --------
-    Load single file
-
-    >>> dd.read_json('myfile.1.json')  # doctest: +SKIP
-
-    Load multiple files
-
-    >>> dd.read_json('myfile.*.json')  # doctest: +SKIP
-
-    >>> dd.read_json(['myfile.1.json', 'myfile.2.json'])  # doctest: +SKIP
-
-    Load large line-delimited JSON files using partitions of approx
-    256MB size
-
-    >> dd.read_json('data/file*.csv', blocksize=2**28)
-    """
-    from dask.dataframe.io.json import read_json
-
-    return read_json(
-        url_path,
-        orient=orient,
-        lines=lines,
-        storage_options=storage_options,
-        blocksize=blocksize,
-        sample=sample,
-        encoding=encoding,
-        errors=errors,
-        compression=compression,
-        meta=meta,
-        engine=engine,
-        include_path_column=include_path_column,
-        path_converter=path_converter,
-        **kwargs,
-    )
-
-
-def to_json(
-    df,
-    url_path,
-    orient="records",
-    lines=None,
-    storage_options=None,
-    compute=True,
-    encoding="utf-8",
-    errors="strict",
-    compression=None,
-    compute_kwargs=None,
-    name_function=None,
-    **kwargs,
-):
-    """Write dataframe into JSON text files
-
-    This utilises ``pandas.DataFrame.to_json()``, and most parameters are
-    passed through - see its docstring.
-
-    Differences: orient is 'records' by default, with lines=True; this
-    produces the kind of JSON output that is most common in big-data
-    applications, and which can be chunked when reading (see ``read_json()``).
-
-    Parameters
-    ----------
-    df: dask.DataFrame
-        Data to save
-    url_path: str, list of str
-        Location to write to. If a string, and there are more than one
-        partitions in df, should include a glob character to expand into a
-        set of file names, or provide a ``name_function=`` parameter.
-        Supports protocol specifications such as ``"s3://"``.
-    encoding, errors:
-        The text encoding to implement, e.g., "utf-8" and how to respond
-        to errors in the conversion (see ``str.encode()``).
-    orient, lines, kwargs
-        passed to pandas; if not specified, lines=True when orient='records',
-        False otherwise.
-    storage_options: dict
-        Passed to backend file-system implementation
-    compute: bool
-        If true, immediately executes. If False, returns a set of delayed
-        objects, which can be computed at a later time.
-    compute_kwargs : dict, optional
-        Options to be passed in to the compute method
-    compression : string or None
-        String like 'gzip' or 'xz'.
-    name_function : callable, default None
-        Function accepting an integer (partition index) and producing a
-        string to replace the asterisk in the given filename globstring.
-        Should preserve the lexicographic order of partitions.
-    """
-    from dask.dataframe.io.json import to_json
-
-    return to_json(
-        df,
-        url_path,
-        orient=orient,
-        lines=lines,
-        storage_options=storage_options,
-        compute=compute,
-        encoding=encoding,
-        errors=errors,
-        compression=compression,
-        compute_kwargs=compute_kwargs,
-        name_function=name_function,
-        **kwargs,
-    )
diff --git a/dask/dataframe/dask_expr/io/orc.py b/dask/dataframe/dask_expr/io/orc.py
deleted file mode 100644
index 89243f426..000000000
--- a/dask/dataframe/dask_expr/io/orc.py
+++ /dev/null
@@ -1,82 +0,0 @@
-from __future__ import annotations
-
-from dask.dataframe.dask_expr._backends import dataframe_creation_dispatch
-
-
-@dataframe_creation_dispatch.register_inplace("pandas")
-def read_orc(
-    path,
-    engine="pyarrow",
-    columns=None,
-    index=None,
-    split_stripes=1,
-    aggregate_files=None,
-    storage_options=None,
-):
-    """Read dataframe from ORC file(s)
-
-    Parameters
-    ----------
-    path: str or list(str)
-        Location of file(s), which can be a full URL with protocol
-        specifier, and may include glob character if a single string.
-    engine: 'pyarrow' or ORCEngine
-        Backend ORC engine to use for I/O. Default is "pyarrow".
-    columns: None or list(str)
-        Columns to load. If None, loads all.
-    index: str
-        Column name to set as index.
-    split_stripes: int or False
-        Maximum number of ORC stripes to include in each output-DataFrame
-        partition. Use False to specify a 1-to-1 mapping between files
-        and partitions. Default is 1.
-    aggregate_files : bool, default False
-        Whether distinct file paths may be aggregated into the same output
-        partition. A setting of True means that any two file paths may be
-        aggregated into the same output partition, while False means that
-        inter-file aggregation is prohibited.
-    storage_options: None or dict
-        Further parameters to pass to the bytes backend.
-
-    Returns
-    -------
-    Dask.DataFrame (even if there is only one column)
-
-    Examples
-    --------
-    >>> df = dd.read_orc('https://github.com/apache/orc/raw/'
-    ...                  'master/examples/demo-11-zlib.orc')  # doctest: +SKIP
-    """
-    from dask.dataframe.io import read_orc as _read_orc
-
-    return _read_orc(
-        path,
-        engine=engine,
-        columns=columns,
-        index=index,
-        split_stripes=split_stripes,
-        aggregate_files=aggregate_files,
-        storage_options=storage_options,
-    )
-
-
-def to_orc(
-    df,
-    path,
-    engine="pyarrow",
-    write_index=True,
-    storage_options=None,
-    compute=True,
-    compute_kwargs=None,
-):
-    from dask.dataframe.io import to_orc as _to_orc
-
-    return _to_orc(
-        df.optimize(),
-        path,
-        engine=engine,
-        write_index=write_index,
-        storage_options=storage_options,
-        compute=compute,
-        compute_kwargs=compute_kwargs,
-    )
diff --git a/dask/dataframe/dask_expr/io/sql.py b/dask/dataframe/dask_expr/io/sql.py
deleted file mode 100644
index d783b2334..000000000
--- a/dask/dataframe/dask_expr/io/sql.py
+++ /dev/null
@@ -1,367 +0,0 @@
-from __future__ import annotations
-
-
-def read_sql(sql, con, index_col, **kwargs):
-    """
-    Read SQL query or database table into a DataFrame.
-
-    This function is a convenience wrapper around ``read_sql_table`` and
-    ``read_sql_query``. It will delegate to the specific function depending
-    on the provided input. A SQL query will be routed to ``read_sql_query``,
-    while a database table name will be routed to ``read_sql_table``.
-    Note that the delegated function might have more specific notes about
-    their functionality not listed here.
-
-    Parameters
-    ----------
-    sql : str or SQLAlchemy Selectable
-        Name of SQL table in database or SQL query to be executed. TextClause is not supported
-    con : str
-        Full sqlalchemy URI for the database connection
-    index_col : str
-        Column which becomes the index, and defines the partitioning. Should
-        be a indexed column in the SQL server, and any orderable type. If the
-        type is number or time, then partition boundaries can be inferred from
-        ``npartitions`` or ``bytes_per_chunk``; otherwise must supply explicit
-        ``divisions``.
-
-    Returns
-    -------
-    dask.dataframe
-
-    See Also
-    --------
-    read_sql_table : Read SQL database table into a DataFrame.
-    read_sql_query : Read SQL query into a DataFrame.
-    """
-    from dask.dataframe.io.sql import read_sql
-
-    return read_sql(sql, con, index_col, **kwargs)
-
-
-def read_sql_table(
-    table_name,
-    con,
-    index_col,
-    divisions=None,
-    npartitions=None,
-    limits=None,
-    columns=None,
-    bytes_per_chunk="256 MiB",
-    head_rows=5,
-    schema=None,
-    meta=None,
-    engine_kwargs=None,
-    **kwargs,
-):
-    """
-    Read SQL database table into a DataFrame.
-
-    If neither ``divisions`` or ``npartitions`` is given, the memory footprint of the
-    first few rows will be determined, and partitions of size ~256MB will
-    be used.
-
-    Parameters
-    ----------
-    table_name : str
-        Name of SQL table in database.
-    con : str
-        Full sqlalchemy URI for the database connection
-    index_col : str
-        Column which becomes the index, and defines the partitioning. Should
-        be a indexed column in the SQL server, and any orderable type. If the
-        type is number or time, then partition boundaries can be inferred from
-        ``npartitions`` or ``bytes_per_chunk``; otherwise must supply explicit
-        ``divisions``.
-    columns : sequence of str or SqlAlchemy column or None
-        Which columns to select; if None, gets all. Note can be a mix of str and SqlAlchemy columns
-    schema : str or None
-        Pass this to sqlalchemy to select which DB schema to use within the
-        URI connection
-    divisions: sequence
-        Values of the index column to split the table by. If given, this will
-        override ``npartitions`` and ``bytes_per_chunk``. The divisions are the value
-        boundaries of the index column used to define the partitions. For
-        example, ``divisions=list('acegikmoqsuwz')`` could be used to partition
-        a string column lexographically into 12 partitions, with the implicit
-        assumption that each partition contains similar numbers of records.
-    npartitions : int
-        Number of partitions, if ``divisions`` is not given. Will split the values
-        of the index column linearly between ``limits``, if given, or the column
-        max/min. The index column must be numeric or time for this to work
-    limits: 2-tuple or None
-        Manually give upper and lower range of values for use with ``npartitions``;
-        if None, first fetches max/min from the DB. Upper limit, if
-        given, is inclusive.
-    bytes_per_chunk : str or int
-        If both ``divisions`` and ``npartitions`` is None, this is the target size of
-        each partition, in bytes
-    head_rows : int
-        How many rows to load for inferring the data-types, and memory per row
-    meta : empty DataFrame or None
-        If provided, do not attempt to infer dtypes, but use these, coercing
-        all chunks on load
-    engine_kwargs : dict or None
-        Specific db engine parameters for sqlalchemy
-    kwargs : dict
-        Additional parameters to pass to `pd.read_sql()`
-
-    Returns
-    -------
-    dask.dataframe
-
-    See Also
-    --------
-    read_sql_query : Read SQL query into a DataFrame.
-
-    Examples
-    --------
-    >>> df = dd.read_sql_table('accounts', 'sqlite:///path/to/bank.db',
-    ...                  npartitions=10, index_col='id')  # doctest: +SKIP
-    """
-    from dask.dataframe.io.sql import read_sql_table as _read_sql_table
-
-    return _read_sql_table(
-        table_name,
-        con,
-        index_col,
-        divisions=divisions,
-        npartitions=npartitions,
-        limits=limits,
-        columns=columns,
-        bytes_per_chunk=bytes_per_chunk,
-        head_rows=head_rows,
-        schema=schema,
-        meta=meta,
-        engine_kwargs=engine_kwargs,
-        **kwargs,
-    )
-
-
-def read_sql_query(
-    sql,
-    con,
-    index_col,
-    divisions=None,
-    npartitions=None,
-    limits=None,
-    bytes_per_chunk="256 MiB",
-    head_rows=5,
-    meta=None,
-    engine_kwargs=None,
-    **kwargs,
-):
-    """
-    Read SQL query into a DataFrame.
-
-    If neither ``divisions`` or ``npartitions`` is given, the memory footprint of the
-    first few rows will be determined, and partitions of size ~256MB will
-    be used.
-
-    Parameters
-    ----------
-    sql : SQLAlchemy Selectable
-        SQL query to be executed. TextClause is not supported
-    con : str
-        Full sqlalchemy URI for the database connection
-    index_col : str
-        Column which becomes the index, and defines the partitioning. Should
-        be a indexed column in the SQL server, and any orderable type. If the
-        type is number or time, then partition boundaries can be inferred from
-        ``npartitions`` or ``bytes_per_chunk``; otherwise must supply explicit
-        ``divisions``.
-    divisions: sequence
-        Values of the index column to split the table by. If given, this will
-        override ``npartitions`` and ``bytes_per_chunk``. The divisions are the value
-        boundaries of the index column used to define the partitions. For
-        example, ``divisions=list('acegikmoqsuwz')`` could be used to partition
-        a string column lexographically into 12 partitions, with the implicit
-        assumption that each partition contains similar numbers of records.
-    npartitions : int
-        Number of partitions, if ``divisions`` is not given. Will split the values
-        of the index column linearly between ``limits``, if given, or the column
-        max/min. The index column must be numeric or time for this to work
-    limits: 2-tuple or None
-        Manually give upper and lower range of values for use with ``npartitions``;
-        if None, first fetches max/min from the DB. Upper limit, if
-        given, is inclusive.
-    bytes_per_chunk : str or int
-        If both ``divisions`` and ``npartitions`` is None, this is the target size of
-        each partition, in bytes
-    head_rows : int
-        How many rows to load for inferring the data-types, and memory per row
-    meta : empty DataFrame or None
-        If provided, do not attempt to infer dtypes, but use these, coercing
-        all chunks on load
-    engine_kwargs : dict or None
-        Specific db engine parameters for sqlalchemy
-    kwargs : dict
-        Additional parameters to pass to `pd.read_sql()`
-
-    Returns
-    -------
-    dask.dataframe
-
-    See Also
-    --------
-    read_sql_table : Read SQL database table into a DataFrame.
-    """
-    from dask.dataframe.io.sql import read_sql_query as _read_sql_query
-
-    return _read_sql_query(
-        sql,
-        con,
-        index_col,
-        divisions=divisions,
-        npartitions=npartitions,
-        limits=limits,
-        bytes_per_chunk=bytes_per_chunk,
-        head_rows=head_rows,
-        meta=meta,
-        engine_kwargs=engine_kwargs,
-        **kwargs,
-    )
-
-
-def to_sql(
-    df,
-    name: str,
-    uri: str,
-    schema=None,
-    if_exists: str = "fail",
-    index: bool = True,
-    index_label=None,
-    chunksize=None,
-    dtype=None,
-    method=None,
-    compute=True,
-    parallel=False,
-    engine_kwargs=None,
-):
-    """Store Dask Dataframe to a SQL table
-
-    An empty table is created based on the "meta" DataFrame (and conforming to the caller's "if_exists" preference), and
-    then each block calls pd.DataFrame.to_sql (with `if_exists="append"`).
-
-    Databases supported by SQLAlchemy [1]_ are supported. Tables can be
-    newly created, appended to, or overwritten.
-
-    Parameters
-    ----------
-    name : str
-        Name of SQL table.
-    uri : string
-        Full sqlalchemy URI for the database connection
-    schema : str, optional
-        Specify the schema (if database flavor supports this). If None, use
-        default schema.
-    if_exists : {'fail', 'replace', 'append'}, default 'fail'
-        How to behave if the table already exists.
-
-        * fail: Raise a ValueError.
-        * replace: Drop the table before inserting new values.
-        * append: Insert new values to the existing table.
-
-    index : bool, default True
-        Write DataFrame index as a column. Uses `index_label` as the column
-        name in the table.
-    index_label : str or sequence, default None
-        Column label for index column(s). If None is given (default) and
-        `index` is True, then the index names are used.
-        A sequence should be given if the DataFrame uses MultiIndex.
-    chunksize : int, optional
-        Specify the number of rows in each batch to be written at a time.
-        By default, all rows will be written at once.
-    dtype : dict or scalar, optional
-        Specifying the datatype for columns. If a dictionary is used, the
-        keys should be the column names and the values should be the
-        SQLAlchemy types or strings for the sqlite3 legacy mode. If a
-        scalar is provided, it will be applied to all columns.
-    method : {None, 'multi', callable}, optional
-        Controls the SQL insertion clause used:
-
-        * None : Uses standard SQL ``INSERT`` clause (one per row).
-        * 'multi': Pass multiple values in a single ``INSERT`` clause.
-        * callable with signature ``(pd_table, conn, keys, data_iter)``.
-
-        Details and a sample callable implementation can be found in the
-        section :ref:`insert method <io.sql.method>`.
-    compute : bool, default True
-        When true, call dask.compute and perform the load into SQL; otherwise, return a Dask object (or array of
-        per-block objects when parallel=True)
-    parallel : bool, default False
-        When true, have each block append itself to the DB table concurrently. This can result in DB rows being in a
-        different order than the source DataFrame's corresponding rows. When false, load each block into the SQL DB in
-        sequence.
-    engine_kwargs : dict or None
-        Specific db engine parameters for sqlalchemy
-
-    Raises
-    ------
-    ValueError
-        When the table already exists and `if_exists` is 'fail' (the
-        default).
-
-    See Also
-    --------
-    read_sql : Read a DataFrame from a table.
-
-    Notes
-    -----
-    Timezone aware datetime columns will be written as
-    ``Timestamp with timezone`` type with SQLAlchemy if supported by the
-    database. Otherwise, the datetimes will be stored as timezone unaware
-    timestamps local to the original timezone.
-
-    .. versionadded:: 0.24.0
-
-    References
-    ----------
-    .. [1] https://docs.sqlalchemy.org
-    .. [2] https://www.python.org/dev/peps/pep-0249/
-
-    Examples
-    --------
-    Create a table from scratch with 4 rows.
-
-    >>> import pandas as pd
-    >>> import dask.dataframe as dd
-    >>> df = pd.DataFrame([ {'i':i, 's':str(i)*2 } for i in range(4) ])
-    >>> ddf = dd.from_pandas(df, npartitions=2)
-    >>> ddf  # doctest: +SKIP
-    Dask DataFrame Structure:
-                       i       s
-    npartitions=2
-    0              int64  object
-    2                ...     ...
-    3                ...     ...
-    Dask Name: from_pandas, 2 tasks
-
-    >>> from dask.utils import tmpfile
-    >>> from sqlalchemy import create_engine, text
-    >>> with tmpfile() as f:
-    ...     db = 'sqlite:///%s' %f
-    ...     ddf.to_sql('test', db)
-    ...     engine = create_engine(db, echo=False)
-    ...     with engine.connect() as conn:
-    ...         result = conn.execute(text("SELECT * FROM test")).fetchall()
-    >>> result
-    [(0, 0, '00'), (1, 1, '11'), (2, 2, '22'), (3, 3, '33')]
-    """
-    from dask.dataframe.io.sql import to_sql as _to_sql
-
-    return _to_sql(
-        df,
-        name=name,
-        uri=uri,
-        schema=schema,
-        if_exists=if_exists,
-        index=index,
-        index_label=index_label,
-        chunksize=chunksize,
-        dtype=dtype,
-        method=method,
-        compute=compute,
-        parallel=parallel,
-        engine_kwargs=engine_kwargs,
-    )
diff --git a/dask/dataframe/io/csv.py b/dask/dataframe/io/csv.py
index c172fa5f9..4dc9ad8a0 100644
--- a/dask/dataframe/io/csv.py
+++ b/dask/dataframe/io/csv.py
@@ -550,6 +550,7 @@ def read_pandas(
     # rows to skip.
     names = kwargs.get("names", None)
     header = kwargs.get("header", "infer" if names is None else None)
+    kwargs["header"] = header
     need = 1 if header is None else 2
     if isinstance(header, int):
         firstrow += header
@@ -891,6 +892,8 @@ def to_csv(
     --------
     fsspec.open_files
     """
+    df = df.optimize()
+
     if single_file and name_function is not None:
         raise ValueError("name_function is not supported under the single file mode")
     if header_first_partition_only is None:
diff --git a/dask/dataframe/io/hdf.py b/dask/dataframe/io/hdf.py
index 58ed97a90..2d77e9c99 100644
--- a/dask/dataframe/io/hdf.py
+++ b/dask/dataframe/io/hdf.py
@@ -135,6 +135,7 @@ def to_hdf(
     read_hdf:
     to_parquet:
     """
+    df = df.optimize()
     if dask_kwargs is None:
         dask_kwargs = {}
 
diff --git a/dask/dataframe/io/orc/core.py b/dask/dataframe/io/orc/core.py
index 20496a3ad..f562077f7 100644
--- a/dask/dataframe/io/orc/core.py
+++ b/dask/dataframe/io/orc/core.py
@@ -165,6 +165,7 @@ def to_orc(
     --------
     read_orc: Read ORC data to dask.dataframe
     """
+    df = df.optimize()
 
     # Get engine
     engine = _get_engine(engine)
diff --git a/dask/dataframe/io/parquet/arrow.py b/dask/dataframe/io/parquet/arrow.py
index 458f1e790..9c231f9bd 100644
--- a/dask/dataframe/io/parquet/arrow.py
+++ b/dask/dataframe/io/parquet/arrow.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import itertools
 import json
 import operator
 import textwrap
@@ -284,9 +285,19 @@ def _get_rg_statistics(row_group, col_names):
     statistics for all columns.
     """
 
-    row_group_schema = {
-        col_name: i for i, col_name in enumerate(row_group.schema.names)
-    }
+    row_group_schema = dict(
+        zip(
+            row_group.schema.names,
+            itertools.accumulate(
+                [
+                    # Need to account for multi-field struct columns
+                    max(row_group.schema.types[i].num_fields, 1)
+                    for i in range(len(row_group.schema.names) - 1)
+                ],
+                initial=0,
+            ),
+        )
+    )
 
     def name_stats(column_name):
         col = row_group.metadata.column(row_group_schema[column_name])
diff --git a/dask/dataframe/tseries/resample.py b/dask/dataframe/tseries/resample.py
index 3352d1167..2ca1bcddb 100644
--- a/dask/dataframe/tseries/resample.py
+++ b/dask/dataframe/tseries/resample.py
@@ -1,8 +1,26 @@
 from __future__ import annotations
 
+import functools
+from collections import namedtuple
+
+import numpy as np
 import pandas as pd
+from pandas.core.resample import Resampler as pd_Resampler
 
 from dask.dataframe import methods
+from dask.dataframe.dask_expr._collection import new_collection
+from dask.dataframe.dask_expr._expr import (
+    Blockwise,
+    Expr,
+    Projection,
+    make_meta,
+    plain_column_projection,
+)
+from dask.dataframe.dask_expr._repartition import Repartition
+from dask.dataframe.dispatch import meta_nonempty
+from dask.utils import derived_from
+
+BlockwiseDep = namedtuple(typename="BlockwiseDep", field_names=["iterable"])  # type: ignore
 
 
 def _resample_series(
@@ -83,3 +101,274 @@ def _resample_bin_and_out_divs(divisions, rule, closed="left", label="left"):
             setter(outdivs, temp.index[-1])
 
     return tuple(map(pd.Timestamp, newdivs)), tuple(map(pd.Timestamp, outdivs))
+
+
+class ResampleReduction(Expr):
+    _parameters = [
+        "frame",
+        "rule",
+        "kwargs",
+        "how_args",
+        "how_kwargs",
+    ]
+    _defaults = {
+        "closed": None,
+        "label": None,
+        "kwargs": None,
+        "how_args": (),
+        "how_kwargs": None,
+    }
+    how: str | None = None
+    fill_value = np.nan
+
+    @functools.cached_property
+    def npartitions(self):
+        return self.frame.npartitions
+
+    def _divisions(self):
+        return self._resample_divisions[1]
+
+    @functools.cached_property
+    def _meta(self):
+        resample = meta_nonempty(self.frame._meta).resample(self.rule, **self.kwargs)
+        meta = getattr(resample, self.how)(*self.how_args, **self.how_kwargs or {})
+        return make_meta(meta)
+
+    @functools.cached_property
+    def kwargs(self):
+        return {} if self.operand("kwargs") is None else self.operand("kwargs")
+
+    @functools.cached_property
+    def how_kwargs(self):
+        return {} if self.operand("how_kwargs") is None else self.operand("how_kwargs")
+
+    @functools.cached_property
+    def _resample_divisions(self):
+        return _resample_bin_and_out_divs(
+            self.frame.divisions, self.rule, **self.kwargs or {}
+        )
+
+    def _simplify_up(self, parent, dependents):
+        if isinstance(parent, Projection):
+            return plain_column_projection(self, parent, dependents)
+
+    def _lower(self):
+        partitioned = Repartition(
+            self.frame, new_divisions=self._resample_divisions[0], force=True
+        )
+        output_divisions = self._resample_divisions[1]
+        return ResampleAggregation(
+            partitioned,
+            BlockwiseDep(output_divisions[:-1]),
+            BlockwiseDep(output_divisions[1:]),
+            BlockwiseDep(["left"] * (len(output_divisions[1:]) - 1) + [None]),
+            self.rule,
+            self.kwargs,
+            self.how,
+            self.fill_value,
+            list(self.how_args),
+            self.how_kwargs,
+        )
+
+
+class ResampleAggregation(Blockwise):
+    _parameters = [
+        "frame",
+        "divisions_left",
+        "divisions_right",
+        "closed",
+        "rule",
+        "kwargs",
+        "how",
+        "fill_value",
+        "how_args",
+        "how_kwargs",
+    ]
+    operation = staticmethod(_resample_series)
+
+    @functools.cached_property
+    def _meta(self):
+        return self.frame._meta
+
+    def _divisions(self):
+        return list(self.divisions_left.iterable) + [self.divisions_right.iterable[-1]]
+
+    def _blockwise_arg(self, arg, i):
+        if isinstance(arg, BlockwiseDep):
+            return arg.iterable[i]
+        return super()._blockwise_arg(arg, i)
+
+
+class ResampleCount(ResampleReduction):
+    how = "count"
+    fill_value = 0
+
+
+class ResampleSum(ResampleReduction):
+    how = "sum"
+    fill_value = 0
+
+
+class ResampleProd(ResampleReduction):
+    how = "prod"
+
+
+class ResampleMean(ResampleReduction):
+    how = "mean"
+
+
+class ResampleMin(ResampleReduction):
+    how = "min"
+
+
+class ResampleMax(ResampleReduction):
+    how = "max"
+
+
+class ResampleFirst(ResampleReduction):
+    how = "first"
+
+
+class ResampleLast(ResampleReduction):
+    how = "last"
+
+
+class ResampleVar(ResampleReduction):
+    how = "var"
+
+
+class ResampleStd(ResampleReduction):
+    how = "std"
+
+
+class ResampleSize(ResampleReduction):
+    how = "size"
+    fill_value = 0
+
+
+class ResampleNUnique(ResampleReduction):
+    how = "nunique"
+    fill_value = 0
+
+
+class ResampleMedian(ResampleReduction):
+    how = "median"
+
+
+class ResampleQuantile(ResampleReduction):
+    how = "quantile"
+
+
+class ResampleOhlc(ResampleReduction):
+    how = "ohlc"
+
+
+class ResampleSem(ResampleReduction):
+    how = "sem"
+
+
+class ResampleAgg(ResampleReduction):
+    how = "agg"
+
+    def _simplify_up(self, parent, dependents):
+        # Disable optimization in `agg`; function may access other columns
+        return
+
+
+class Resampler:
+    """Aggregate using one or more operations
+
+    The purpose of this class is to expose an API similar
+    to Pandas' `Resampler` for dask-expr
+    """
+
+    def __init__(self, obj, rule, **kwargs):
+        if obj.divisions[0] is None:
+            msg = (
+                "Can only resample dataframes with known divisions\n"
+                "See https://docs.dask.org/en/latest/dataframe-design.html#partitions\n"
+                "for more information."
+            )
+            raise ValueError(msg)
+        self.obj = obj
+        self.rule = rule
+        self.kwargs = kwargs
+
+    def _single_agg(self, expr_cls, how_args=(), how_kwargs=None):
+        return new_collection(
+            expr_cls(
+                self.obj,
+                self.rule,
+                self.kwargs,
+                how_args=how_args,
+                how_kwargs=how_kwargs,
+            )
+        )
+
+    @derived_from(pd_Resampler)
+    def count(self):
+        return self._single_agg(ResampleCount)
+
+    @derived_from(pd_Resampler)
+    def sum(self):
+        return self._single_agg(ResampleSum)
+
+    @derived_from(pd_Resampler)
+    def prod(self):
+        return self._single_agg(ResampleProd)
+
+    @derived_from(pd_Resampler)
+    def mean(self):
+        return self._single_agg(ResampleMean)
+
+    @derived_from(pd_Resampler)
+    def min(self):
+        return self._single_agg(ResampleMin)
+
+    @derived_from(pd_Resampler)
+    def max(self):
+        return self._single_agg(ResampleMax)
+
+    @derived_from(pd_Resampler)
+    def first(self):
+        return self._single_agg(ResampleFirst)
+
+    @derived_from(pd_Resampler)
+    def last(self):
+        return self._single_agg(ResampleLast)
+
+    @derived_from(pd_Resampler)
+    def var(self):
+        return self._single_agg(ResampleVar)
+
+    @derived_from(pd_Resampler)
+    def std(self):
+        return self._single_agg(ResampleStd)
+
+    @derived_from(pd_Resampler)
+    def size(self):
+        return self._single_agg(ResampleSize)
+
+    @derived_from(pd_Resampler)
+    def nunique(self):
+        return self._single_agg(ResampleNUnique)
+
+    @derived_from(pd_Resampler)
+    def median(self):
+        return self._single_agg(ResampleMedian)
+
+    @derived_from(pd_Resampler)
+    def quantile(self):
+        return self._single_agg(ResampleQuantile)
+
+    @derived_from(pd_Resampler)
+    def ohlc(self):
+        return self._single_agg(ResampleOhlc)
+
+    @derived_from(pd_Resampler)
+    def sem(self):
+        return self._single_agg(ResampleSem)
+
+    @derived_from(pd_Resampler)
+    def agg(self, func, *args, **kwargs):
+        return self._single_agg(ResampleAgg, how_args=(func, *args), how_kwargs=kwargs)
diff --git a/dask/dataframe/dask_expr/tests/test_resample.py b/dask/dataframe/tseries/tests/test_resample_expr.py
similarity index 100%
rename from dask/dataframe/dask_expr/tests/test_resample.py
rename to dask/dataframe/tseries/tests/test_resample_expr.py
diff --git a/docs/source/conf.py b/docs/source/conf.py
index 903d533f1..b68225158 100644
--- a/docs/source/conf.py
+++ b/docs/source/conf.py
@@ -415,6 +415,46 @@ redirect_files = [
     ("how-to/setup-prometheus.html", "prometheus.html"),
 ]
 
+for func in [
+    "aggregate",
+    "apply",
+    "bfill",
+    "count",
+    "cumcount",
+    "cumprod",
+    "cumsum",
+    "fillna",
+    "ffill",
+    "get_group",
+    "max",
+    "mean",
+    "min",
+    "size",
+    "std",
+    "sum",
+    "var",
+    "cov",
+    "corr",
+    "first",
+    "last",
+    "idxmin",
+    "idxmax",
+    "rolling",
+    "transform",
+]:
+    redirect_files.append(
+        (
+            f"generated/dask.dataframe.groupby.DataFrameGroupBy.{func}.html",
+            f"generated/dask.dataframe.api.GroupBy.{func}.html",
+        )
+    )
+    redirect_files.append(
+        (
+            f"generated/dask.dataframe.groupby.SeriesGroupBy.{func}.html",
+            f"generated/dask.dataframe.api.SeriesGroupBy.{func}.html",
+        )
+    )
+
 
 redirect_template = """\
 <html>
diff --git a/docs/source/dataframe-api.rst b/docs/source/dataframe-api.rst
index 3093d1a34..a5f74762b 100644
--- a/docs/source/dataframe-api.rst
+++ b/docs/source/dataframe-api.rst
@@ -369,12 +369,146 @@ Accessors
 Similar to pandas, Dask provides dtype-specific methods under various accessors.
 These are separate namespaces within :class:`Series` that only apply to specific data types.
 
-The accessor implementations are consistent with the current Dask DataFrame implementation.
+Datetime Accessor
+*****************
+
+**Methods**
+
+.. autosummary::
+   :toctree: generated/
+   :template: autosummary/accessor_method.rst
+
+   Series.dt.ceil
+   Series.dt.floor
+   Series.dt.isocalendar
+   Series.dt.normalize
+   Series.dt.round
+   Series.dt.strftime
+
+**Attributes**
+
+.. autosummary::
+   :toctree: generated/
+   :template: autosummary/accessor_attribute.rst
+
+   Series.dt.date
+   Series.dt.day
+   Series.dt.dayofweek
+   Series.dt.dayofyear
+   Series.dt.daysinmonth
+   Series.dt.freq
+   Series.dt.hour
+   Series.dt.microsecond
+   Series.dt.minute
+   Series.dt.month
+   Series.dt.nanosecond
+   Series.dt.quarter
+   Series.dt.second
+   Series.dt.time
+   Series.dt.timetz
+   Series.dt.tz
+   Series.dt.week
+   Series.dt.weekday
+   Series.dt.weekofyear
+   Series.dt.year
+
+String Accessor
+***************
+
+**Methods**
+
+.. autosummary::
+   :toctree: generated/
+   :template: autosummary/accessor_method.rst
+
+   Series.str.capitalize
+   Series.str.casefold
+   Series.str.cat
+   Series.str.center
+   Series.str.contains
+   Series.str.count
+   Series.str.decode
+   Series.str.encode
+   Series.str.endswith
+   Series.str.extract
+   Series.str.extractall
+   Series.str.find
+   Series.str.findall
+   Series.str.fullmatch
+   Series.str.get
+   Series.str.index
+   Series.str.isalnum
+   Series.str.isalpha
+   Series.str.isdecimal
+   Series.str.isdigit
+   Series.str.islower
+   Series.str.isnumeric
+   Series.str.isspace
+   Series.str.istitle
+   Series.str.isupper
+   Series.str.join
+   Series.str.len
+   Series.str.ljust
+   Series.str.lower
+   Series.str.lstrip
+   Series.str.match
+   Series.str.normalize
+   Series.str.pad
+   Series.str.partition
+   Series.str.repeat
+   Series.str.replace
+   Series.str.rfind
+   Series.str.rindex
+   Series.str.rjust
+   Series.str.rpartition
+   Series.str.rsplit
+   Series.str.rstrip
+   Series.str.slice
+   Series.str.split
+   Series.str.startswith
+   Series.str.strip
+   Series.str.swapcase
+   Series.str.title
+   Series.str.translate
+   Series.str.upper
+   Series.str.wrap
+   Series.str.zfill
+
+Categorical Accessor
+********************
+
+**Methods**
+
+.. autosummary::
+   :toctree: generated/
+   :template: autosummary/accessor_method.rst
+
+   Series.cat.add_categories
+   Series.cat.as_known
+   Series.cat.as_ordered
+   Series.cat.as_unknown
+   Series.cat.as_unordered
+   Series.cat.remove_categories
+   Series.cat.remove_unused_categories
+   Series.cat.rename_categories
+   Series.cat.reorder_categories
+   Series.cat.set_categories
+
+**Attributes**
+
+.. autosummary::
+   :toctree: generated/
+   :template: autosummary/accessor_attribute.rst
+
+   Series.cat.categories
+   Series.cat.codes
+   Series.cat.known
+   Series.cat.ordered
 
 Groupby Operations
 ~~~~~~~~~~~~~~~~~~
 
-.. currentmodule:: dask.dataframe.dask_expr._groupby
+.. currentmodule:: dask.dataframe.api
 
 DataFrame Groupby
 *****************
@@ -441,6 +575,8 @@ Series Groupby
 Custom Aggregation
 ******************
 
+.. currentmodule:: dask.dataframe
+
 .. autosummary::
    :toctree: generated/
 
@@ -449,15 +585,13 @@ Custom Aggregation
 Rolling Operations
 ~~~~~~~~~~~~~~~~~~
 
-.. currentmodule:: dask.dataframe
-
 .. autosummary::
    :toctree: generated/
 
    Series.rolling
    DataFrame.rolling
 
-.. currentmodule:: dask.dataframe.dask_expr._rolling
+.. currentmodule:: dask.dataframe.api
 
 .. autosummary::
    :toctree: generated/
@@ -479,7 +613,7 @@ Rolling Operations
 Create DataFrames
 ~~~~~~~~~~~~~~~~~
 
-.. currentmodule:: dask.dataframe.dask_expr
+.. currentmodule:: dask.dataframe
 
 .. autosummary::
    :toctree: generated/
@@ -541,7 +675,7 @@ Reshape DataFrames
 Concatenate DataFrames
 ~~~~~~~~~~~~~~~~~~~~~~
 
-.. currentmodule:: dask.dataframe.dask_expr._collection
+.. currentmodule:: dask.dataframe
 
 .. autosummary::
    :toctree: generated/
@@ -555,7 +689,7 @@ Concatenate DataFrames
 Resampling
 ~~~~~~~~~~
 
-.. currentmodule:: dask.dataframe.dask_expr._resample
+.. currentmodule:: dask.dataframe.tseries.resample
 
 .. autosummary::
    :toctree: generated/


Test patch:
diff --git a/dask/array/tests/test_creation.py b/dask/array/tests/test_creation.py
index 7b6f947c8..d7687e43f 100644
--- a/dask/array/tests/test_creation.py
+++ b/dask/array/tests/test_creation.py
@@ -988,6 +988,22 @@ def test_nan_zeros_ones_like(fn, shape_chunks, dtype):
     )
 
 
+def test_from_array_getitem_fused():
+    arr = np.arange(100).reshape(10, 10)
+    darr = da.from_array(arr, chunks=(5, 5))
+    result = darr[slice(1, 5), :][slice(1, 3), :]
+    dsk = collections_to_dsk([result])
+    # Ensure that slices are merged properly
+    key = [k for k in dsk if "array-getitem" in k[0]][0]
+    key_2 = [
+        k
+        for k, v in dsk[key].args[0].items()
+        if "getitem" in k[0] and not isinstance(v, Alias)
+    ][0]
+    assert dsk[key].args[0][key_2].args[1] == ((slice(2, 4), slice(0, None)))
+    assert_eq(result, arr[slice(1, 5), :][slice(1, 3), :])
+
+
 @pytest.mark.parametrize("shape_chunks", [((50, 4), (10, 2)), ((50,), (10,))])
 @pytest.mark.parametrize("dtype", ["u4", np.float32, None, np.int64])
 def test_nan_empty_like(shape_chunks, dtype):
diff --git a/dask/dataframe/dask_expr/io/tests/test_io.py b/dask/dataframe/dask_expr/io/tests/test_io.py
index 967f61f53..157f3dc3a 100644
--- a/dask/dataframe/dask_expr/io/tests/test_io.py
+++ b/dask/dataframe/dask_expr/io/tests/test_io.py
@@ -11,6 +11,7 @@ import dask.array as da
 import dask.dataframe as dd
 from dask import config
 from dask.array.utils import assert_eq as array_assert_eq
+from dask.dataframe import read_csv
 from dask.dataframe.dask_expr import (
     DataFrame,
     from_array,
@@ -19,7 +20,6 @@ from dask.dataframe.dask_expr import (
     from_map,
     from_pandas,
     optimize,
-    read_csv,
     read_parquet,
 )
 from dask.dataframe.dask_expr._expr import Replace
diff --git a/dask/dataframe/dask_expr/io/tests/test_sql.py b/dask/dataframe/dask_expr/io/tests/test_sql.py
index dec86b84f..789c12c52 100644
--- a/dask/dataframe/dask_expr/io/tests/test_sql.py
+++ b/dask/dataframe/dask_expr/io/tests/test_sql.py
@@ -2,7 +2,8 @@ from __future__ import annotations
 
 import pytest
 
-from dask.dataframe.dask_expr import from_pandas, read_sql_table
+from dask.dataframe import read_sql_table
+from dask.dataframe.dask_expr import from_pandas
 from dask.dataframe.dask_expr.tests._util import _backend_library, assert_eq
 from dask.utils import tmpfile
 
diff --git a/dask/dataframe/dask_expr/tests/test_merge_asof.py b/dask/dataframe/dask_expr/tests/test_merge_asof.py
index 70de2f252..745232cd3 100644
--- a/dask/dataframe/dask_expr/tests/test_merge_asof.py
+++ b/dask/dataframe/dask_expr/tests/test_merge_asof.py
@@ -24,6 +24,26 @@ def test_merge_asof_indexed():
     assert_eq(c, C)
 
 
+def test_merge_asof_indexed_optimiser():
+    A = pd.DataFrame(
+        {"left_val": list("abcd" * 3), "foo": 1},
+        index=[1, 3, 7, 9, 10, 13, 14, 17, 20, 24, 25, 28],
+    )
+    a = from_pandas(A, npartitions=4)
+    B = pd.DataFrame(
+        {"right_val": list("xyz" * 4), "bar": 2},
+        index=[1, 2, 3, 6, 7, 10, 12, 14, 16, 19, 23, 26],
+    )
+    b = from_pandas(B, npartitions=3)
+
+    C = pd.merge_asof(A, B, left_index=True, right_index=True)
+    c = merge_asof(a, b, left_index=True, right_index=True)
+    c = c.optimize(fuse=False)
+    c = c[["left_val", "right_val"]]
+
+    assert_eq(c, C[["left_val", "right_val"]])
+
+
 def test_merge_asof_on_basic():
     A = pd.DataFrame({"a": [1, 5, 10], "left_val": ["a", "b", "c"]})
     a = from_pandas(A, npartitions=2)
diff --git a/dask/dataframe/io/tests/test_parquet.py b/dask/dataframe/io/tests/test_parquet.py
index 3cba1bda6..75e445db9 100644
--- a/dask/dataframe/io/tests/test_parquet.py
+++ b/dask/dataframe/io/tests/test_parquet.py
@@ -919,6 +919,48 @@ def test_append_dict_column(tmpdir):
     assert_eq(expect, result)
 
 
+@PYARROW_MARK
+def test_filter_with_struct_column(tmpdir):
+    # Check filtering on a table containing
+    # a multi-field struct column.
+    table = pa.Table.from_arrays(
+        [
+            pa.array(
+                [
+                    {"subfield1": 10, "subfield2": 12},
+                    {"subfield1": 20, "subfield2": 12},
+                    {"subfield1": 30, "subfield2": 12},
+                ]
+            ),
+            pa.array(["aa", "bb", "bb"]),
+        ],
+        schema=pa.schema(
+            [
+                (
+                    "nested_column",
+                    pa.struct([("subfield1", pa.int32()), ("subfield2", pa.int32())]),
+                ),
+                ("id", pa.string()),
+            ]
+        ),
+    )
+    fn = str(tmpdir) + "file.parq"
+    pq.write_table(table, fn)
+
+    pdf = table.to_pandas()
+    assert_eq(dd.read_parquet(fn), pdf)
+    assert_eq(
+        dd.read_parquet(fn, filters=[("id", "not in", ["bb"])]),
+        pdf[pdf["id"] != "bb"],
+        check_index=False,
+    )
+    assert_eq(
+        dd.read_parquet(fn, filters=[("id", "in", ["bb"])]),
+        pdf[pdf["id"] == "bb"],
+        check_index=False,
+    )
+
+
 def test_ordering(tmpdir, write_engine, read_engine):
     tmp = str(tmpdir)
     df = pd.DataFrame(