Repository: dask/dask
Pull request #10697 (2023-12-13 13:05:54)

[Dask.order] Remove non-runnable leaf nodes from ordering
This is a follow up to https://github.com/dask/dask/pull/10660 and addresses the still failing test in https://github.com/dask/distributed/issues/8255

Most notably, this removes a special case in the `process_runnable` code path that caused this kind of graph to be executed too eagerly. The special code path was not hit in the specific unit test I added in `test_order.py`. I extended the test logic to cover this now.

This special branch was added to enable eager execution of some dangling code branches as observed in `test_array_store_final_order`

![image](https://github.com/dask/dask/assets/8629629/c5e7093b-9639-4a64-b5d8-6ea37ba59941)

(The image shows the correct order)

The dangling branches I was referring to are those linear branches ending in P51 and P15. Without that special casing, those two branches would not have been executed causing the root to be left in memory until the end.

What this visualization is not showing is what causes the problem. This graph is actually reducing to a single task `"store-12345": ["store-map-1", "store-map-2", ...]` that is effectively an alias. This alias is throwing off the critical path algorithm such that those thin, small branches are effectively ignored and deemed as not valuable. Without that final reducer, the algorithm is forced to finish the connected graph first.
This weird alias is actually something that is not even properly understood by the visualization code. It's rendered as the box in the lower right corner but it doesn't know what to do with it.

fjetter: Note: this also includes a mypy fix that is breaking distributed linting

Patch:
diff --git a/continuous_integration/environment-3.12.yaml b/continuous_integration/environment-3.12.yaml
index cae467f38..85b32b2b7 100644
--- a/continuous_integration/environment-3.12.yaml
+++ b/continuous_integration/environment-3.12.yaml
@@ -77,3 +77,4 @@ dependencies:
   - pip
   - pip:
     - git+https://github.com/dask/distributed
+    - git+https://github.com/dask-contrib/dask-expr
diff --git a/dask/dask-schema.yaml b/dask/dask-schema.yaml
index f11faad79..a9ce92e80 100644
--- a/dask/dask-schema.yaml
+++ b/dask/dask-schema.yaml
@@ -89,6 +89,10 @@ properties:
         type: [boolean, 'null']
         description: |
           Whether to convert string-like data to pyarrow strings.
+      query-planning:
+        type: [boolean]
+        description: |
+          Whether to use query planning.
 
   array:
     type: object
diff --git a/dask/dask.yaml b/dask/dask.yaml
index 808ac2c71..ac6030061 100644
--- a/dask/dask.yaml
+++ b/dask/dask.yaml
@@ -15,6 +15,7 @@ dataframe:
     metadata-task-size-local: 512  # Number of files per local metadata-processing task
     metadata-task-size-remote: 1  # Number of files per remote metadata-processing task
   convert-string: null  # Whether to convert string-like data to pyarrow strings
+  query-planning: false  # Whether to use dask-expr
 
 array:
   backend: "numpy"  # Backend array library for input IO and data creation
diff --git a/dask/dataframe/__init__.py b/dask/dataframe/__init__.py
index bff0a4ba5..48288a81c 100644
--- a/dask/dataframe/__init__.py
+++ b/dask/dataframe/__init__.py
@@ -1,66 +1,141 @@
 from __future__ import annotations
 
-try:
-    import dask.dataframe._pyarrow_compat
-    from dask.base import compute
-    from dask.dataframe import backends, dispatch, rolling
-    from dask.dataframe.core import (
+
+def _dask_expr_enabled() -> bool:
+    import dask
+
+    use_dask_expr = dask.config.get("dataframe.query-planning")
+    if use_dask_expr:
+        try:
+            import dask_expr  # noqa: F401
+        except ImportError:
+            raise ValueError("Must install dask-expr to activate query planning.")
+    return use_dask_expr
+
+
+if _dask_expr_enabled():
+    from dask_expr import (
         DataFrame,
         Index,
         Series,
-        _Frame,
-        map_partitions,
-        repartition,
-        to_datetime,
-        to_timedelta,
-    )
-    from dask.dataframe.groupby import Aggregation
-    from dask.dataframe.io import (
-        demo,
-        from_array,
-        from_dask_array,
-        from_delayed,
-        from_dict,
+        concat,
+        from_dask_dataframe,
+        from_graph,
         from_map,
         from_pandas,
+        merge,
+        pivot_table,
         read_csv,
-        read_fwf,
-        read_hdf,
-        read_json,
-        read_sql,
-        read_sql_query,
-        read_sql_table,
-        read_table,
-        to_bag,
-        to_csv,
-        to_hdf,
-        to_json,
-        to_records,
-        to_sql,
+        read_parquet,
+        repartition,
+        to_datetime,
+        to_numeric,
+        to_parquet,
+        to_timedelta,
     )
-    from dask.dataframe.multi import concat, merge, merge_asof
-    from dask.dataframe.numeric import to_numeric
-    from dask.dataframe.optimize import optimize
-    from dask.dataframe.reshape import get_dummies, melt, pivot_table
+
+    import dask.dataframe._pyarrow_compat
+    from dask.base import compute
+    from dask.dataframe import backends, dispatch
     from dask.dataframe.utils import assert_eq
 
+    def raise_not_implemented_error(attr_name):
+        def inner_func(*args, **kwargs):
+            raise NotImplementedError(
+                f"Function {attr_name} is not implemented for dask-expr."
+            )
+
+        return inner_func
+
+    _Frame = raise_not_implemented_error("_Frame")
+    map_partitions = raise_not_implemented_error("map_partitions")
+    Aggregation = raise_not_implemented_error("Aggregation")
+    from_array = raise_not_implemented_error("from_array")
+    from_dask_array = raise_not_implemented_error("from_dask_array")
+    from_delayed = raise_not_implemented_error("from_delayed")
+    from_dict = raise_not_implemented_error("from_dict")
+    read_fwf = raise_not_implemented_error("read_fwf")
+    read_hdf = raise_not_implemented_error("read_hdf")
+    read_json = raise_not_implemented_error("read_json")
+    read_sql = raise_not_implemented_error("read_sql")
+    read_sql_query = raise_not_implemented_error("read_sql_query")
+    read_sql_table = raise_not_implemented_error("read_sql_table")
+    read_table = raise_not_implemented_error("read_table")
+    to_bag = raise_not_implemented_error("to_bag")
+    to_csv = raise_not_implemented_error("to_csv")
+    to_hdf = raise_not_implemented_error("to_hdf")
+    to_json = raise_not_implemented_error("to_json")
+    to_records = raise_not_implemented_error("to_records")
+    to_sql = raise_not_implemented_error("to_sql")
+    merge_asof = raise_not_implemented_error("merge_asof")
+    get_dummies = raise_not_implemented_error("get_dummies")
+    melt = raise_not_implemented_error("melt")
+    read_orc = raise_not_implemented_error("read_orc")
+    to_orc = raise_not_implemented_error("to_orc")
+    isna = raise_not_implemented_error("isna")
+
+else:
     try:
-        from dask.dataframe.io import read_parquet, to_parquet
-    except ImportError:
-        pass
-    try:
-        from dask.dataframe.io import read_orc, to_orc
-    except ImportError:
-        pass
-    try:
-        from dask.dataframe.core import isna
-    except ImportError:
-        pass
-except ImportError as e:
-    msg = (
-        "Dask dataframe requirements are not installed.\n\n"
-        "Please either conda or pip install as follows:\n\n"
-        "  conda install dask                     # either conda install\n"
-        '  python -m pip install "dask[dataframe]" --upgrade  # or python -m pip install'
-    )
-    raise ImportError(msg) from e
+        import dask.dataframe._pyarrow_compat
+        from dask.base import compute
+        from dask.dataframe import backends, dispatch, rolling
+        from dask.dataframe.core import (
+            DataFrame,
+            Index,
+            Series,
+            _Frame,
+            map_partitions,
+            repartition,
+            to_datetime,
+            to_timedelta,
+        )
+        from dask.dataframe.groupby import Aggregation
+        from dask.dataframe.io import (
+            demo,
+            from_array,
+            from_dask_array,
+            from_delayed,
+            from_dict,
+            from_map,
+            from_pandas,
+            read_csv,
+            read_fwf,
+            read_hdf,
+            read_json,
+            read_sql,
+            read_sql_query,
+            read_sql_table,
+            read_table,
+            to_bag,
+            to_csv,
+            to_hdf,
+            to_json,
+            to_records,
+            to_sql,
+        )
+        from dask.dataframe.multi import concat, merge, merge_asof
+        from dask.dataframe.numeric import to_numeric
+        from dask.dataframe.optimize import optimize
+        from dask.dataframe.reshape import get_dummies, melt, pivot_table
+        from dask.dataframe.utils import assert_eq
+
+        try:
+            from dask.dataframe.io import read_parquet, to_parquet
+        except ImportError:
+            pass
+        try:
+            from dask.dataframe.io import read_orc, to_orc
+        except ImportError:
+            pass
+        try:
+            from dask.dataframe.core import isna
+        except ImportError:
+            pass
+    except ImportError as e:
+        msg = (
+            "Dask dataframe requirements are not installed.\n\n"
+            "Please either conda or pip install as follows:\n\n"
+            "  conda install dask                     # either conda install\n"
+            '  python -m pip install "dask[dataframe]" --upgrade  # or python -m pip install'
+        )
+        raise ImportError(msg) from e
diff --git a/dask/dataframe/io/parquet/arrow.py b/dask/dataframe/io/parquet/arrow.py
index d47c6214a..209ad3fa4 100644
--- a/dask/dataframe/io/parquet/arrow.py
+++ b/dask/dataframe/io/parquet/arrow.py
@@ -1815,6 +1815,15 @@ class ArrowDatasetEngine(Engine):
         # next in priority is converting strings
         if convert_string:
             type_mappers.append({pa.string(): pd.StringDtype("pyarrow")}.get)
+            type_mappers.append({pa.date32(): pd.ArrowDtype(pa.date32())}.get)
+            type_mappers.append({pa.date64(): pd.ArrowDtype(pa.date64())}.get)
+
+            def _convert_decimal_type(type):
+                if pa.types.is_decimal(type):
+                    return pd.ArrowDtype(type)
+                return None
+
+            type_mappers.append(_convert_decimal_type)
 
         # and then nullable types
         if dtype_backend == "numpy_nullable":
diff --git a/dask/order.py b/dask/order.py
index c29196c3f..da7d6082f 100644
--- a/dask/order.py
+++ b/dask/order.py
@@ -74,7 +74,7 @@ def order(
     dsk: Mapping[Key, Any],
     dependencies: Mapping[Key, set[Key]] | None = None,
     *,
-    return_stats: Literal[False],
+    return_stats: Literal[False] = False,
 ) -> dict[Key, int]:
     ...
 
@@ -106,6 +106,7 @@ def order(
         return {}  # type: ignore
 
     dsk = dict(dsk)
+    expected_len = len(dsk)
 
     if dependencies is None:
         dependencies = {k: get_dependencies(dsk, k) for k in dsk}
@@ -121,11 +122,43 @@ def order(
 
     leaf_nodes = {k for k, v in dependents.items() if not v}
     root_nodes = {k for k, v in dependencies.items() if not v}
+
+    result: dict[Key, Order | int] = {}
+
+    # Normalize the graph by removing leaf nodes that are not actual tasks, see
+    # for instance da.store where the task is merely an alias
+    # to multiple keys, i.e. [key1, key2, ...,]
+    all_tasks = False
+    n_removed_leaves = 0
+    while not all_tasks:
+        all_tasks = True
+        for leaf in list(leaf_nodes):
+            if (
+                not istask(dsk[leaf])
+                # Having a linear chain is fine
+                and len(dependencies[leaf]) > 1
+            ):
+                all_tasks = False
+                # Put non-tasks at the very end since they are merely aliases
+                # and have no impact on performance at all
+                prio = len(dsk) - 1 - n_removed_leaves
+                if return_stats:
+                    result[leaf] = Order(prio, -1)
+                else:
+                    result[leaf] = prio
+                n_removed_leaves += 1
+                leaf_nodes.remove(leaf)
+                del dsk[leaf]
+                del dependents[leaf]
+                for dep in dependencies[leaf]:
+                    dependents[dep].remove(leaf)
+                    if not dependents[dep]:
+                        leaf_nodes.add(dep)
+
     assert dependencies is not None
     roots_connected, max_dependents = _connecting_to_roots(dependencies, dependents)
     leafs_connected, _ = _connecting_to_roots(dependents, dependencies)
     i = 0
-    result: dict[Key, Order | int] = {}
 
     runnable_hull = set()
     reachable_hull = set()
@@ -253,15 +286,9 @@ def order(
                             # FIXME: The fact that it is possible for
                             # num_needed[current] == 0 means we're doing some
                             # work twice
-                            if num_needed[current] <= 1 or (
-                                not branches
-                                # FIXME: This is a very magical number
-                                and len(path) > 2
-                            ):
-                                for k in path[:-1]:
+                            if num_needed[current] <= 1:
+                                for k in path:
                                     add_to_result(k)
-                                if not num_needed[current]:
-                                    add_to_result(current)
                         elif len(path) == 1 or len(deps_upstream) == 1:
                             if len(deps_downstream) > 1:
                                 for d in sorted(deps_downstream, key=sort_key):
@@ -485,7 +512,7 @@ def order(
         scpath_discard(item)
         return item
 
-    while len(result) < len(dsk):
+    while len(result) < expected_len:
         crit_path_counter += 1
         assert not critical_path
         assert not scrit_path
@@ -541,6 +568,7 @@ def order(
                 add_to_result(item)
         process_runnables()
 
+    assert len(result) == expected_len
     return result  # type: ignore
 
 
diff --git a/docs/source/changelog.rst b/docs/source/changelog.rst
index 4a104fa6d..1970ef812 100644
--- a/docs/source/changelog.rst
+++ b/docs/source/changelog.rst
@@ -1,6 +1,26 @@
 Changelog
 =========
 
+.. _v2023.12.1:
+
+2023.12.1
+---------
+
+Released on December 15, 2023
+
+Highlights
+^^^^^^^^^^
+
+Dtype inference in ``read_parquet``
+"""""""""""""""""""""""""""""""""""
+
+``read_parquet`` will now infer the Arrow types ``pa.date32()``, ``pa.date64()`` and
+``pa.decimal()`` as a ``ArrowDtype`` in pandas. These dtypes are backed by the
+original Arrow array, and thus avoid the conversion to NumPy object. Additionally,
+``read_parquet`` will no longer infer nested and binary types as strings, they will
+be stored in NumPy object arrays.
+
+
 .. _v2023.12.0:
 
 2023.12.0
diff --git a/docs/source/deployment-considerations.rst b/docs/source/deployment-considerations.rst
index 1508ee99a..131b07268 100644
--- a/docs/source/deployment-considerations.rst
+++ b/docs/source/deployment-considerations.rst
@@ -10,7 +10,7 @@ These challenges are especially relevant when managing Dask for a team or organi
 
 Consistent software environments
 --------------------------------
-For Dask to function properly, the same set of Python packages, at the same versions, need to be installed on the scheduler and workers as on the client. One of the most common stumbling points in deploying Dask on multiple machines is keeping what's installed on the cluster up to date with what's on the client, especially if they run on different systems (a laptop and a cloud provider, for example). For possible approaches, see :doc:`how-to/manage-environments`.
+For Dask to function properly, the same set of Python packages, at the same versions, need to be installed on the scheduler and workers as on the client. One of the most common stumbling points in deploying Dask on multiple machines is keeping what's installed on the cluster up to date with what's on the client, especially if they run on different systems (a laptop and a cloud provider, for example). For possible approaches, see :doc:`software-environments`.
 
 Some ways of maintaining consistent environments may also require extra infrastructure. For example, using Docker images is common in cloud deployments, but then you need somewhere to store the images, such as `DockerHub <https://hub.docker.com/>`_, `AWS ECR <https://aws.amazon.com/ecr/>`_, `GCP Container Registry <https://cloud.google.com/container-registry>`_, etc. As use matures, you'll also want to version-control your Dockerfile and automatically build and publish new images when it changes, using a CI/CD system like `GitHub Actions <https://github.com/marketplace/actions/build-and-push-docker-images>`_, `Google Cloud Build <https://cloud.google.com/build/docs/build-push-docker-image>`_, or others.
 
diff --git a/docs/source/install.rst b/docs/source/install.rst
index de38b5b89..c3e2a340f 100644
--- a/docs/source/install.rst
+++ b/docs/source/install.rst
@@ -2,94 +2,97 @@ Dask Installation
 =================
 
 .. meta::
-    :description: Dask Installation | You can install Dask with conda, pip install Dask, or install from source.
+    :description: Dask Installation | You can easily install Dask with conda or pip
 
 .. _Anaconda distribution: https://www.anaconda.com/download/
 
-Dask currently supports Linux, macOS, and Windows. See the :doc:`changelog`
-for comprehensive release notes for each Dask version.
-
 How to Install Dask
 -------------------
 
-Dask installation can happen in a few different ways. You can install Dask with ``conda``, with ``pip``, or install from source.
+You can install Dask with ``conda``, with ``pip``, or install from source.
 
-Conda
------
+.. tab-set::
 
-If you use the `Anaconda distribution`_, Dask installation will occur by default.
-You can also install or upgrade Dask using the
-`conda install <https://docs.conda.io/projects/conda/en/latest/commands/install.html>`_ command::
+   .. tab-item:: Conda
 
-   conda install dask
+      If you use the `Anaconda distribution`_, Dask will be installed by default.
 
-This installs Dask and **all** common dependencies, including pandas and NumPy.
-Dask packages are maintained both on the default channel and on `conda-forge <https://conda-forge.github.io/>`_.
-You can select the channel with the ``-c`` flag::
+      You can also install or upgrade Dask using the
+      `conda install <https://docs.conda.io/projects/conda/en/latest/commands/install.html>`_ command::
 
-    conda install dask -c conda-forge
+         conda install dask
 
-Optionally, you can obtain a minimal Dask installation using the following command::
+      This installs Dask and **all** common dependencies, including pandas and NumPy.
+      Dask packages are maintained both on the defaults channel and on
+      `conda-forge <https://conda-forge.github.io/>`_.
+      You can select the channel with the ``-c`` flag::
 
-   conda install dask-core
+         conda install dask -c conda-forge
 
-This will install a minimal set of dependencies required to run Dask similar to (but not exactly the same as) ``python -m pip install dask`` below.
+      Optionally, you can obtain a minimal Dask installation using the following command::
 
-Pip
----
+         conda install dask-core
 
-You can use pip to install everything required for most common uses of Dask
-(e.g. Dask Array, Dask DataFrame, etc.).
-This installs both Dask and dependencies, like NumPy and pandas,
-that are necessary for different workloads. This is often the right
-choice for Dask users::
+      This will install a minimal set of dependencies required to run Dask similar to (but not exactly the same as) ``python -m pip install dask``.
 
-   python -m pip install "dask[complete]"    # Install everything
+   .. tab-item:: Pip
 
-You can also install only the Dask library.  Modules like ``dask.array``,
-``dask.dataframe``, or ``dask.distributed`` won't work until you also install NumPy,
-pandas, or Tornado, respectively.  This is common for downstream library
-maintainers::
+      To install Dask with ``pip`` run the following::
 
-   python -m pip install dask                # Install only core parts of dask
+         python -m pip install "dask[complete]"    # Install everything
 
-We also maintain other dependency sets for different subsets of functionality::
+      This installs Dask, the distributed scheduler, and common dependencies
+      like pandas, Numpy, and others.
 
-   python -m pip install "dask[array]"       # Install requirements for dask array
-   python -m pip install "dask[dataframe]"   # Install requirements for dask dataframe
-   python -m pip install "dask[diagnostics]" # Install requirements for dask diagnostics
-   python -m pip install "dask[distributed]" # Install requirements for distributed dask
+      You can also install only the Dask library and no optional dependencies::
 
-We have these options so that users of the lightweight core Dask scheduler
-aren't required to download the more exotic dependencies of the collections
-(Numpy, pandas, Tornado, etc.).
+         python -m pip install dask                # Install only core parts of dask
 
-Install from Source
--------------------
+      Dask modules like ``dask.array``, ``dask.dataframe``, or
+      ``dask.distributed`` won't work until you also install NumPy, pandas, or
+      Tornado, respectively.  This is uncommon for users but more common for
+      downstream library maintainers.
 
-To install Dask from source, clone the repository from `github
-<https://github.com/dask/dask>`_::
+      We also maintain other dependency sets for different subsets of functionality::
 
-    git clone https://github.com/dask/dask.git
-    cd dask
-    python -m pip install .
+         python -m pip install "dask[array]"       # Install requirements for dask array
+         python -m pip install "dask[dataframe]"   # Install requirements for dask dataframe
+         python -m pip install "dask[diagnostics]" # Install requirements for dask diagnostics
+         python -m pip install "dask[distributed]" # Install requirements for distributed dask
+
+      We have these options so that users of the lightweight core Dask scheduler
+      aren't required to download the more exotic dependencies of the collections
+      (Numpy, pandas, Tornado, etc.).
+
+   .. tab-item:: Source
+
+      To install Dask from source, clone the repository from `GitHub
+      <https://github.com/dask/dask>`_::
+
+         git clone https://github.com/dask/dask.git
+         cd dask
+         python -m pip install .
+
+      You can also install all dependencies as well::
 
-You can also install all dependencies as well::
+         python -m pip install ".[complete]"
 
-    python -m pip install ".[complete]"
+      You can view the list of all dependencies within the ``project.optional-dependencies`` field
+      of ``pyproject.toml``.
 
-You can view the list of all dependencies within the ``extras_require`` field
-of ``setup.py``.
+      Or do a developer install by using the ``-e`` flag
+      (see the :ref:`Install section <develop-install>` in the Development Guidelines)::
 
-Or do a developer install by using the ``-e`` flag
-(see the :ref:`Install section <develop-install>` in the Development Guidelines)::
+         python -m pip install -e .
 
-    python -m pip install -e .
+Distributed Deployment
+----------------------
 
-Anaconda
---------
+To run Dask on a distributed cluster you will want to also install the Dask
+cluster manager that matches your resource manager, like Kubernetes, SLURM, PBS,
+LSF, AWS, GCP, Azure, or similar technology.
 
-Dask is included by default in the `Anaconda distribution`_.
+Read more on this topic at :bdg-link-primary:`Deploy Documentation <deploying.html>`
 
 Optional dependencies
 ---------------------
@@ -172,7 +175,7 @@ Test Dask with ``py.test``::
     cd dask
     py.test dask
 
-Installing Dask naively may not install all requirements by default (see the `Pip`_ section above).
+Installing Dask naively may not install all requirements by default (see the ``pip`` section above).
 You may choose to install the ``dask[complete]`` version which includes
 all dependencies for all collections::
 


Test patch:
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
index a0cb9daa4..3d56e44f7 100644
--- a/.github/workflows/tests.yml
+++ b/.github/workflows/tests.yml
@@ -48,7 +48,7 @@ jobs:
           # Pyarrow strings turned on
           - os: "ubuntu-latest"
             environment: "3.12"
-            extra: "pyarrow"
+            extra: "dask-expr"
 
     env:
       PARALLEL: "true"
@@ -84,6 +84,13 @@ jobs:
           echo "DASK_DATAFRAME__CONVERT_STRING: $DASK_DATAFRAME__CONVERT_STRING"
           echo "DASK_DATAFRAME__CONVERT_STRING=$DASK_DATAFRAME__CONVERT_STRING" >> $GITHUB_ENV
 
+      - name: Turn on dask-expr config option
+        if: ${{ matrix.extra == 'dask-expr' }}
+        run: |
+          export DASK_DATAFRAME__QUERY_PLANNING="True"
+          echo "DASK_DATAFRAME__QUERY_PLANNING: DASK_DATAFRAME__QUERY_PLANNING"
+          echo "DASK_DATAFRAME__QUERY_PLANNING=DASK_DATAFRAME__QUERY_PLANNING" >> $GITHUB_ENV
+
       - name: Reconfigure pytest-timeout
         # No SIGALRM available on Windows
         if: ${{ matrix.os != 'windows-latest' }}
@@ -101,6 +108,7 @@ jobs:
           path: env.yaml
 
       - name: Run tests
+        continue-on-error: ${{ matrix.extra == 'dask-expr' }}
         run: source continuous_integration/scripts/run_tests.sh
 
       - name: Coverage
diff --git a/dask/dataframe/io/tests/test_parquet.py b/dask/dataframe/io/tests/test_parquet.py
index 980098146..3b0d63a0a 100644
--- a/dask/dataframe/io/tests/test_parquet.py
+++ b/dask/dataframe/io/tests/test_parquet.py
@@ -5,6 +5,7 @@ import glob
 import math
 import os
 import warnings
+from datetime import date
 from decimal import Decimal
 from unittest.mock import MagicMock
 
@@ -4141,7 +4142,36 @@ def test_roundtrip_decimal_dtype(tmpdir):
     ddf1.to_parquet(path=tmpdir, engine="pyarrow", schema={"col1": pa.decimal128(5, 2)})
     ddf2 = dd.read_parquet(tmpdir, engine="pyarrow")
 
-    assert ddf1["col1"].dtype == ddf2["col1"].dtype
+    if pyarrow_strings_enabled():
+        assert pa.types.is_decimal(ddf2["col1"].dtype.pyarrow_dtype)
+        ddf1 = ddf1.astype({"col1": pd.ArrowDtype(pa.decimal128(5, 2))})
+    else:
+        assert ddf1["col1"].dtype == ddf2["col1"].dtype
+    assert_eq(ddf1, ddf2, check_divisions=False)
+
+
+@PYARROW_MARK
+def test_roundtrip_date_dtype(tmpdir):
+    # https://github.com/dask/dask/issues/6948
+    tmpdir = str(tmpdir)
+
+    data = [
+        {
+            "ts": pd.to_datetime("2021-01-01", utc="Europe/Berlin"),
+            "col1": date(2020, 10, 10),
+        }
+        for _ in range(23)
+    ]
+    ddf1 = dd.from_pandas(pd.DataFrame(data), npartitions=1)
+
+    ddf1.to_parquet(path=tmpdir, engine="pyarrow", schema={"col1": pa.date32()})
+    ddf2 = dd.read_parquet(tmpdir, engine="pyarrow")
+
+    if pyarrow_strings_enabled():
+        assert pa.types.is_date32(ddf2["col1"].dtype.pyarrow_dtype)
+        ddf1 = ddf1.astype({"col1": pd.ArrowDtype(pa.date32())})
+    else:
+        assert ddf1["col1"].dtype == ddf2["col1"].dtype
     assert_eq(ddf1, ddf2, check_divisions=False)
 
 
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index 769933333..7c661f896 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -75,8 +75,12 @@ dsk = {
 meta = make_meta(
     {"a": "i8", "b": "i8"}, index=pd.Index([], "i8"), parent_meta=pd.DataFrame()
 )
-d = dd.DataFrame(dsk, "x", meta, [0, 5, 9, 9])
-full = d.compute()
+if not dd._dask_expr_enabled():
+    d = dd.DataFrame(dsk, "x", meta, [0, 5, 9, 9])
+    full = d.compute()
+else:
+    d = dd.repartition(pd.concat(dsk.values()), divisions=[0, 5, 9, 9])
+    full = d.compute()
 
 
 def _drop_mean(df, col=None):
@@ -3161,6 +3165,8 @@ def test_drop_meta_mismatch():
 
 
 def test_gh580():
+    if dd._dask_expr_enabled():
+        pytest.skip("don't know")
     df = pd.DataFrame({"x": np.arange(10, dtype=float)})
     ddf = dd.from_pandas(df, 2)
     assert_eq(np.cos(df["x"]), np.cos(ddf["x"]))
diff --git a/dask/dataframe/tests/test_indexing.py b/dask/dataframe/tests/test_indexing.py
index caa664b2d..54dd057df 100644
--- a/dask/dataframe/tests/test_indexing.py
+++ b/dask/dataframe/tests/test_indexing.py
@@ -21,8 +21,12 @@ dsk = {
 meta = make_meta(
     {"a": "i8", "b": "i8"}, index=pd.Index([], "i8"), parent_meta=pd.DataFrame()
 )
-d = dd.DataFrame(dsk, "x", meta, [0, 5, 9, 9])
-full = d.compute()
+if not dd._dask_expr_enabled():
+    d = dd.DataFrame(dsk, "x", meta, [0, 5, 9, 9])
+    full = d.compute()
+else:
+    d = dd.repartition(pd.concat(dsk.values()), divisions=[0, 5, 9, 9])
+    full = d.compute()
 
 
 def test_loc():
diff --git a/dask/dataframe/tests/test_ufunc.py b/dask/dataframe/tests/test_ufunc.py
index e23bfe402..8e1328d5c 100644
--- a/dask/dataframe/tests/test_ufunc.py
+++ b/dask/dataframe/tests/test_ufunc.py
@@ -12,6 +12,10 @@ import dask.array as da
 import dask.dataframe as dd
 from dask.dataframe.utils import assert_eq
 
+if dd._dask_expr_enabled():
+    pytest.skip("doesn't work", allow_module_level=True)
+
+
 _BASE_UFUNCS = [
     "conj",
     "exp",
diff --git a/dask/tests/test_order.py b/dask/tests/test_order.py
index 16e387acc..c25e9e769 100644
--- a/dask/tests/test_order.py
+++ b/dask/tests/test_order.py
@@ -269,7 +269,7 @@ def test_break_ties_by_str(abcde):
     a, b, c, d, e = abcde
     dsk = {("x", i): (inc, i) for i in range(10)}
     x_keys = sorted(dsk)
-    dsk["y"] = list(x_keys)
+    dsk["y"] = (f, list(x_keys))
 
     o = order(dsk)
     expected = {"y": 10}
@@ -1220,7 +1220,6 @@ def test_anom_mean():
     # `mean_chunk` which is the primary reducer in this graph. Therefore we want
     # to run those as quickly as possible.
     # This is difficult to assert on but the pressure is an ok-ish proxy
-    # visualize(graph)
     assert max(pressure) <= 177
     from collections import defaultdict
 
@@ -1757,9 +1756,11 @@ def test_flox_reduction(abcde):
     assert max(of1) < min(of2) or max(of2) < min(of1)
 
 
+@pytest.mark.parametrize("optimize", [True, False])
+@pytest.mark.parametrize("keep_self", [True, False])
 @pytest.mark.parametrize("ndeps", [2, 5])
 @pytest.mark.parametrize("n_reducers", [4, 7])
-def test_reduce_with_many_common_dependents(ndeps, n_reducers):
+def test_reduce_with_many_common_dependents(optimize, keep_self, ndeps, n_reducers):
     da = pytest.importorskip("dask.array")
     import numpy as np
 
@@ -1780,21 +1781,29 @@ def test_reduce_with_many_common_dependents(ndeps, n_reducers):
     graph = x.sum(axis=1, split_every=20)
     from dask.order import order
 
-    dsk = collections_to_dsk([graph])
+    if keep_self:
+        # Keeping self adds a layer that cannot be fused
+        dsk = collections_to_dsk([x, graph], optimize_graph=optimize)
+    else:
+        dsk = collections_to_dsk([graph], optimize_graph=optimize)
     dependencies, dependents = get_deps(dsk)
     # Verify assumptions
     o = order(dsk)
     # Verify assumptions (specifically that the reducers are sum-aggregate)
-    assert {key_split(k) for k in o} == {"object", "sum", "sum-aggregate"}
-
+    assert ({"object", "sum", "sum-aggregate"}).issubset({key_split(k) for k in o})
     reducers = {k for k in o if key_split(k) == "sum-aggregate"}
-    drift = dict()
-    for r in reducers:
-        prios_deps = []
-        for dep in dependencies[r]:
-            prios_deps.append(o[dep])
-        drift[r] = (min(prios_deps), max(prios_deps))
-        assert max(prios_deps) - min(prios_deps) == len(dependencies[r]) - 1
+    assert (p := max(diagnostics(dsk)[1])) <= n_reducers + ndeps, p
+    # With optimize the metrics below change since there are many layers that
+    # are being fused but the pressure above should already be a strong
+    # indicator if something is wrong
+    if optimize:
+        for r in reducers:
+            prios_deps = []
+            for dep in dependencies[r]:
+                prios_deps.append(o[dep])
+            assert max(prios_deps) - min(prios_deps) == (len(dependencies[r]) - 1) * (
+                2 if keep_self else 1
+            )
 
 
 def test_doublediff(abcde):
@@ -1921,7 +1930,7 @@ def test_gh_3055_explicit(abcde):
     assert o[(a, 2)] < o[(a, 3)] < o[(a, 4)]
 
 
-def test_order_flux_reduction_2(abcde):
+def test_order_flox_reduction_2(abcde):
     # https://github.com/dask/dask/issues/10618
     a, b, c, d, e = abcde
     dsk = {
@@ -1949,7 +1958,6 @@ def test_order_flux_reduction_2(abcde):
         (d, 1, 0): (c, 1, 0),
         (d, 1, 1): (c, 1, 1),
     }
-
     o = order(dsk)
     final_nodes = sorted(
         [(d, ix, jx) for ix in range(2) for jx in range(2)], key=o.__getitem__
@@ -2118,3 +2126,19 @@ def test_connecting_to_roots_asym():
         k: max((len(dependents[r]) for r in v), default=len(dependents[k]))
         for k, v in connected_roots.items()
     }
+
+
+def test_do_not_mutate_input():
+    # Internally we may modify the graph but we don't want to mutate the
+    # external dsk
+    np = pytest.importorskip("numpy")
+    dsk = {
+        "a": "data",
+        "b": (f, 1),
+        "c": np.array([[1, 2], [3, 4]]),
+        "d": ["a", "b", "c"],
+    }
+    dsk_copy = dsk.copy()
+    o = order(dsk)
+    assert o["d"] == len(dsk) - 1
+    assert len(dsk) == len(dsk_copy)