Repository: dask/dask
Pull request #11638 (2025-01-07 13:41:06)

Reintroduce slice fusion
- [ ] Closes #xxxx
- [ ] Tests added / passed
- [ ] Passes `pre-commit run --all-files`

This is mostly what the old implementation did...

I still have to add more docs and probably more tests too, but the general idea is there. We are only looking at fused tasks,
since they might have the pattern we care about that caused https://github.com/pydata/xarray/issues/9926

The way we are currently modifying these fused tasks isn't great though, we are using the dictionary that defines the fused task and replace fused getitem tasks with an alias before modifying the parent task with the new slice, i.e.

```

inner_graph = {
    "a": Task("a", getitem, TaskRef("original_open_dataset"), slice(0, 100)),
    "b": Task("b", getitem, TaskRef("a"), slice(30, 50)),
}

# translates to

inner_graph = {
    "a": Task("a", getitem, TaskRef("original_open_dataset"), slice(30, 50)),
    "b": Alias("b", "a"),
}
```

We might want to consider adding an API for this instead of modifying the dictionary inplace? cc @fjetter for thoughts?

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   3h 54m 47s ⏱️ + 3m 53s
 12 613 tests + 1   11 450 ✅ + 1   1 163 💤 ±0  0 ❌ ±0 
156 911 runs  +13  140 256 ✅ +14  16 655 💤  - 1  0 ❌ ±0 

Results for commit 21241c49. ± Comparison against base commit d280e163.

[test-results]:data:application/gzip;base64,H4sIAF05fWcC/02NzQ6DIBAGX8Vw7oHdAkJfpsFFElJ/GoRT03cvWhWPM/tl9sN8GPqFPRqQt4YtOaRKLkebwjytLLhuiyrXtN1Rwf3g55KJVglC8ipf4b1JVYfehqG4OupjnONuYp7+r5UB2PlMC45SnXJPKyXl4S7pja9lmscxpAIMAQWQMOicRk3Ikaz3EkxHrZZdZ53RLVlg3x+A0gTcGgEAAA==

fjetter: > We might want to consider adding an API for this instead of modifying the dictionary inplace?

I'm open to adding new APIs as long as they are somewhat general. I wouldn't want to add a `Task.maybe_fuse_slices` method but a `Task.unfuse` which returns a list of the nested tasks (i.e. disassemble the dict for you) is something I could see working. Such an API is more costly than the inplace mutation, of course.

Is there any other API you have in mind that would be useful to have?
fjetter: cc @hendrikmakait who's thinking about slimming down fuse and possibly changing its internal representation
phofl: I honestly don't know what a good API here would look like. This all feel like horrible abstraction leaks. We might consider just living with the ugly thing here?

Rebuilding the list of nested tasks just puts the ugliness somewhere else
fjetter: I find the fuse/unfuse to not be horrible. In fact, I could see this being useful outside of this application, e.g. if we encountered two sequential fused tasks and wanted to fuse those more efficiently like... `Task.fuse(*t.unfuse() for t in all_tasks)`

Patch:
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/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/_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/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/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