Repository: dask/dask
Pull request #10865 (2024-01-26 15:43:41)

Deterministic tokenize() for global random functions
- Blocked by #10863
- Blocks https://github.com/dask/distributed/pull/8185

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 17m 39s :stopwatch: - 17m 0s
 12 982 tests +  7   12 036 :white_check_mark: + 3     929 :zzz: ±0  16 :x: +3  1 :fire: +1 
160 156 runs  +105  143 560 :white_check_mark: +91  16 576 :zzz: +7  17 :x: +4  3 :fire: +3 

For more details on these failures and errors, see [this check](https://github.com/dask/dask/runs/20908543237).

Results for commit c2414409. ± Comparison against base commit 4978f6a6.

<details>
  <summary>This pull request <b>removes</b> 70 and <b>adds</b> 77 tests. <i>Note that renamed tests count towards both.</i></summary>

```
dask.tests.test_base ‑ test_normalize_base
dask.tests.test_base ‑ test_normalize_function
dask.tests.test_base ‑ test_normalize_function_dataclass_field_no_repr
dask.tests.test_base ‑ test_normalize_function_limited_size
dask.tests.test_base ‑ test_tokenize
dask.tests.test_base ‑ test_tokenize_base_types[1.0]
dask.tests.test_base ‑ test_tokenize_base_types[1]
dask.tests.test_base ‑ test_tokenize_base_types[1j0]
dask.tests.test_base ‑ test_tokenize_base_types[1j1]
dask.tests.test_base ‑ test_tokenize_base_types[None]
…
```

```
dask.tests.test_tokenize ‑ test_dask_tokenize_method
dask.tests.test_tokenize ‑ test_normalize_base
dask.tests.test_tokenize ‑ test_normalize_function
dask.tests.test_tokenize ‑ test_normalize_function_dataclass_field_no_repr
dask.tests.test_tokenize ‑ test_normalize_function_limited_size
dask.tests.test_tokenize ‑ test_tokenize
dask.tests.test_tokenize ‑ test_tokenize_base_types[1.0]
dask.tests.test_tokenize ‑ test_tokenize_base_types[1]
dask.tests.test_tokenize ‑ test_tokenize_base_types[1j0]
dask.tests.test_tokenize ‑ test_tokenize_base_types[1j1]
…
```
</details>

[test-results]:data:application/gzip;base64,H4sIAL7ks2UC/0WMSw7DIAwFrxKx7gLzMdDLRJSAhJpPRWBV9e4laSHLGT/Pm4Q4+53cB5C3gewl5oumkmyO23owaGmqqtd83pnRrPG4F+dOSTle8hlfVRrW/8Zg43zsrpFPaUuHqiaV9UwjBYl/7mnBJdIuf2lAqfqwtVUTLc2rcNuyxFyBOCZACGoeAnhAYN4Y9DxQ6p21Cp0Gr9SkJfl8Acg+lGwbAQAA


Patch:
diff --git a/dask/base.py b/dask/base.py
index bf06debce..712e0ba56 100644
--- a/dask/base.py
+++ b/dask/base.py
@@ -8,6 +8,7 @@ import inspect
 import os
 import pathlib
 import pickle
+import random
 import threading
 import uuid
 import warnings
@@ -1112,6 +1113,11 @@ def normalize_enum(e):
     return type(e).__name__, e.name, e.value
 
 
+@normalize_token.register(random.Random)
+def normalize_random_state(state):
+    return normalize_token(state.getstate())
+
+
 @normalize_token.register(object)
 def normalize_object(o):
     method = getattr(o, "__dask_tokenize__", None)
@@ -1119,6 +1125,14 @@ def normalize_object(o):
         return method()
 
     if callable(o):
+        # For bound methods, normalize object owning the method to allow for custom
+        # hooks. This is particularly important for random and numpy.random functions
+        # with global state.
+        self = getattr(o, "__self__", None)
+        name = getattr(o, "__name__", None)
+        if self is not None and not inspect.ismodule(self) and isinstance(name, str):
+            return normalize_token(self), name
+
         return normalize_function(o)
 
     if dataclasses.is_dataclass(o):
@@ -1351,6 +1365,10 @@ def register_numpy():
         except AttributeError:
             return normalize_function(x)
 
+    @normalize_token.register(np.random.RandomState)
+    def normalize_np_random_state(state):
+        return normalize_token(state.get_state())
+
     @normalize_token.register(np.random.BitGenerator)
     def normalize_bit_generator(bg):
         return normalize_token(bg.state)
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index ca09cb9c8..e5b5a40bb 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -1880,6 +1880,7 @@ Dask Name: {name}, {layers}"""
         else:
             return self
 
+    @_deprecated_kwarg("method", None, comment="Use ffill or bfill instead.")
     @derived_from(pd.DataFrame)
     def fillna(self, value=None, method=None, limit=None, axis=None):
         if method is None and limit is not None:
diff --git a/dask/utils.py b/dask/utils.py
index b93ce9fa1..0da2953c8 100644
--- a/dask/utils.py
+++ b/dask/utils.py
@@ -148,6 +148,7 @@ def _deprecated_kwarg(
     new_arg_name: str | None = None,
     mapping: Mapping[Any, Any] | Callable[[Any], Any] | None = None,
     stacklevel: int = 2,
+    comment: str | None = None,
 ) -> Callable[[F], F]:
     """
     Decorator to deprecate a keyword argument of a function.
@@ -163,6 +164,9 @@ def _deprecated_kwarg(
         If mapping is present, use it to translate old arguments to
         new arguments. A callable must do its own value checking;
         values not found in a dict will be forwarded unchanged.
+    comment :  str, optional
+        Additional message to deprecation message. Useful to pass
+        on suggestions with the deprecation warning.
 
     Examples
     --------
@@ -215,6 +219,8 @@ def _deprecated_kwarg(
             "mapping from old to new argument values must be dict or callable!"
         )
 
+    comment_ = f"\n{comment}" or ""
+
     def _deprecated_kwarg(func: F) -> F:
         @wraps(func)
         def wrapper(*args, **kwargs) -> Callable[..., Any]:
@@ -226,7 +232,7 @@ def _deprecated_kwarg(
                         f"the {repr(old_arg_name)} keyword is deprecated and "
                         "will be removed in a future version. Please take "
                         f"steps to stop the use of {repr(old_arg_name)}"
-                    )
+                    ) + comment_
                     warnings.warn(msg, FutureWarning, stacklevel=stacklevel)
                     kwargs[old_arg_name] = old_arg_value
                     return func(*args, **kwargs)
@@ -248,7 +254,7 @@ def _deprecated_kwarg(
                         f"use {repr(new_arg_name)} instead."
                     )
 
-                warnings.warn(msg, FutureWarning, stacklevel=stacklevel)
+                warnings.warn(msg + comment_, FutureWarning, stacklevel=stacklevel)
                 if kwargs.get(new_arg_name) is not None:
                     msg = (
                         f"Can only specify {repr(old_arg_name)} "
diff --git a/pyproject.toml b/pyproject.toml
index 0359c3dc1..82b277a6b 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -144,7 +144,6 @@ filterwarnings = [
     "ignore:Not prepending group keys:FutureWarning",
     "ignore:.*:dask.tests.warning_aliases.RemovedIn20Warning",
     "ignore:When grouping with a length-1 list-like, you will need to pass a length-1 tuple to get_group in a future version of pandas:FutureWarning",
-    'ignore:DataFrameGroupBy\.apply operated on the grouping columns\. This behavior is deprecated, and in a future version of pandas the grouping columns will be excluded from the operation\. Either pass `include_groups=False` to exclude the groupings or explicitly select the grouping columns after groupby to silence this warning\.:FutureWarning',
     'ignore:Passing a BlockManager to DataFrame is deprecated and will raise in a future version. Use public APIs instead:DeprecationWarning',  # https://github.com/apache/arrow/issues/35081
     'ignore:The previous implementation of stack is deprecated and will be removed in a future version of pandas\.:FutureWarning',
     "ignore:Minimal version of pyarrow will soon be increased to 14.0.1",


Test patch:
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index 808af64eb..52c29ba7a 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -2645,6 +2645,10 @@ def test_fillna():
         pytest.raises(NotImplementedError, lambda: ddf.fillna(0, limit=10))
         pytest.raises(NotImplementedError, lambda: ddf.fillna(0, limit=10, axis=1))
 
+    if not DASK_EXPR_ENABLED:
+        with pytest.warns(FutureWarning, match="'method' keyword is deprecated"):
+            ddf.fillna(method="ffill")
+
 
 def test_ffill():
     df = _compat.makeMissingDataframe()
diff --git a/dask/tests/test_tokenize.py b/dask/tests/test_tokenize.py
index 745970d1a..d077b8a02 100644
--- a/dask/tests/test_tokenize.py
+++ b/dask/tests/test_tokenize.py
@@ -3,7 +3,10 @@ from __future__ import annotations
 import dataclasses
 import datetime
 import decimal
+import operator
 import pathlib
+import pickle
+import random
 import subprocess
 import sys
 import textwrap
@@ -243,13 +246,8 @@ def test_tokenize_function_cloudpickle():
     # No error by default
     tokenize(a)
 
-    try:
-        import dask.dataframe as dd
-
-        if dd._dask_expr_enabled():
-            pytest.xfail("dask-expr does a check and serializes if possible")
-    except ImportError:
-        pass
+    if dd and dd._dask_expr_enabled():
+        return  # dask-expr does a check and serializes if possible
 
     with dask.config.set({"tokenize.ensure-deterministic": True}):
         with pytest.raises(RuntimeError, match="may not be deterministically hashed"):
@@ -388,13 +386,20 @@ def test_tokenize_method():
         def __dask_tokenize__(self):
             return self.x
 
+        def hello(self):
+            return "Hello world"
+
     a, b = Foo(1), Foo(2)
     assert tokenize(a) == tokenize(a)
     assert tokenize(a) != tokenize(b)
 
+    assert tokenize(a.hello) == tokenize(a.hello)
+    assert tokenize(a.hello) != tokenize(b.hello)
+
     for ensure in [True, False]:
         with dask.config.set({"tokenize.ensure-deterministic": ensure}):
             assert tokenize(a) == tokenize(a)
+            assert tokenize(a.hello) == tokenize(a.hello)
 
     # dispatch takes precedence
     before = tokenize(a)
@@ -705,3 +710,74 @@ def test_normalize_function_dataclass_field_no_repr():
 
     assert normalize_function(a1) == normalize_function(a1)
     assert normalize_function(a1) != normalize_function(a2)
+
+
+def test_tokenize_operator():
+    """Top-level functions in the operator module have a __self__ attribute, which is
+    the module itself
+    """
+    assert tokenize(operator.add) == tokenize(operator.add)
+    assert tokenize(operator.add) != tokenize(operator.mul)
+
+
+def test_tokenize_random_state():
+    a = random.Random(123)
+    b = random.Random(123)
+    c = random.Random(456)
+    assert tokenize(a) == tokenize(b)
+    assert tokenize(a) != tokenize(c)
+    a.random()
+    assert tokenize(a) != tokenize(b)
+
+
+@pytest.mark.skipif("not np")
+def test_tokenize_random_state_numpy():
+    a = np.random.RandomState(123)
+    b = np.random.RandomState(123)
+    c = np.random.RandomState(456)
+    assert tokenize(a) == tokenize(b)
+    assert tokenize(a) != tokenize(c)
+    a.random()
+    assert tokenize(a) != tokenize(b)
+
+
+@pytest.mark.parametrize(
+    "module",
+    ["random", pytest.param("np.random", marks=pytest.mark.skipif("not np"))],
+)
+def test_tokenize_random_functions(module):
+    """random.random() and other methods of the global random state do not compare as
+    equal to themselves after a pickle roundtrip"""
+    module = eval(module)
+
+    a = module.random
+    b = pickle.loads(pickle.dumps(a))
+    assert tokenize(a) == tokenize(b)
+
+    # Drawing elements or reseeding changes the global state
+    a()
+    c = pickle.loads(pickle.dumps(a))
+    assert tokenize(a) == tokenize(c)
+    assert tokenize(a) != tokenize(b)
+
+    module.seed(123)
+    d = pickle.loads(pickle.dumps(a))
+    assert tokenize(a) == tokenize(d)
+    assert tokenize(a) != tokenize(c)
+
+
+def test_tokenize_random_functions_with_state():
+    a = random.Random(123).random
+    b = random.Random(123).random
+    c = random.Random(456).random
+    assert tokenize(a) == tokenize(b)
+    assert tokenize(a) != tokenize(c)
+
+
+@pytest.mark.skipif("not np")
+def test_tokenize_random_functions_with_state_numpy():
+    a = np.random.RandomState(123).random
+    b = np.random.RandomState(123).random
+    c = np.random.RandomState(456).random
+    assert tokenize(a) == tokenize(b)
+    assert tokenize(a) != tokenize(c)