Repository: dask/dask
Pull request: Cross-scheduler inconsistent result structure
Raises a warning when `compute` or `persist` are called with a scheduler different from "dask.distributed" or "distributed" in Dask.distributed mode

- [X] Closes #8474
- [X] Tests added / passed
- [X] Passes `pre-commit run --all-files`


Problem statement:
Cross-scheduler inconsistent result structure
**What happened**:

Due to a mistake in our code, we were persisting a dask dataframe in one scheduler, but then ran the compute while specifying threads scheduler. What was weird was that the computation returned a pandas Series of Futures instead of the expected DataFrame (or exception).

**What you expected to happen**:

I'm not sure what the best behavior is. Could be erroring on conflicting scheduler, running on the original scheduler, or running on the specified scheduler. Either way, the result structure should be consistent.

**Minimal Complete Verifiable Example**:

```python
from distributed import Client, LocalCluster
import pandas as pd
import dask.dataframe as dd

cluster = LocalCluster()
client=Client(cluster)

df = pd.DataFrame({'a': [1,2,3]})

ddf = dd.from_pandas(df, npartitions=3)

ddf = client.persist(ddf)

# Works, and returns a DF as expected
print(ddf.compute())

# Returns a pandas Series with a row per-partition with a future as the result
print(ddf.compute(scheduler="threads"))
```

**Anything else we need to know?**:

**Environment**:

- Dask version:
- Python version:
- Operating System:
- Install method (conda, pip, source):

Hints from issue discussion:
Hmmm that's an interesting issue! Raising an error seems like the best bet. So then the question is where to raise the error from. I think `compute` should probably be in charge of making sure that the task graph doesn't contain references to objects that the scheduler can't understand. The other option is to put that logic into the `get` itself, 
Thanks @davidhao3300 for opening up this issue! Would you feel comfortable opening up a PR, with the suggestion from @jsignell?
@jrbourbeau pointed me to https://github.com/dask/dask/issues/7082 as a broader idea that would help with this specific issue. It is probably overkill if this is the main improvement, but keeping track of some metadata about persisted layers could make cases like this more obvious.
Hello,
I would love to have a look on this issue! Could you assign it for me please? 
Thanks! :D

Feel free to just open a pull request @ppsbs as soon as you start work.
Thanks, @jsignell !

I've worked a few hours on this task in the last 2 days and (based on my little experience with dask) I cannot see an unexpected behavior. As `ddf.compute(scheduler="threads")` runs asynchronously and `compute` always update the scheduler inside of `get_scheduler`, for me it is expected that it would return a future. And, on the other hand, as `ddf.compute()` runs synchronously, for me it's expected that it would get the result from the futures created by `client.persist(ddf)`. 

Or, maybe I haven't understood which would be the "objects that the scheduler can't understand" that you mentioned, @jsignell '^ ^   

Thank you!

cc @davidhao3300 
Once you are understand what is going on it might make sense, but it is not the behavior that a user would naively expect. This is not likely to be what the user intends to get back:

```python
In [6]: ddf.compute(scheduler="threads")
Out[6]: 
0    <Future: finished, type: pandas.core.frame.Dat...
1    <Future: finished, type: pandas.core.frame.Dat...
dtype: object
```

To close this issue, we would like to be able to at least give the user some warning that the operation is likely not what they intended.
I get your point @jsignell ! So, we could maybe raise an exception in the case where `compute` (and consequently `get_async`) are called with a dask graph whose result has already started being computed. Do you think that is a good path? Another option that I can think of is forcing `start_state_from_dask` to always convert cached futures to "results" (hence giving to `ddf.compute(scheduler="threads")` the same behavior of `ddf.compute()`), but I'm not sure of the impact of that modification.
I think raising an exception or even just a warning is the right path.

