diff --git a/dask/bag/core.py b/dask/bag/core.py
index aed4e6ca2..1cb43c01b 100644
--- a/dask/bag/core.py
+++ b/dask/bag/core.py
@@ -1987,5 +1987,5 @@ def split(seq, n):
 
 def to_dataframe(seq, columns, dtypes):
     import pandas as pd
-    res = pd.DataFrame(seq, columns=list(columns))
+    res = pd.DataFrame(reify(seq), columns=list(columns))
     return res.astype(dtypes, copy=False)
diff --git a/dask/dataframe/io/io.py b/dask/dataframe/io/io.py
index 93b47bb3c..5c82683aa 100644
--- a/dask/dataframe/io/io.py
+++ b/dask/dataframe/io/io.py
@@ -343,7 +343,7 @@ def dataframe_from_ctable(x, slc, columns=None, categories=None, lock=lock):
 
 
 def from_dask_array(x, columns=None):
-    """ Create Dask Array from a Dask DataFrame
+    """ Create a Dask DataFrame from a Dask Array.
 
     Converts a 2d array into a DataFrame and a 1d array into a Series.
 
diff --git a/dask/delayed.py b/dask/delayed.py
index a9cd531e0..8f0943d0d 100644
--- a/dask/delayed.py
+++ b/dask/delayed.py
@@ -64,26 +64,33 @@ def to_task_dask(expr):
     """
     if isinstance(expr, Delayed):
         return expr.key, expr.dask
+
     if isinstance(expr, base.Base):
         name = 'finalize-' + tokenize(expr, pure=True)
         keys = expr._keys()
         dsk = expr._optimize(ensure_dict(expr.dask), keys)
         dsk[name] = (expr._finalize, (concrete, keys))
         return name, dsk
-    if isinstance(expr, tuple) and type(expr) != tuple:
-        return expr, {}
-    if isinstance(expr, (Iterator, list, tuple, set)):
+
+    if isinstance(expr, Iterator):
+        expr = list(expr)
+    typ = type(expr)
+
+    if typ in (list, tuple, set):
         args, dasks = unzip((to_task_dask(e) for e in expr), 2)
         args = list(args)
         dsk = sharedict.merge(*dasks)
         # Ensure output type matches input type
-        if isinstance(expr, (tuple, set)):
-            return (type(expr), args), dsk
-        else:
-            return args, dsk
-    if type(expr) is dict:
+        return (args, dsk) if typ is list else ((typ, args), dsk)
+
+    if typ is dict:
         args, dsk = to_task_dask([[k, v] for k, v in expr.items()])
         return (dict, args), dsk
+
+    if typ is slice:
+        args, dsk = to_task_dask([expr.start, expr.stop, expr.step])
+        return (slice,) + tuple(args), dsk
+
     return expr, {}
 
 

Can't convert bag -> dataframe -> bag after flatten()
When I try to use `flatten()/concat()` on a bag and then proceed to convert to dataframe and back, the `pd.DataFrame` constructor fails because the data is passed in as an Iterator. See example below. Commenting out `.flatten()` or `.to_bag()` both cause it to be successful. But as written it causes an exception.

```
import dask.bag as db

(db
 .from_sequence(5 * [range(10)])
 .flatten()
 .to_dataframe()
 .to_bag()
 .compute()
)
```
```
TypeError                                 Traceback (most recent call last)
<ipython-input-1-31fb669b888e> in <module>()
      2 
      3 (db
----> 4  .from_sequence(5 * [range(10)])
      5  .flatten()
      6  .to_dataframe()

/Users/lsglick/.virtualenvs/test/lib/python2.7/site-packages/dask/base.pyc in compute(self, **kwargs)
     95             Extra keywords to forward to the scheduler ``get`` function.
     96         """
---> 97         (result,) = compute(self, traverse=False, **kwargs)
     98         return result
     99 

/Users/lsglick/.virtualenvs/test/lib/python2.7/site-packages/dask/base.pyc in compute(*args, **kwargs)
    202     dsk = collections_to_dsk(variables, optimize_graph, **kwargs)
    203     keys = [var._keys() for var in variables]
--> 204     results = get(dsk, keys, **kwargs)
    205 
    206     results_iter = iter(results)

