Repository: dask/dask
Pull request #11847 (2025-03-24 15:16:01)

Remove nightly ``h5py`` from ``upstream`` CI job
Looks like these aren't currently being published (xref https://github.com/h5py/h5py/issues/2563), leading to the `upstream` build failing ([latest job](https://github.com/dask/dask/actions/runs/14037827831/job/39300168373))

```
++ python -m pip install --no-deps --pre --retries 10 -i https://pypi.anaconda.org/scientific-python-nightly-wheels/simple numpy pandas scipy scikit-image h5py
Looking in indexes: https://pypi.anaconda.org/scientific-python-nightly-wheels/simple
Collecting numpy
  Downloading https://pypi.anaconda.org/scientific-python-nightly-wheels/simple/numpy/2.3.0.dev0/numpy-2.3.0.dev0-cp312-cp312-manylinux_2_28_x86_64.whl (16.8 MB)
     ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 16.8/16.8 MB 29.7 MB/s eta 0:00:00
Collecting pandas
  Downloading https://pypi.anaconda.org/scientific-python-nightly-wheels/simple/pandas/3.0.0.dev0%2B2022.gdc8401afea/pandas-3.0.0.dev0%2B2022.gdc8401afea-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (12.7 MB)
     ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 12.7/12.7 MB 25.6 MB/s eta 0:00:00
Collecting scipy
  Downloading https://pypi.anaconda.org/scientific-python-nightly-wheels/simple/scipy/1.16.0.dev0/scipy-1.16.0.dev0-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (37.8 MB)
     ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 37.8/37.8 MB 41.7 MB/s eta 0:00:00
Collecting scikit-image
  Downloading https://pypi.anaconda.org/scientific-python-nightly-wheels/simple/scikit-image/0.26.0rc0.dev0/scikit_image-0.26.0rc0.dev0-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (13.9 MB)
     ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 13.9/13.9 MB 41.0 MB/s eta 0:00:00
ERROR: Could not find a version that satisfies the requirement h5py (from versions: none)
ERROR: No matching distribution found for h5py
```


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._

      9 files  ±0        9 suites  ±0   3h 26m 8s ⏱️ +7s
 17 819 tests ±0   16 604 ✅ ±0   1 215 💤 ±0  0 ❌ ±0 
159 445 runs  ±0  147 332 ✅ +1  12 113 💤  - 1  0 ❌ ±0 

Results for commit 4bbfcc2b. ± Comparison against base commit e0877d0a.

[test-results]:data:application/gzip;base64,H4sIALqA4WcC/03Myw6DIBCF4VcxrLvgOkBfppFBElIvDcLK9N2LRqnL/5vM2UiI47CSZ2cfHVlLzC18SX2Oy1yTcQGmUj3m/cy0Yfbq11oQdwSg8o/v+Dk+mWoW+jhWow2GlJZ0SirzMa2slOrsNi21ELxhm2bistv00fdlXKYp5hpEOhcQuWOeGy051xoCBA/OW0cRlEIwAS2S7w97mgrNGAEAAA==


Patch:
diff --git a/continuous_integration/scripts/install.sh b/continuous_integration/scripts/install.sh
index 37908ca3a..b3ebdbf33 100644
--- a/continuous_integration/scripts/install.sh
+++ b/continuous_integration/scripts/install.sh
@@ -25,14 +25,16 @@ if [[ ${UPSTREAM_DEV} ]]; then
         git+https://github.com/dask/zict \
         git+https://github.com/dask/distributed \
         git+https://github.com/zarr-developers/zarr-python
-    mamba uninstall --force numpy pandas scipy numexpr numba sparse scikit-image h5py numbagg
+    # TODO: Add nightly `h5py` back once https://github.com/h5py/h5py/issues/2563 is resolved
+    # mamba uninstall --force numpy pandas scipy numexpr numba sparse scikit-image h5py numbagg
+    mamba uninstall --force numpy pandas scipy numexpr numba sparse scikit-image numbagg
     python -m pip install --no-deps --pre --retries 10 \
         -i https://pypi.anaconda.org/scientific-python-nightly-wheels/simple \
         numpy \
         pandas \
         scipy \