Commit message: Cross-scheduler inconsistent result structure (#8474)

Raises a warning when `compute` or `persist` are called with a scheduler different from "dask.distributed" or "distributed" in Dask.distributed

Patch to review:
diff --git a/dask/base.py b/dask/base.py
index 0fb5fc65d93..9ccdb7b2a16 100644
--- a/dask/base.py
+++ b/dask/base.py
@@ -4,6 +4,7 @@
 import pickle
 import threading
 import uuid
+import warnings
 from collections import OrderedDict
 from concurrent.futures import Executor
 from contextlib import contextmanager
@@ -1248,7 +1249,13 @@ def get_scheduler(get=None, scheduler=None, collections=None, cls=None):
             return scheduler.get
         elif isinstance(scheduler, str):
             scheduler = scheduler.lower()
+
             if scheduler in named_schedulers:
+                if config.get("scheduler", None) in ("dask.distributed", "distributed"):
+                    warnings.warn(
+                        "Passing a local execution scheduler in Dask.distributed might "
+                        "lead to unexpected results."
+                    )
                 return named_schedulers[scheduler]
             elif scheduler in ("dask.distributed", "distributed"):
                 from distributed.worker import get_client
diff --git a/dask/dataframe/io/tests/test_hdf.py b/dask/dataframe/io/tests/test_hdf.py
index 0d083a8312d..a42648abd11 100644
--- a/dask/dataframe/io/tests/test_hdf.py
+++ b/dask/dataframe/io/tests/test_hdf.py
@@ -478,6 +478,10 @@ def test_to_hdf_exceptions():
                 a.to_hdf(hdf, "/data_*_*")
 
 
+@pytest.mark.filterwarnings(
+    "ignore:Passing a local execution scheduler in Dask.distributed might "
+    "lead to unexpected results."
+)
 @pytest.mark.parametrize("scheduler", ["sync", "threads", "processes"])
 @pytest.mark.parametrize("npartitions", [1, 4, 10])
 def test_to_hdf_schedulers(scheduler, npartitions):
diff --git a/dask/tests/test_base.py b/dask/tests/test_base.py
index 41968c72c94..cd3bf3bcd61 100644
--- a/dask/tests/test_base.py
+++ b/dask/tests/test_base.py
@@ -1384,6 +1384,19 @@ def test_get_scheduler():
     assert get_scheduler() is None
 
 
+def test_get_scheduler_with_distributed_active():
+
+    with dask.config.set(scheduler="dask.distributed"):
+        warning_message = (
+            "Passing a local execution scheduler in Dask.distributed might "
+            "lead to unexpected results."
+        )
+        with pytest.warns(UserWarning, match=warning_message) as user_warnings_a:
+            get_scheduler(scheduler="threads")
+            get_scheduler(scheduler="sync")
+        assert len(user_warnings_a) == 2
+
+
 def test_callable_scheduler():
     called = [False]
 
diff --git a/dask/tests/test_distributed.py b/dask/tests/test_distributed.py
index e274e782f13..b8706a070bb 100644
--- a/dask/tests/test_distributed.py
+++ b/dask/tests/test_distributed.py
@@ -130,8 +130,20 @@ def test_futures_to_delayed_array(c):
     assert_eq(A.compute(), np.concatenate([x, x], axis=0))
 
 
+# with pytest.warns(UserWarning, match=warning_message) as user_warnings_b:
+#     get_scheduler(scheduler="dask.distributed")
+#     get_scheduler(scheduler="distributed")
+#     get_scheduler()
+# assert len(user_warnings_b) == 0
+
+
+@pytest.mark.filterwarnings(
+    "ignore:Passing a local execution scheduler in Dask.distributed might "
+    "lead to unexpected results."
+)
 @gen_cluster(client=True)
 async def test_local_get_with_distributed_active(c, s, a, b):
+
     with dask.config.set(scheduler="sync"):
         x = delayed(inc)(1).persist()
     await asyncio.sleep(0.01)
@@ -151,6 +163,10 @@ def test_to_hdf_distributed(c):
     test_to_hdf()
 
 
+@pytest.mark.filterwarnings(
+    "ignore:Passing a local execution scheduler in Dask.distributed might "
+    "lead to unexpected results."
+)
 @pytest.mark.parametrize(
     "npartitions",
     [
@@ -293,6 +309,10 @@ def reliable_double(x):
         "full",
     ],
 )
+@pytest.mark.filterwarnings(
+    "ignore:Passing a local execution scheduler in Dask.distributed might "
+    "lead to unexpected results."
+)
 @pytest.mark.parametrize("fuse", [True, False, None])
 def test_blockwise_array_creation(c, io, fuse):
     np = pytest.importorskip("numpy")
@@ -318,9 +338,14 @@ def test_blockwise_array_creation(c, io, fuse):
         dsk = dask.array.optimize(darr.dask, darr.__dask_keys__())
         # dsk should be a dict unless fuse is explicitly False
         assert isinstance(dsk, dict) == (fuse is not False)
+
         da.assert_eq(darr, narr)
 
 
+@pytest.mark.filterwarnings(
+    "ignore:Passing a local execution scheduler in Dask.distributed might "
+    "lead to unexpected results."
+)
 @pytest.mark.parametrize(
     "io",
     ["parquet-pyarrow", "parquet-fastparquet", "csv", "hdf"],
@@ -365,6 +390,7 @@ def test_blockwise_dataframe_io(c, tmpdir, io, fuse, from_futures):
         dsk = dask.dataframe.optimize(ddf.dask, ddf.__dask_keys__())
         # dsk should not be a dict unless fuse is explicitly True
         assert isinstance(dsk, dict) == bool(fuse)
+
         dd.assert_eq(ddf, df, check_index=False)
 
 
@@ -478,6 +504,10 @@ def add(x, y, z, extra_arg):
     assert res == 21
 
 
+@pytest.mark.filterwarnings(
+    "ignore:Passing a local execution scheduler in Dask.distributed might "
+    "lead to unexpected results."
+)
 @gen_cluster(client=True)
 async def test_blockwise_concatenate(c, s, a, b):
     """Test a blockwise operation with concatenated axes"""
@@ -502,6 +532,7 @@ def f(x, y):
     )
 
     await c.compute(z, optimize_graph=False)
+
     da.assert_eq(z, x)
 
 


Code review comments:
diff_hunk: @@ -130,8 +130,20 @@ def test_futures_to_delayed_array(c):
     assert_eq(A.compute(), np.concatenate([x, x], axis=0))
 
 
+# with pytest.warns(UserWarning, match=warning_message) as user_warnings_b:
+#     get_scheduler(scheduler="dask.distributed")
+#     get_scheduler(scheduler="distributed")
+#     get_scheduler()
+# assert len(user_warnings_b) == 0
path: dask/tests/test_distributed.py
text: @user1:
Was this intended to be added as a proper test?

@author:
I'm sorry, I forgot to remove those comments. I just pushed a new version after removing them.