Repository: dask/dask
Pull request: Evaluate graph lazily if meta is provided in from_delayed
The `from_delayed` no longer is lazy and always computes the first element. This is unfortunate, particularly since the tasks typically involve remote connections. 

This is a regression which was introduced in https://github.com/dask/dask/pull/7586

Apparently this was somehow purposefully done, see https://github.com/dask/dask/pull/7586#discussion_r637579242 but I don't understand the reasoning. @galipremsagar can you please shed a bit more light onto this?

Disclaimer: I haven't run the full test suite, yet, in case this lights up like a xmas tree. Anyhow, I believe we should try to preserve the lazy evaluation if meta is provided

- [x] Tests added / passed
- [x] Passes `black dask` / `flake8 dask` / `isort dask`

xref https://github.com/JDASoftwareGroup/kartothek/issues/475


Problem statement:
from_delayed can trigger large computation
**What happened**:

Creation of a dask dataframe from delayed objects now triggers computation.

**What you expected to happen**:

When a using a `dd.from_delayed` with the `meta` keyword, no computation should be done.

**Minimal Complete Verifiable Example**:

The following code triggers the function with the sleep statement:

```python
import dask
import dask.dataframe as dd
import numpy as np
import pandas as pd

def gimme_a_hull():
    import time
    print("I'm so sleepy")
    time.sleep(1000)
    return pd.DataFrame({"segment_id": np.array([1,2,3])})

results = dask.delayed(gimme_a_hull)()
meta = dd.utils.make_meta([("segment_id", np.int64)])
dd.from_delayed(results, meta=meta)
```

**Anything else we need to know?**:

**Environment**:

- Dask version: 2021.6.0
- Python version: 3.8.5
- Operating System: Ubuntu 20.04.01
- Install method (conda, pip, source): pip

Hints from issue discussion:
I think [this line](https://github.com/dask/dask/blob/d9df00058f3b52b6c6d26ed836fbed9a7e3b4232/dask/dataframe/io/io.py#L591) is the culprit.

I'm not familiar with what `parent_meta` is representing here.  @galipremsagar  , what is the intention here, and is it possible to avoid triggering a computation?

Commit message: Evaluate graph lazily if meta is provided in from_delayed

Patch to review:
diff --git a/dask/dataframe/io/io.py b/dask/dataframe/io/io.py
index 266f375f3d3..8ec331fcaf5 100644
--- a/dask/dataframe/io/io.py
+++ b/dask/dataframe/io/io.py
@@ -584,16 +584,18 @@ def from_delayed(
         delayed(df) if not isinstance(df, Delayed) and hasattr(df, "key") else df
         for df in dfs
     ]
+
     for df in dfs:
         if not isinstance(df, Delayed):
             raise TypeError("Expected Delayed object, got %s" % type(df).__name__)
 
-    parent_meta = delayed(make_meta)(dfs[0]).compute()
-
     if meta is None:
-        meta = parent_meta
+        meta = delayed(make_meta)(dfs[0]).compute()
     else:
-        meta = make_meta(meta, parent_meta=parent_meta)
+        meta = make_meta(meta)
+
+    if not dfs:
+        dfs = [delayed(make_meta)(meta)]
 
     name = prefix + "-" + tokenize(*dfs)
     dsk = merge(df.dask for df in dfs)
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index 86f1aa1cd12..844d762e768 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -2189,6 +2189,27 @@ def test_fillna():
     assert_eq(df.fillna(method="pad", limit=3), ddf.fillna(method="pad", limit=3))
 
 
+def test_from_delayed_lazy_if_meta_provided():
+    """Ensure that the graph is 100% lazily evaluted if meta is provided"""
+
+    @dask.delayed
+    def raise_exception():
+        raise RuntimeError()
+
+    tasks = [raise_exception()]
+    ddf = dd.from_delayed(tasks, meta=dict(a=float))
+
+    with pytest.raises(RuntimeError):
+        ddf.compute()
+
+
+def test_from_delayed_empty_meta_provided():
+    ddf = dd.from_delayed([], meta=dict(a=float))
+    df = ddf.compute()
+    expected = pd.DataFrame({"a": [0.1]}).iloc[:0]
+    assert_eq(df, expected)
+
+
 def test_fillna_duplicate_index():
     @dask.delayed
     def f():


Code review comments:
diff_hunk: @@ -2189,6 +2189,27 @@ def test_fillna():
     assert_eq(df.fillna(method="pad", limit=3), ddf.fillna(method="pad", limit=3))
 
 
+def test_from_delayed_lazy_if_meta_provided():
+    """Ensure that the graph is 100% lazily evaluted if meta is provided"""
+
+    @dask.delayed
+    def raise_exception():
+        raise RuntimeError()
+
+    tasks = [raise_exception()]
+    ddf = dd.from_delayed(tasks, meta=dict(a=float))
+
+    with pytest.raises(RuntimeError):
+        ddf.compute()
+
+
+def test_from_delayed_empty_meta_provided():
+    ddf = dd.from_delayed([], meta=dict(a=float))
+    df = ddf.compute()
+    expected = pd.DataFrame({"a": [0.1]}).iloc[:0]
+    assert_eq(df, expected)
path: dask/dataframe/tests/test_dataframe.py
text: @user1:
It's better not to let `assert_eq` call `compute` internally

```suggestion
    expected = pd.DataFrame({"a": [0.1]}).iloc[:0]
    assert_eq(ddf, expected)
```

@author:
But you are proposing to let it be called internally, don't you? Did you mean "**It is better** to let assert_eq call compute"?

@user1:
Yes :facepalm: "**It is better** to let assert_eq call compute" is what I mean.