Repository: dask/dask
Pull request #11926 (2025-04-30 10:27:36)

Update pre-commit


Patch:
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 10a620323..01b974dd5 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -1,6 +1,6 @@
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.6.0
+    rev: v5.0.0
     hooks:
       - id: end-of-file-fixer
       - id: debug-statements
@@ -10,25 +10,25 @@ repos:
       - id: absolufy-imports
         name: absolufy-imports
   - repo: https://github.com/pycqa/isort
-    rev: 5.13.2
+    rev: 6.0.1
     hooks:
       - id: isort
         language_version: python3
   - repo: https://github.com/asottile/pyupgrade
-    rev: v3.17.0
+    rev: v3.19.1
     hooks:
       - id: pyupgrade
         args:
           - --py310-plus
   - repo: https://github.com/psf/black
-    rev: 24.8.0
+    rev: 25.1.0
     hooks:
       - id: black
         language_version: python3
         args:
           - --target-version=py310
   - repo: https://github.com/pycqa/flake8
-    rev: 7.1.1
+    rev: 7.2.0
     hooks:
       - id: flake8
         language_version: python3
@@ -37,7 +37,7 @@ repos:
           #  dependency. Make sure to update flake8-bugbear manually on a regular basis.
           - flake8-bugbear==24.8.19
   - repo: https://github.com/codespell-project/codespell
-    rev: v2.3.0
+    rev: v2.4.1
     hooks:
       -   id: codespell
           types_or: [rst, markdown]
@@ -48,7 +48,7 @@ repos:
     # pinned due to
     # https://github.com/python/typeshed/pull/9771 and
     # https://github.com/python/mypy/issues/15257 for DaskCollection.__dask_scheduler__
-    rev: v1.11.2
+    rev: v1.15.0
     hooks:
       - id: mypy
         # Override default --ignore-missing-imports
diff --git a/dask/_task_spec.py b/dask/_task_spec.py
index bf404f166..3950f2a4d 100644
--- a/dask/_task_spec.py
+++ b/dask/_task_spec.py
@@ -799,7 +799,7 @@ class Task(GraphNode):
                 key or self.key,
                 self.func,
                 *new_args,
-                **new_kwargs,
+                **new_kwargs,  # type: ignore[arg-type]
                 **extra_kwargs,
             )
         elif key is None or key == self.key:
@@ -954,15 +954,15 @@ class Dict(NestedContainer, Mapping):
     def __iter__(self):
         yield from self.args[::2]
 
+    def __len__(self):
+        return len(self.args) // 2
+
     def __getitem__(self, key):
         for k, v in batched(self.args, 2, strict=True):
             if k == key:
                 return v
         raise KeyError(key)
 
-    def __len__(self):
-        return len(self.args) // 2
-
     @staticmethod
     def constructor(args):
         return dict(batched(args, 2, strict=True))
diff --git a/dask/array/__init__.py b/dask/array/__init__.py
index 56054d807..4f496dce6 100644
--- a/dask/array/__init__.py
+++ b/dask/array/__init__.py
@@ -27,7 +27,7 @@ def _array_expr_enabled() -> bool:
             )
         return ARRAY_EXPR_ENABLED  # type: ignore[return-value]
 
-    return use_array_expr if use_array_expr is not None else False
+    return bool(use_array_expr if use_array_expr is not None else False)
 
 
 def array_expr_enabled():
diff --git a/dask/array/chunk.py b/dask/array/chunk.py
index 398ae9339..ddefd6308 100644
--- a/dask/array/chunk.py
+++ b/dask/array/chunk.py
@@ -1,4 +1,4 @@
-""" A set of NumPy functions to apply per chunk """
+"""A set of NumPy functions to apply per chunk"""
 
 from __future__ import annotations
 