-        scikit-image \
-        h5py
+        scikit-image
+        # h5py
 
     # Used when automatically opening an issue when the `upstream` CI build fails
     mamba install pytest-reportlog
diff --git a/dask/_expr.py b/dask/_expr.py
index dfb1d9093..4e3d7ed5d 100644
--- a/dask/_expr.py
+++ b/dask/_expr.py
@@ -136,7 +136,13 @@ class Expr:
         return hash(self._name)
 
     def __dask_tokenize__(self):
-        return self._name
+        if not self._determ_token:
+            # If the subclass does not implement a __dask_tokenize__ we'll want
+            # to tokenize all operands.
+            # Note how this differs to the implementation of
+            # Expr.deterministic_token
+            self._determ_token = _tokenize_deterministic(type(self), *self.operands)
+        return self._determ_token
 
     @staticmethod
     def _reconstruct(*args):
@@ -494,7 +500,9 @@ class Expr:
     @property
     def deterministic_token(self):
         if not self._determ_token:
-            self._determ_token = _tokenize_deterministic(*self.operands)
+            # Just tokenize self to fall back on __dask_tokenize__
+            # Note how this differs to the implementation of __dask_tokenize__
+            self._determ_token = self.__dask_tokenize__()
         return self._determ_token
 
     @functools.cached_property
@@ -1074,6 +1082,11 @@ class _ExprSequence(Expr):
             all_keys.append(op.__dask_keys__())
         return all_keys
 
