Repository: dask/dask
Pull request #10698 (2023-12-13 16:34:34)

Infer dates and decimals in arrow as ArrowDtypes in pandas
cc @mrocklin @fjetter 

I couldn't get arrow to preserve date64, that's why there is no test. not sure what this type is actually used for

mrocklin: Cool.  Thanks @phofl 

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/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",