/Users/lsglick/.virtualenvs/test/lib/python2.7/site-packages/dask/multiprocessing.pyc in get(dsk, keys, num_workers, func_loads, func_dumps, optimize_graph, **kwargs)
    175                            get_id=_process_get_id, dumps=dumps, loads=loads,
    176                            pack_exception=pack_exception,
--> 177                            raise_exception=reraise, **kwargs)
    178     finally:
    179         if cleanup:

/Users/lsglick/.virtualenvs/test/lib/python2.7/site-packages/dask/local.pyc in get_async(apply_async, num_workers, dsk, result, cache, get_id, rerun_exceptions_locally, pack_exception, raise_exception, callbacks, dumps, loads, **kwargs)
    519                         _execute_task(task, data)  # Re-execute locally
    520                     else:
--> 521                         raise_exception(exc, tb)
    522                 res, worker_id = loads(res_info)
    523                 state['cache'][key] = res

/Users/lsglick/.virtualenvs/test/lib/python2.7/site-packages/dask/local.pyc in execute_task()
    288     try:
    289         task, data = loads(task_info)
--> 290         result = _execute_task(task, data)
    291         id = get_id()
    292         result = dumps((result, id))

/Users/lsglick/.virtualenvs/test/lib/python2.7/site-packages/dask/local.pyc in _execute_task()
    268     elif istask(arg):
    269         func, args = arg[0], arg[1:]
--> 270         args2 = [_execute_task(a, cache) for a in args]
    271         return func(*args2)
    272     elif not ishashable(arg):

/Users/lsglick/.virtualenvs/test/lib/python2.7/site-packages/dask/local.pyc in _execute_task()
    269         func, args = arg[0], arg[1:]
    270         args2 = [_execute_task(a, cache) for a in args]
--> 271         return func(*args2)
    272     elif not ishashable(arg):
    273         return arg

/Users/lsglick/.virtualenvs/test/lib/python2.7/site-packages/dask/bag/core.pyc in to_dataframe()
   1988 def to_dataframe(seq, columns, dtypes):
   1989     import pandas as pd
-> 1990     res = pd.DataFrame(seq, columns=list(columns))
   1991     return res.astype(dtypes, copy=False)

/Users/lsglick/.virtualenvs/test/lib/python2.7/site-packages/pandas/core/frame.pyc in __init__()
    323                 mgr = self._init_dict({}, index, columns, dtype=dtype)
    324         elif isinstance(data, collections.Iterator):
--> 325             raise TypeError("data argument can't be an iterator")
    326         else:
    327             try:

TypeError: data argument can't be an iterator
```

Versions:
Python: 3.5.2
Pandas: 0.20.2
Dask: 0.15.0

AND

Python: 2.7.10
Pandas: 0.19.2
Dask: 0.15.0
diff --git a/dask/bag/tests/test_bag.py b/dask/bag/tests/test_bag.py
index bd3679cbe..0053dc0b6 100644
--- a/dask/bag/tests/test_bag.py
+++ b/dask/bag/tests/test_bag.py
@@ -822,6 +822,13 @@ def test_to_dataframe():
     dd.utils.assert_eq(df, sol, check_index=False)
     check_parts(df, sol)
 
+    # Works with iterators
+    b = db.from_sequence(range(100), npartitions=5).map_partitions(iter)
+    sol = pd.DataFrame({'a': range(100)})
+    df = b.to_dataframe(columns=sol)
+    dd.utils.assert_eq(df, sol, check_index=False)
+    check_parts(df, sol)
+
 
 ext_open = [('gz', GzipFile), ('', open)]
 if not PY2:
diff --git a/dask/tests/test_delayed.py b/dask/tests/test_delayed.py
index 837e739a4..8c89cca23 100644
--- a/dask/tests/test_delayed.py
+++ b/dask/tests/test_delayed.py
@@ -1,4 +1,4 @@
-from collections import Iterator, namedtuple
+from collections import namedtuple
 from operator import add, setitem
 import pickle
 from random import random
@@ -32,6 +32,10 @@ def test_to_task_dask():
     assert task == x
     assert dict(dask) == {}
 
+    task, dask = to_task_dask(slice(a, b, 3))
+    assert task == (slice, 'a', 'b', 3)
+    assert dict(dask) == merge(a.dask, b.dask)
+
     # Issue https://github.com/dask/dask/issues/2107
     class MyClass(dict):
         pass
@@ -58,6 +62,8 @@ def test_operators():
     a = delayed([1, 2, 3])
     assert a[0].compute() == 1
     assert (a + a).compute() == [1, 2, 3, 1, 2, 3]
+    b = delayed(2)
+    assert a[:b].compute() == [1, 2]
 
     a = delayed(10)
     assert (a + 1).compute() == 11
@@ -154,7 +160,6 @@ def test_lists_are_concrete():
     assert c.compute() == 20
 
 
-@pytest.mark.xfail
 def test_iterators():
     a = delayed(1)
     b = delayed(2)
@@ -163,7 +168,6 @@ def test_iterators():
     assert c.compute() == 3
 
     def f(seq):
-        assert isinstance(seq, Iterator)
         return sum(seq)
 
     c = delayed(f)(iter([a, b]))

pip install -e .[complete]
parse_log_pytest
pytest
pytest moto mock
3.6
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
aiobotocore==2.1.2
aiohttp==3.8.6
aioitertools==0.11.0
aiosignal==1.2.0
async-timeout==4.0.2
asynctest==0.13.0
attrs @ file:///opt/conda/conda-bld/attrs_1642510447205/work
boto3==1.23.10
botocore==1.23.24
certifi==2021.5.30
cffi==1.15.1
charset-normalizer==2.0.12
click==8.0.4
cloudpickle==2.2.1
cryptography==40.0.2
-e git+https://github.com/dask/dask.git@ecdae84aa3b8ee5408876b0b7d7eade3d127e177#egg=dask
dataclasses==0.8
distributed==1.19.3
frozenlist==1.2.0
fsspec==2022.1.0
HeapDict==1.0.1
idna==3.10
idna-ssl==1.1.0
importlib-metadata @ file:///tmp/build/80754af9/importlib-metadata_1631916693255/work
iniconfig @ file:///home/linux1/recipes/ci/iniconfig_1610983019677/work
Jinja2==3.0.3
jmespath==0.10.0
locket==1.0.0
MarkupSafe==2.0.1
mock==5.2.0
more-itertools @ file:///tmp/build/80754af9/more-itertools_1637733554872/work
moto==4.0.13
msgpack-python==0.5.6
multidict==5.2.0
numpy==1.19.5
packaging @ file:///tmp/build/80754af9/packaging_1637314298585/work
pandas==1.1.5
partd==1.2.0
pluggy @ file:///tmp/build/80754af9/pluggy_1615976315926/work
psutil==7.0.0
py @ file:///opt/conda/conda-bld/py_1644396412707/work
pycparser==2.21
pyparsing @ file:///tmp/build/80754af9/pyparsing_1635766073266/work
pytest==6.2.4
python-dateutil==2.9.0.post0
pytz==2025.2
requests==2.27.1
responses==0.17.0
s3fs==2022.1.0
s3transfer==0.5.2
six==1.17.0
sortedcontainers==2.4.0
tblib==1.7.0
toml @ file:///tmp/build/80754af9/toml_1616166611790/work
toolz==0.12.0
tornado==6.1
typing_extensions @ file:///opt/conda/conda-bld/typing_extensions_1647553014482/work
urllib3==1.26.20
Werkzeug==2.0.3
wrapt==1.16.0
xmltodict==0.14.2
yarl==1.7.2
zict==2.1.0
zipp @ file:///tmp/build/80754af9/zipp_1633618647012/work

name: dask
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - attrs=21.4.0=pyhd3eb1b0_0
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2021.5.30=py36h06a4308_0
  - importlib-metadata=4.8.1=py36h06a4308_0
  - importlib_metadata=4.8.1=hd3eb1b0_0
  - iniconfig=1.1.1=pyhd3eb1b0_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.3=he6710b0_2
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - more-itertools=8.12.0=pyhd3eb1b0_0
  - ncurses=6.4=h6a678d5_0
  - openssl=1.1.1w=h7f8727e_0
  - packaging=21.3=pyhd3eb1b0_0
  - pip=21.2.2=py36h06a4308_0
  - pluggy=0.13.1=py36h06a4308_0
  - py=1.11.0=pyhd3eb1b0_0
  - pyparsing=3.0.4=pyhd3eb1b0_0
  - pytest=6.2.4=py36h06a4308_2
  - python=3.6.13=h12debd9_1
  - readline=8.2=h5eee18b_0
  - setuptools=58.0.4=py36h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - toml=0.10.2=pyhd3eb1b0_0
  - typing_extensions=4.1.1=pyh06a4308_0
  - wheel=0.37.1=pyhd3eb1b0_0
  - xz=5.6.4=h5eee18b_1
  - zipp=3.6.0=pyhd3eb1b0_0
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - aiobotocore==2.1.2
      - aiohttp==3.8.6
      - aioitertools==0.11.0
      - aiosignal==1.2.0
      - async-timeout==4.0.2
      - asynctest==0.13.0
      - boto3==1.23.10
      - botocore==1.23.24
      - cffi==1.15.1
      - charset-normalizer==2.0.12
      - click==8.0.4
      - cloudpickle==2.2.1
      - cryptography==40.0.2
      - dataclasses==0.8
      - distributed==1.19.3
      - frozenlist==1.2.0
      - fsspec==2022.1.0
      - heapdict==1.0.1
      - idna==3.10
      - idna-ssl==1.1.0
      - jinja2==3.0.3
      - jmespath==0.10.0
      - locket==1.0.0
      - markupsafe==2.0.1
      - mock==5.2.0
      - moto==4.0.13
      - msgpack-python==0.5.6
      - multidict==5.2.0
      - numpy==1.19.5
      - pandas==1.1.5
      - partd==1.2.0
      - psutil==7.0.0
      - pycparser==2.21
      - python-dateutil==2.9.0.post0
      - pytz==2025.2
      - requests==2.27.1
      - responses==0.17.0
      - s3fs==2022.1.0
      - s3transfer==0.5.2
      - six==1.17.0
      - sortedcontainers==2.4.0
      - tblib==1.7.0
      - toolz==0.12.0
      - tornado==6.1
      - urllib3==1.26.20
      - werkzeug==2.0.3
      - wrapt==1.16.0
      - xmltodict==0.14.2
      - yarl==1.7.2
      - zict==2.1.0
prefix: /opt/conda/envs/dask

dask/tests/test_delayed.py::test_to_task_dask dask/tests/test_delayed.py::test_operators
dask/bag/tests/test_bag.py::test_to_dataframe
dask/bag/tests/test_bag.py::test_Bag dask/bag/tests/test_bag.py::test_keys dask/bag/tests/test_bag.py::test_map dask/bag/tests/test_bag.py::test_map_function_with_multiple_arguments dask/bag/tests/test_bag.py::test_map_with_constructors dask/bag/tests/test_bag.py::test_map_with_builtins dask/bag/tests/test_bag.py::test_map_with_kwargs dask/bag/tests/test_bag.py::test_bag_map dask/bag/tests/test_bag.py::test_map_method dask/bag/tests/test_bag.py::test_starmap dask/bag/tests/test_bag.py::test_filter dask/bag/tests/test_bag.py::test_remove dask/bag/tests/test_bag.py::test_iter dask/bag/tests/test_bag.py::test_repr[str] dask/bag/tests/test_bag.py::test_repr[repr] dask/bag/tests/test_bag.py::test_pluck dask/bag/tests/test_bag.py::test_pluck_with_default dask/bag/tests/test_bag.py::test_unzip dask/bag/tests/test_bag.py::test_fold dask/bag/tests/test_bag.py::test_distinct dask/bag/tests/test_bag.py::test_frequencies dask/bag/tests/test_bag.py::test_topk dask/bag/tests/test_bag.py::test_topk_with_non_callable_key[1] dask/bag/tests/test_bag.py::test_topk_with_non_callable_key[2] dask/bag/tests/test_bag.py::test_topk_with_multiarg_lambda dask/bag/tests/test_bag.py::test_lambdas dask/bag/tests/test_bag.py::test_reductions dask/bag/tests/test_bag.py::test_reduction_names dask/bag/tests/test_bag.py::test_tree_reductions dask/bag/tests/test_bag.py::test_aggregation[1] dask/bag/tests/test_bag.py::test_aggregation[3] dask/bag/tests/test_bag.py::test_aggregation[4] dask/bag/tests/test_bag.py::test_non_splittable_reductions[1] dask/bag/tests/test_bag.py::test_non_splittable_reductions[10] dask/bag/tests/test_bag.py::test_std dask/bag/tests/test_bag.py::test_var dask/bag/tests/test_bag.py::test_join dask/bag/tests/test_bag.py::test_foldby dask/bag/tests/test_bag.py::test_map_partitions dask/bag/tests/test_bag.py::test_map_partitions_args_kwargs dask/bag/tests/test_bag.py::test_random_sample_size dask/bag/tests/test_bag.py::test_random_sample_prob_range dask/bag/tests/test_bag.py::test_random_sample_repeated_computation dask/bag/tests/test_bag.py::test_random_sample_different_definitions dask/bag/tests/test_bag.py::test_random_sample_random_state dask/bag/tests/test_bag.py::test_lazify_task dask/bag/tests/test_bag.py::test_lazify dask/bag/tests/test_bag.py::test_inline_singleton_lists dask/bag/tests/test_bag.py::test_take dask/bag/tests/test_bag.py::test_take_npartitions dask/bag/tests/test_bag.py::test_take_npartitions_warn dask/bag/tests/test_bag.py::test_map_is_lazy dask/bag/tests/test_bag.py::test_can_use_dict_to_make_concrete dask/bag/tests/test_bag.py::test_read_text dask/bag/tests/test_bag.py::test_read_text_large dask/bag/tests/test_bag.py::test_read_text_encoding dask/bag/tests/test_bag.py::test_read_text_large_gzip dask/bag/tests/test_bag.py::test_from_sequence dask/bag/tests/test_bag.py::test_from_long_sequence dask/bag/tests/test_bag.py::test_product dask/bag/tests/test_bag.py::test_partition_collect dask/bag/tests/test_bag.py::test_groupby dask/bag/tests/test_bag.py::test_groupby_with_indexer dask/bag/tests/test_bag.py::test_groupby_with_npartitions_changed dask/bag/tests/test_bag.py::test_concat dask/bag/tests/test_bag.py::test_flatten dask/bag/tests/test_bag.py::test_concat_after_map dask/bag/tests/test_bag.py::test_args dask/bag/tests/test_bag.py::test_to_textfiles[gz-GzipFile] dask/bag/tests/test_bag.py::test_to_textfiles[-open] dask/bag/tests/test_bag.py::test_to_textfiles[bz2-BZ2File] dask/bag/tests/test_bag.py::test_to_textfiles_name_function_preserves_order dask/bag/tests/test_bag.py::test_to_textfiles_name_function_warn dask/bag/tests/test_bag.py::test_to_textfiles_encoding dask/bag/tests/test_bag.py::test_to_textfiles_inputs dask/bag/tests/test_bag.py::test_to_textfiles_endlines dask/bag/tests/test_bag.py::test_string_namespace dask/bag/tests/test_bag.py::test_string_namespace_with_unicode dask/bag/tests/test_bag.py::test_str_empty_split dask/bag/tests/test_bag.py::test_map_with_iterator_function dask/bag/tests/test_bag.py::test_ensure_compute_output_is_concrete dask/bag/tests/test_bag.py::test_bag_class_extend dask/bag/tests/test_bag.py::test_gh715 dask/bag/tests/test_bag.py::test_bag_compute_forward_kwargs dask/bag/tests/test_bag.py::test_to_delayed dask/bag/tests/test_bag.py::test_to_delayed_optimizes dask/bag/tests/test_bag.py::test_from_delayed dask/bag/tests/test_bag.py::test_from_delayed_iterator dask/bag/tests/test_bag.py::test_range dask/bag/tests/test_bag.py::test_zip[1] dask/bag/tests/test_bag.py::test_zip[7] dask/bag/tests/test_bag.py::test_zip[10] dask/bag/tests/test_bag.py::test_zip[28] dask/bag/tests/test_bag.py::test_repartition[1-1] dask/bag/tests/test_bag.py::test_repartition[1-2] dask/bag/tests/test_bag.py::test_repartition[1-7] dask/bag/tests/test_bag.py::test_repartition[1-11] dask/bag/tests/test_bag.py::test_repartition[1-23] dask/bag/tests/test_bag.py::test_repartition[2-1] dask/bag/tests/test_bag.py::test_repartition[2-2] dask/bag/tests/test_bag.py::test_repartition[2-7] dask/bag/tests/test_bag.py::test_repartition[2-11] dask/bag/tests/test_bag.py::test_repartition[2-23] dask/bag/tests/test_bag.py::test_repartition[5-1] dask/bag/tests/test_bag.py::test_repartition[5-2] dask/bag/tests/test_bag.py::test_repartition[5-7] dask/bag/tests/test_bag.py::test_repartition[5-11] dask/bag/tests/test_bag.py::test_repartition[5-23] dask/bag/tests/test_bag.py::test_repartition[12-1] dask/bag/tests/test_bag.py::test_repartition[12-2] dask/bag/tests/test_bag.py::test_repartition[12-7] dask/bag/tests/test_bag.py::test_repartition[12-11] dask/bag/tests/test_bag.py::test_repartition[12-23] dask/bag/tests/test_bag.py::test_repartition[23-1] dask/bag/tests/test_bag.py::test_repartition[23-2] dask/bag/tests/test_bag.py::test_repartition[23-7] dask/bag/tests/test_bag.py::test_repartition[23-11] dask/bag/tests/test_bag.py::test_repartition[23-23] dask/bag/tests/test_bag.py::test_repartition_names dask/bag/tests/test_bag.py::test_accumulate dask/bag/tests/test_bag.py::test_groupby_tasks dask/bag/tests/test_bag.py::test_groupby_tasks_names dask/bag/tests/test_bag.py::test_groupby_tasks_2[1000-20-100] dask/bag/tests/test_bag.py::test_groupby_tasks_2[12345-234-1042] dask/bag/tests/test_bag.py::test_groupby_tasks_3 dask/bag/tests/test_bag.py::test_to_textfiles_empty_partitions dask/bag/tests/test_bag.py::test_reduction_empty dask/bag/tests/test_bag.py::test_reduction_empty_aggregate[1] dask/bag/tests/test_bag.py::test_reduction_empty_aggregate[2] dask/bag/tests/test_bag.py::test_reduction_empty_aggregate[4] dask/bag/tests/test_bag.py::test_reduction_with_non_comparable_objects dask/bag/tests/test_bag.py::test_empty dask/bag/tests/test_bag.py::test_bag_picklable dask/bag/tests/test_bag.py::test_msgpack_unicode dask/bag/tests/test_bag.py::test_bag_with_single_callable dask/bag/tests/test_bag.py::test_optimize_fuse_keys dask/bag/tests/test_bag.py::test_reductions_are_lazy dask/bag/tests/test_bag.py::test_repeated_groupby dask/bag/tests/test_bag.py::test_temporary_directory dask/bag/tests/test_bag.py::test_empty_bag dask/tests/test_delayed.py::test_delayed dask/tests/test_delayed.py::test_methods dask/tests/test_delayed.py::test_attributes dask/tests/test_delayed.py::test_method_getattr_optimize dask/tests/test_delayed.py::test_delayed_errors dask/tests/test_delayed.py::test_common_subexpressions dask/tests/test_delayed.py::test_lists dask/tests/test_delayed.py::test_literates dask/tests/test_delayed.py::test_literates_keys dask/tests/test_delayed.py::test_lists_are_concrete dask/tests/test_delayed.py::test_iterators dask/tests/test_delayed.py::test_traverse_false dask/tests/test_delayed.py::test_pure dask/tests/test_delayed.py::test_pure_global_setting dask/tests/test_delayed.py::test_nout dask/tests/test_delayed.py::test_kwargs dask/tests/test_delayed.py::test_array_delayed dask/tests/test_delayed.py::test_array_bag_delayed dask/tests/test_delayed.py::test_delayed_picklable dask/tests/test_delayed.py::test_delayed_compute_forward_kwargs dask/tests/test_delayed.py::test_delayed_method_descriptor dask/tests/test_delayed.py::test_delayed_callable dask/tests/test_delayed.py::test_delayed_name_on_call dask/tests/test_delayed.py::test_callable_obj dask/tests/test_delayed.py::test_name_consistent_across_instances dask/tests/test_delayed.py::test_sensitive_to_partials dask/tests/test_delayed.py::test_delayed_name dask/tests/test_delayed.py::test_finalize_name
dask/delayed.py dask/bag/core.py dask/dataframe/io/io.py
dask/delayed.py dask/bag/core.py dask/dataframe/io/io.py