+    def __repr__(self):
+        return "ExprSequence(" + ", ".join(map(repr, self.operands)) + ")"
+
+    __str__ = __repr__
+
     def finalize_compute(self):
         return _ExprSequence(
             *(op.finalize_compute() for op in self.operands),
diff --git a/dask/array/_array_expr/_blockwise.py b/dask/array/_array_expr/_blockwise.py
index 394bccfb6..b123494cd 100644
--- a/dask/array/_array_expr/_blockwise.py
+++ b/dask/array/_array_expr/_blockwise.py
@@ -116,8 +116,7 @@ class Blockwise(ArrayExpr):
     def dtype(self):
         return self.operand("dtype")
 
-    @property
-    def deterministic_token(self):
+    def __dask_tokenize__(self):
         if not self._determ_token:
             # TODO: Is there an actual need to overwrite this?
             self._determ_token = _tokenize_deterministic(
diff --git a/dask/array/_array_expr/_expr.py b/dask/array/_array_expr/_expr.py
index b727e763b..1357ef0c8 100644
--- a/dask/array/_array_expr/_expr.py
+++ b/dask/array/_array_expr/_expr.py
@@ -104,9 +104,6 @@ class ArrayExpr(Expr):
 
         return unwrap(key_refs)
 
-    def __dask_tokenize__(self):
-        return self._name
-
     def __hash__(self):
         return hash(self._name)
 
diff --git a/dask/array/_array_expr/_reductions.py b/dask/array/_array_expr/_reductions.py
index faf6827ba..44ff47161 100644
--- a/dask/array/_array_expr/_reductions.py
+++ b/dask/array/_array_expr/_reductions.py
@@ -276,8 +276,7 @@ class PartialReduce(ArrayExpr):
         "reduced_meta": None,
     }
 
-    @property
-    def deterministic_token(self):
+    def __dask_tokenize__(self):
         if not self._determ_token:
             # TODO: Is there an actual need to overwrite this?
             self._determ_token = _tokenize_deterministic(
diff --git a/dask/dataframe/dask_expr/_expr.py b/dask/dataframe/dask_expr/_expr.py
index 408bab56b..6adf948f9 100644
--- a/dask/dataframe/dask_expr/_expr.py
+++ b/dask/dataframe/dask_expr/_expr.py
@@ -49,7 +49,6 @@ from dask.dataframe.utils import (
     raise_on_meta_error,
     valid_divisions,
 )
-from dask.tokenize import normalize_token
 from dask.typing import Key, no_default
 from dask.utils import (
     M,
@@ -3112,11 +3111,6 @@ class DelayedsExpr(Expr):
         return 0
 
 
-@normalize_token.register(Expr)
-def normalize_expression(expr):
-    return expr._name
-
-
 def is_broadcastable(dfs, s):
     """
     This Series is broadcastable against another dataframe in the sequence
diff --git a/dask/dataframe/dask_expr/io/parquet.py b/dask/dataframe/dask_expr/io/parquet.py
index 8e12e9abd..3f3c9ff5c 100644
--- a/dask/dataframe/dask_expr/io/parquet.py
+++ b/dask/dataframe/dask_expr/io/parquet.py
@@ -776,8 +776,7 @@ class ReadParquet(PartitionsFiltered, BlockwiseIO):
     def _funcname(self):
         return "read_parquet"
 
-    @property
-    def deterministic_token(self):
+    def __dask_tokenize__(self):
         if not self._determ_token:
             # TODO: Is there an actual need to overwrite this?
             self._determ_token = _tokenize_deterministic(


Test patch:
diff --git a/dask/tests/test_base.py b/dask/tests/test_base.py
index 76df1b678..6d3ecced6 100644
--- a/dask/tests/test_base.py
+++ b/dask/tests/test_base.py
@@ -19,7 +19,6 @@ import dask.bag as db
 from dask.base import (
     DaskMethodsMixin,
     clone_key,
-    collections_to_expr,
     compute,
     compute_as_if_collection,
     get_collection_names,
@@ -931,16 +930,6 @@ def test_num_workers_config(scheduler):
     assert len(workers) == num_workers
 
 
-def test_optimizations_ctd():
-    pytest.importorskip("numpy")
-    da = pytest.importorskip("dask.array")
-    x = da.arange(2, chunks=1)[:1]
-    dsk1 = collections_to_expr([x])
-    with dask.config.set({"optimizations": [lambda dsk, keys: dsk]}):
-        dsk2 = collections_to_expr([x])
-    assert dsk1 == dsk2
-
-
 def test_clone_key():
     for key, seed in [("x", 123), (("x", 1), 456), (("sum-1-2-3", h1, 1), 123)]:
         validate_key(clone_key(key, seed))
diff --git a/dask/tests/test_hlgexpr.py b/dask/tests/test_hlgexpr.py
new file mode 100644
index 000000000..391d83d83
--- /dev/null
+++ b/dask/tests/test_hlgexpr.py
@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+import pickle
+
+from dask._expr import HLGExpr
+from dask.tokenize import tokenize
+
+
+def test_tokenize():
+    # Ensure tokens are different for different high-level graphs The current
+    # implementation actually ensures that no HLGExpr are tokenizing equally.
+    # Technically, we do not need such a strong guarantee. but tokenizing a full
+    # HLG reliably is tricky and we do not require the reproducibility for
+    # HLGExpr since they do not undergo the same kind of optimization as the
+    # rest of the graph.
+    from dask.highlevelgraph import HighLevelGraph
+
+    dsk = HighLevelGraph.from_collections("x", {"foo": None})
+    dsk2 = HighLevelGraph.from_collections("x", {"bar": None})
+    dsk3 = HighLevelGraph.from_collections("y", {"foo": None})
+    assert tokenize(HLGExpr(dsk)) != tokenize(HLGExpr(dsk2))
+    assert tokenize(HLGExpr(dsk)) != tokenize(HLGExpr(dsk3))
+    assert tokenize(HLGExpr(dsk2)) != tokenize(HLGExpr(dsk3))
+
+    # Roundtrip preserves the tokens
+    for expr in [HLGExpr(dsk), HLGExpr(dsk2), HLGExpr(dsk3)]:
+        assert tokenize(pickle.loads(pickle.dumps(expr))) == tokenize(expr)