diff --git a/dask/backends.py b/dask/backends.py
index 4164bfc65..2c3706944 100644
--- a/dask/backends.py
+++ b/dask/backends.py
@@ -127,7 +127,7 @@ class CreationDispatch(Generic[BackendEntrypointType]):
         """Register dispatchable function"""
 
         def decorator(
-            fn: Callable[BackendFuncParams, BackendFuncReturn]
+            fn: Callable[BackendFuncParams, BackendFuncReturn],
         ) -> Callable[BackendFuncParams, BackendFuncReturn]:
             dispatch_name = name or fn.__name__
             dispatcher = self.dispatch(backend)
diff --git a/dask/bag/__init__.py b/dask/bag/__init__.py
index 95cdfdaab..34a42c560 100644
--- a/dask/bag/__init__.py
+++ b/dask/bag/__init__.py
@@ -20,7 +20,10 @@ __all__ = [
 
 try:
     from dask.bag.avro import read_avro
-    from dask.bag.core import Bag, Item
+    from dask.bag.core import (
+        Bag,
+        Item,
+    )
     from dask.bag.core import bag_map as map
     from dask.bag.core import bag_range as range
     from dask.bag.core import bag_zip as zip
diff --git a/dask/bag/core.py b/dask/bag/core.py
index 04c5138ae..aa54972b4 100644
--- a/dask/bag/core.py
+++ b/dask/bag/core.py
@@ -2578,7 +2578,7 @@ def random_state_data_python(
 
         random_data = np_rng.bytes(624 * n * 4)  # `n * 624` 32-bit integers
         arr = np.frombuffer(random_data, dtype=np.uint32).reshape((n, -1))
-        return [(3, tuple(row) + (624,), None) for row in arr.tolist()]  # type: ignore
+        return [(3, tuple(row) + (624,), None) for row in arr.tolist()]
 
     except ImportError:
         # Pure python (much slower)
diff --git a/dask/dataframe/dask_expr/_collection.py b/dask/dataframe/dask_expr/_collection.py
index 7afcb6898..91fea7840 100644
--- a/dask/dataframe/dask_expr/_collection.py
+++ b/dask/dataframe/dask_expr/_collection.py
@@ -28,6 +28,7 @@ import dask.dataframe.methods as methods
 from dask import compute, get_annotations
 from dask._collections import new_collection
 from dask._expr import OptimizerStage
+from dask._task_spec import Dict, TaskRef
 from dask.array import Array
 from dask.base import DaskMethodsMixin, is_dask_collection, named_schedulers
 from dask.core import flatten
@@ -65,6 +66,7 @@ from dask.dataframe.dask_expr._expr import (
     ToDatetime,
     ToNumeric,
     ToTimedelta,
+    _DelayedExpr,
     no_default,
 )
 from dask.dataframe.dask_expr._merge import JoinRecursive, Merge
@@ -6166,6 +6168,18 @@ def map_partitions(
         # will need to call `Repartition` on operands that are not
         # aligned with `self.expr`.
         raise NotImplementedError()
+    args = [_DelayedExpr(a) if isinstance(a, Delayed) else a for a in args]
+    newkwargs = {}
+    delayed_kwargs = []
+    for k, v in kwargs.items():
+        if isinstance(v, Delayed):
+            dexpr = _DelayedExpr(v)
+            delayed_kwargs.append(dexpr)
+            newkwargs[k] = TaskRef(dexpr.__dask_keys__()[0])
+        else:
+            newkwargs[k] = v
+    del kwargs
+
     new_expr = expr.MapPartitions(
         args[0],
         func,
@@ -6175,9 +6189,11 @@ def map_partitions(
         clear_divisions,
         align_dataframes,
         parent_meta,
-        kwargs.pop("token", None),
-        kwargs,
+        newkwargs.pop("token", None),
+        Dict(newkwargs),
+        len(args) - 1,
         *args[1:],
+        *delayed_kwargs,
     )
     return new_collection(new_expr)
 
diff --git a/dask/dataframe/dask_expr/_expr.py b/dask/dataframe/dask_expr/_expr.py
index 0474830b0..461bc58dc 100644
--- a/dask/dataframe/dask_expr/_expr.py
+++ b/dask/dataframe/dask_expr/_expr.py
@@ -656,12 +656,14 @@ class MapPartitions(Blockwise):
         "parent_meta",
         "token",
         "kwargs",
+        "nargs",
     ]
     _defaults: dict = {
         "kwargs": None,
         "align_dataframes": True,
         "parent_meta": None,
         "token": None,
+        "nargs": 0,
     }
 
     @functools.cached_property
@@ -687,14 +689,20 @@ class MapPartitions(Blockwise):
 
     @functools.cached_property
     def args(self):
-        return [self.frame] + self.operands[len(self._parameters) :]
+        return [self.frame] + self.operands[
+            len(self._parameters) : len(self._parameters) + self.nargs
+        ]
 
     @functools.cached_property
     def _meta(self):
         meta = self.operand("meta")
         return _get_meta_map_partitions(
             self.args,
-            [e for e in self.args if isinstance(e, Expr)],
+            [
+                e
+                for e in self.args
+                if isinstance(e, Expr) and not isinstance(e, _DelayedExpr)
+            ],
             self.func,
             self.kwargs,
             meta,
@@ -725,7 +733,7 @@ class MapPartitions(Blockwise):
 
     def _task(self, name: Key, index: int) -> Task:
         args = [self._blockwise_arg(op, index) for op in self.args]
-        kwargs = (self.kwargs if self.kwargs is not None else {}).copy()
+        kwargs = dict(self.kwargs if self.kwargs is not None else {})
         if self._has_partition_info:
             kwargs["partition_info"] = {
                 "number": index,
@@ -978,6 +986,7 @@ class MapOverlap(MapPartitions):
             None,
             self.token,
             self._kwargs,
+            len(self.args[1:]),
             *self.args[1:],
         )
 
diff --git a/dask/dataframe/dask_expr/_groupby.py b/dask/dataframe/dask_expr/_groupby.py
index 1364dca17..fa543c9e1 100644
--- a/dask/dataframe/dask_expr/_groupby.py
+++ b/dask/dataframe/dask_expr/_groupby.py
@@ -1391,6 +1391,7 @@ class GroupByCumulative(Expr, GroupByBase):
             None,
             None,
             {"chunk": self.chunk, "columns": columns, **dropna, **self.numeric_only},
+            len(self.by),
             *self.by,
         )
         cum_raw = frame
@@ -1420,6 +1421,7 @@ class GroupByCumulative(Expr, GroupByBase):
             None,
             None,
             {"chunk": M.last, "columns": columns, **dropna},
+            len(by),
             *by,
         )
         return GroupByCumulativeFinalizer(
diff --git a/dask/dataframe/dask_expr/_merge_asof.py b/dask/dataframe/dask_expr/_merge_asof.py
index 436eb229c..eb9684cc6 100644
--- a/dask/dataframe/dask_expr/_merge_asof.py
+++ b/dask/dataframe/dask_expr/_merge_asof.py
@@ -132,6 +132,7 @@ class MergeAsof(Merge):
                 None,
                 None,
                 self._kwargs,
+                1,
                 self.right,
             )
 
@@ -154,6 +155,7 @@ class MergeAsof(Merge):
                 None,
                 None,
                 {"left_index": True, "right_index": True},
+                1,
                 right,
             )
 
diff --git a/dask/dataframe/dask_expr/io/parquet.py b/dask/dataframe/dask_expr/io/parquet.py
index bf7bfe372..e5e517fde 100644
--- a/dask/dataframe/dask_expr/io/parquet.py
+++ b/dask/dataframe/dask_expr/io/parquet.py
@@ -1735,7 +1735,7 @@ def _collect_pq_statistics(
         groups = defaultdict(list)
         for part in parts:
             for p in [part] if isinstance(part, dict) else part:
-                path = p.get("piece")[0]
+                path = p.get("piece")[0]  # type: ignore[index]
                 groups[path].append(p)
         group_keys = list(groups.keys())
 
diff --git a/dask/order.py b/dask/order.py
index d76ac853e..b6db4578b 100644
--- a/dask/order.py
+++ b/dask/order.py
@@ -227,7 +227,6 @@ def order(
             return rv
 
     def add_to_result(item: Key) -> None:
-        nonlocal crit_path_counter
         # Earlier versions recursed into this method but this could cause
         # recursion depth errors. This is the only reason for the while loop
         next_items = [item]
@@ -269,7 +268,6 @@ def order(
         # is purely cosmetical and used for some visualizations and I haven't
         # settled on how to implement this best so I didn't want to have large
         # indentations that make things harder to read
-        nonlocal _crit_path_counter_offset
 
         def wrapper(*args: Any, **kwargs: Any) -> None:
             nonlocal _crit_path_counter_offset
@@ -668,30 +666,24 @@ def _connecting_to_roots(
         if len(transitive_deps_ids) == 1:
             result[key] = transitive_deps[0]
         else:
-            prev: set | frozenset | None = None
-            new_set = None
-            for tdeps in transitive_deps:
-                if prev is None:
-                    prev = tdeps
-                    continue
-                if tdeps.issubset(prev):
-                    continue
-                if new_set is None:
-                    prev = new_set = set(prev)
-                prev.update(tdeps)
-            if new_set is not None:
+            d = transitive_deps[0]
+            if all(tdeps.issubset(d) for tdeps in transitive_deps[1:]):
+                result[key] = d
+            else:
+                res = set(d)
+                for tdeps in transitive_deps[1:]:
+                    res.update(tdeps)
                 # frozenset is unfortunately triggering a copy. In the event of
                 # a cache hit, this is wasted time but we can't hash the set
                 # otherwise (unless we did it manually) and can therefore not
                 # deduplicate without this copy
-                frozen_res = frozenset(new_set)
+                frozen_res = frozenset(res)
+                del res, tdeps
                 try:
                     result[key] = dedup_mapping[frozen_res]
                 except KeyError:
                     dedup_mapping[frozen_res] = frozen_res
                     result[key] = frozen_res
-            else:
-                result[key] = prev  # type: ignore
     del dedup_mapping
 
     empty_set: frozenset[Key] = frozenset()
diff --git a/dask/tokenize.py b/dask/tokenize.py
index ab07f363a..5f780adc2 100644
--- a/dask/tokenize.py
+++ b/dask/tokenize.py
@@ -64,7 +64,7 @@ def tokenize(
         tokenized, e.g. two identical objects will return different tokens.
         Defaults to the `tokenize.ensure-deterministic` configuration parameter.
     """
-    global _SEEN, _ENSURE_DETERMINISTIC
+    global _SEEN
     with tokenize_lock:
         seen_before, _SEEN = _SEEN, {}
         token = None


Test patch:
diff --git a/dask/dataframe/dask_expr/tests/test_distributed.py b/dask/dataframe/dask_expr/tests/test_distributed.py
index a1333c1ea..91128141a 100644
--- a/dask/dataframe/dask_expr/tests/test_distributed.py
+++ b/dask/dataframe/dask_expr/tests/test_distributed.py
@@ -350,8 +350,9 @@ def test_merge_combine_similar_squash_merges(add_repartition):
     )
 
 
+@pytest.mark.parametrize("kind", ["future", "delayed"])
 @gen_cluster(client=True)
-async def test_future_in_map_partitions(c, s, a, b):
+async def test_future_in_map_partitions(c, s, a, b, kind):
     # xgboost uses this pattern
 
     def test_func(n):
@@ -361,11 +362,22 @@ async def test_future_in_map_partitions(c, s, a, b):
 
     df = from_pandas(pd.DataFrame({"a": [1, 2, 3, 4]}), npartitions=2)
 
-    f = c.submit(test_func, 100)
-    q = map_partitions(lambda x, y: y + x.sum(), f, df, meta=df._meta)
+    if kind == "future":
+        f = c.submit(test_func, 100)
+        z = c.submit(test_func, 10)
+    elif kind == "delayed":
+        f = dask.delayed(test_func)(100)
+        z = dask.delayed(test_func)(10)
+    else:
+        raise ValueError("kind must be 'future' or 'delayed'")
+
+    def _foo(x, y, z):
+        return x.sum() + y + z.sum()
+
+    q = map_partitions(_foo, f, df, z=z, meta=df._meta)
     result = c.compute(q)
     result = await result
-    expected = pd.DataFrame({"a": [4951, 4952, 4953, 4954]})
+    expected = pd.DataFrame({"a": [4996, 4997, 4998, 4999]})
     pd.testing.assert_frame_equal(result, expected)
 
 
diff --git a/dask/dataframe/io/tests/test_parquet.py b/dask/dataframe/io/tests/test_parquet.py
index fb2127e60..e2481da11 100644
--- a/dask/dataframe/io/tests/test_parquet.py
+++ b/dask/dataframe/io/tests/test_parquet.py
@@ -1382,7 +1382,7 @@ def test_filters_categorical(tmpdir, write_engine, read_engine):
         tmpdir,
         index="dummy",
         engine=read_engine,
-        filters=[(("DatePart", "<=", "2018-01-02"))],
+        filters=[("DatePart", "<=", "2018-01-02")],
         calculate_divisions=True,
     )
     assert len(ddftest_read) == 2
@@ -2543,7 +2543,7 @@ def test_filter_nonpartition_columns(
         index=False,
         engine=read_engine,
         calculate_divisions=calculate_divisions,
-        filters=[(("time", "<", 5))],
+        filters=[("time", "<", 5)],
     )
     df_read = ddf_read.compute()
     assert len(df_read) == len(df_read[df_read["time"] < 5])
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index acbf1eddd..acf7111ec 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -3647,7 +3647,7 @@ def test_inplace_operators():
     "idx",
     [
         np.arange(100),
-        sorted(np.random.random(size=100)),
+        sorted(np.random.random(size=100)),  # type: ignore[type-var]
         pd.date_range("20150101", periods=100),
     ],
 )
diff --git a/dask/tests/test_layers.py b/dask/tests/test_layers.py
index 0597ee997..80241958b 100644
--- a/dask/tests/test_layers.py
+++ b/dask/tests/test_layers.py
@@ -142,7 +142,7 @@ def _pq_pyarrow(tmpdir):
     dd.from_pandas(pd.DataFrame({"a": range(10)}), npartitions=2).to_parquet(
         str(tmpdir)
     )
-    filters = [(("a", "<=", 2))]
+    filters = [("a", "<=", 2)]
 
     ddf1 = dd.read_parquet(str(tmpdir), filters=filters)
     return ddf1
diff --git a/dask/tests/test_task_spec.py b/dask/tests/test_task_spec.py
index becbefca8..4986c3efb 100644
--- a/dask/tests/test_task_spec.py
+++ b/dask/tests/test_task_spec.py
@@ -4,6 +4,7 @@ import itertools
 import pickle
 import sys
 from collections import namedtuple
+from collections.abc import Mapping
 
 import pytest
 
@@ -1013,6 +1014,21 @@ def test_dict_class():
 
     d = Dict([["columns", ["a", "b"]]])
     assert d() == {"columns": ["a", "b"]}
+    # Can be converted to a dict, e.g. also used as **kwargs
+    assert isinstance(t, Mapping)
+    assert dict(t) == {
+        "k": Task("key-1", func, "a", "b"),
+        "v": Task("key-2", func, "c", "d"),
+    }
+    assert len(t) == len(dict(t)) == 2
+
+    def as_kwargs(**kwargs):
+        return kwargs == {
+            "k": Task("key-1", func, "a", "b"),
+            "v": Task("key-2", func, "c", "d"),
+        }
+
+    assert as_kwargs(**t)
 
 
 def test_block_io_fusion():