Repository: dask/dask

Issue:
Flag for raising an error when `normalize_object` doesn't find `__dask_tokenize__`
I ran into a bit of a nasty surprised when I realized some of my Delayed objects resulted in a different token when I called `tokenize()`. This is because in my data I had some objects which didn't have `__dask_tokenize__`. I fixed those errors, but I'm afraid they might come up again silently. It would be amazing if I could set a flag that would raise an error any time that I tried to tokenize something which contained anything that couldn't be tokenized deterministically. I'm thinking something in `dask.base.normalize_object` that raises an error instead of returning `uuid.uuid4().hex`.

Thanks!

Discussion:
hristog: Need to go through the rest of the codebase a bit more to get an idea how it'd be best to go about documenting the newly-proposed config.
hristog: I'm considering documenting the config as part of [this](https://docs.dask.org/en/latest/custom-collections.html?highlight=normalize_token#implementing-deterministic-hashing) section, because I haven't been able to find a more suitable one.
Any suggestions would be greatly appreciated.
hristog: Thanks for looking into this PR, @jrbourbeau!

I've scaled down the functionality, associated with the newly-proposed config, as suggested, and updated its value to `bool`. The accompanying unit-tests have been updated accordingly, too.

One tiny concern I've got about the naming (`allow-random`) is that - as mentioned in [this comment](https://github.com/dask/dask/issues/6555#issuecomment-801116347) of mine - there are other sources of non-determinism that aren't going to be impacted by the proposed logic.

I suppose this could always be added, given actual demand, if GitHub issues are submitted requesting those to be supported at any point later on.

When you get a chance, please, let me know what your thoughts on regarding the best possible way to document the newly-introduced config.
hristog: Hi @jrbourbeau - I'm writing to check if you've had a chance to look into the further changes that I pushed in https://github.com/dask/dask/pull/7413/commits/41b135b4d085fc26a7cfe6ce7475c70e649ef89e, and whether you'd like any additional refinements to be done on top of that.
Thanks!


hristog: Hi @jrbourbeau, I'm writing to check if you've had a chance to have a look at the most recent updates.
As discussed, I've scaled down the tests a bit further - the summary of my updates can be found in the most recent two comments, from [this thread](https://github.com/dask/dask/pull/7413#discussion_r609922956).

Please, let me know if the tests should be further stripped down, despite the discussed concerns.
ncclementi: @jrbourbeau checking in here, it looks like @hristog updated the code according to your requests and CI is green. Do you think this is ready to go, or are there any other issues that should be addressed?
jcrist: I've renamed the config field to `tokenize.ensure-deterministic` to better reflect the meaning of the parameter. I've also added it to the `dask.yaml` config file and the `dask-schema.yaml` schema description. I believe this should be good to go once tests pass.
hristog: > I've renamed the config field to `tokenize.ensure-deterministic` to better reflect the meaning of the parameter. I've also added it to the `dask.yaml` config file and the `dask-schema.yaml` schema description. I believe this should be good to go once tests pass.

Thanks, @jcrist!
jrbourbeau: Thanks for the updates @jcrist! I pushed a small commit to avoid `mock`ing in the tests added here. Will merge after CI finishes 
jcrist: Test failure is unrelated - merging. Thanks @hristog!

Patch:
diff --git a/dask/base.py b/dask/base.py
index a9368bfe8..8a9c10055 100644
--- a/dask/base.py
+++ b/dask/base.py
@@ -839,7 +839,14 @@ def normalize_seq(seq):
         try:
             return list(map(normalize_token, seq))
         except RecursionError:
-            return str(uuid.uuid4())
+            if not config.get("tokenize.ensure-deterministic"):
+                return uuid.uuid4().hex
+
+            raise RuntimeError(
+                f"Sequence {str(seq)} cannot be deterministically hashed. Please, see "
+                "https://docs.dask.org/en/latest/custom-collections.html#implementing-deterministic-hashing "
+                "for more information"
+            )
 
     return type(seq).__name__, func(seq)
 
@@ -859,7 +866,18 @@ def normalize_object(o):
     method = getattr(o, "__dask_tokenize__", None)
     if method is not None:
         return method()
-    return normalize_function(o) if callable(o) else uuid.uuid4().hex
+
+    if callable(o):
+        return normalize_function(o)
+
+    if not config.get("tokenize.ensure-deterministic"):
+        return uuid.uuid4().hex
+
+    raise RuntimeError(
+        f"Object {str(o)} cannot be deterministically hashed. Please, see "
+        "https://docs.dask.org/en/latest/custom-collections.html#implementing-deterministic-hashing "
+        "for more information"
+    )
 
 
 function_cache = {}
@@ -1032,7 +1050,15 @@ def register_numpy():
                     data = hash_buffer_hex(pickle.dumps(x, pickle.HIGHEST_PROTOCOL))
                 except Exception:
                     # pickling not supported, use UUID4-based fallback
-                    data = uuid.uuid4().hex
+                    if not config.get("tokenize.ensure-deterministic"):
+                        data = uuid.uuid4().hex
+                    else:
+                        raise RuntimeError(
+                            f"``np.ndarray`` with object ``dtype`` {str(x)} cannot "
+                            "be deterministically hashed. Please, see "
+                            "https://docs.dask.org/en/latest/custom-collections.html#implementing-deterministic-hashing "  # noqa: E501
+                            "for more information"
+                        )
         else:
             try:
                 data = hash_buffer_hex(x.ravel(order="K").view("i1"))
diff --git a/dask/dask-schema.yaml b/dask/dask-schema.yaml
index 4472ad286..ab747f199 100644
--- a/dask/dask-schema.yaml
+++ b/dask/dask-schema.yaml
@@ -10,6 +10,17 @@ properties:
       When the value is "null" (default), dask will create a directory from
       where dask was launched: `cwd/dask-worker-space`
 
+  tokenize:
+    type: object
+    properties:
+      ensure-deterministic:
+        type:
+        - boolean
+        description: |
+          If ``true``, tokenize will error instead of falling back to uuids
+          when a deterministic token cannot be generated. Defaults to
+          ``false``.
+
   dataframe:
     type: object
     properties:
diff --git a/dask/dask.yaml b/dask/dask.yaml
index dbdf158af..3f5b4f365 100644
--- a/dask/dask.yaml
+++ b/dask/dask.yaml
@@ -1,5 +1,8 @@
 temporary-directory: null  # Directory for local disk like /tmp, /scratch, or /local
 
+tokenize:
+  ensure-deterministic: false  # If true, tokenize will error instead of falling back to uuids
+
 dataframe:
   shuffle-compression: null  # compression for on disk-shuffling. Partd supports ZLib, BZ2, SNAPPY, BLOSC
   parquet:


Test patch:
diff --git a/dask/tests/test_base.py b/dask/tests/test_base.py
index 062d027e7..0698f7582 100644
--- a/dask/tests/test_base.py
+++ b/dask/tests/test_base.py
@@ -128,9 +128,8 @@ def test_tokenize_numpy_scalar_string_rep():
 
 @pytest.mark.skipif("not np")
 def test_tokenize_numpy_array_on_object_dtype():
-    assert tokenize(np.array(["a", "aa", "aaa"], dtype=object)) == tokenize(
-        np.array(["a", "aa", "aaa"], dtype=object)
-    )
+    a = np.array(["a", "aa", "aaa"], dtype=object)
+    assert tokenize(a) == tokenize(a)
     assert tokenize(np.array(["a", None, "aaa"], dtype=object)) == tokenize(
         np.array(["a", None, "aaa"], dtype=object)
     )
@@ -138,6 +137,17 @@ def test_tokenize_numpy_array_on_object_dtype():
         np.array([(1, "a"), (1, None), (1, "aaa")], dtype=object)
     ) == tokenize(np.array([(1, "a"), (1, None), (1, "aaa")], dtype=object))
 
+    # Trigger non-deterministic hashing for object dtype
+    class NoPickle:
+        pass
+
+    x = np.array(["a", None, NoPickle], dtype=object)
+    assert tokenize(x) != tokenize(x)
+
+    with dask.config.set({"tokenize.ensure-deterministic": True}):
+        with pytest.raises(RuntimeError, match="cannot be deterministically hashed"):
+            tokenize(x)
+
 
 @pytest.mark.skipif("not np")
 def test_tokenize_numpy_memmap_offset(tmpdir):
@@ -226,6 +236,23 @@ def test_normalize_base():
         assert normalize_token(i) is i
 
 
+def test_tokenize_object():
+    o = object()
+    # Defaults to non-deterministic tokenization
+    assert normalize_token(o) != normalize_token(o)
+
+    with dask.config.set({"tokenize.ensure-deterministic": True}):
+        with pytest.raises(RuntimeError, match="cannot be deterministically hashed"):
+            normalize_token(o)
+
+
+def test_tokenize_callable():
+    def my_func(a, b, c=1):
+        return a + b + c
+
+    assert tokenize(my_func) == tokenize(my_func)  # Consistent token
+
+
 @pytest.mark.skipif("not pd")
 def test_tokenize_pandas():
     a = pd.DataFrame({"x": [1, 2, 3], "y": ["4", "asd", None]}, index=[1, 2, 3])
@@ -335,6 +362,10 @@ def test_tokenize_method():
     assert tokenize(a) == tokenize(a)
     assert tokenize(a) != tokenize(b)
 
+    for ensure in [True, False]:
+        with dask.config.set({"tokenize.ensure-deterministic": ensure}):
+            assert tokenize(a) == tokenize(a)
+
     # dispatch takes precedence
     before = tokenize(a)
     normalize_token.register(Foo, lambda self: self.x + 1)
@@ -439,12 +470,16 @@ def test_tokenize_dense_sparse_array(cls_name):
     assert tokenize(a) != tokenize(b)
 
 
-def test_tokenize_object_with_recursion_error_returns_uuid():
+def test_tokenize_object_with_recursion_error():
     cycle = dict(a=None)
     cycle["a"] = cycle
 
     assert len(tokenize(cycle)) == 32
 
+    with dask.config.set({"tokenize.ensure-deterministic": True}):
+        with pytest.raises(RuntimeError, match="cannot be deterministically hashed"):
+            tokenize(cycle)
+
 
 def test_tokenize_datetime_date():
     # Same date