Repository: dask/dask

Issue:
Displaying Array/DataFrame in notebook materializes all graph layers
Displaying the `_repr_html_` of an Array or DataFrame includes the number of tasks. Figuring out the number of tasks causes all layers in the HighLevelGraph to be materialized, which makes displaying the repr very slow for large graphs, plus somewhat undoes all the good work of getting more things into Blockwise.

`_repr_html_` gets the task count from `len(self.dask)`. `HighLevelGraph.__len__`, in turn, just returns `len(self.to_dict())`

**Minimal Complete Verifiable Example**:

```python
import dask.array as da
def layers_are_materialized(obj):
    return {k: lyr.is_materialized() for k, lyr in obj.__dask_graph__().layers.items()}

x = da.ones((10000, 10000), chunks=10)

# add some blockwise layers
for i in range(4):
    x = x + 1

layers_are_materialized(x)
# {'ones-e532400d218fbe77c0befc8551fa738a': True,
#  'add-90a3aa343fa08babeff712d337a8f287': False,
#  'add-7c52aeffa2899a67494f584b1b6ce24c': False,
#  'add-ae7ff0a503e9c7464de58300616c3649': False,
#  'add-96feedc695ac0a342339674fb0b1067f': False}

%time _ = x._repr_html_()
CPU times: user 18.8 s, sys: 469 ms, total: 19.3 s
Wall time: 19.3 s

layers_are_materialized(x)
# {'ones-e532400d218fbe77c0befc8551fa738a': True,
#  'add-90a3aa343fa08babeff712d337a8f287': True,
#  'add-7c52aeffa2899a67494f584b1b6ce24c': True,
#  'add-ae7ff0a503e9c7464de58300616c3649': True,
#  'add-96feedc695ac0a342339674fb0b1067f': True}
```

<details><summary>(Just to confirm I profiled, and it is indeed <code>HighLevelGraph.__len__</code> that takes all the time)</summary>

![Screen Shot 2021-02-24 at 6 09 49 PM](https://user-images.githubusercontent.com/3309802/109087870-84008f80-76cb-11eb-94dd-a73869158e76.png)

</details>

**Anything else we need to know?**:

I wasn't sure where to address this—change `__len__` on HighLevelGraph to not materialize the full dict but give an approximation, or have `_repr_html_` use different logic (since it would be okay for display purposes to have the task count be less accurate)?

These are the alternatives I'm imagining:

```python
# sum the lengths of the layers
%%timeit
sum(len(layer) for layer in x.dask.layers.values())
# 11.5 µs ± 244 ns per loop (mean ± std. dev. of 7 runs, 100000 loops each)

# use get_all_external_keys
%%timeit
len(x.dask.get_all_external_keys())
delattr(x.dask, "_all_external_keys")
# 2.81 s ± 42.1 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)
```

The first would over-count duplicate keys between layers; the second would under-count intermediate keys. So probably neither is appropriate for `__len__`, but which is more likely to be accurate for `_repr_html_`?

cc @crusaderky since you worked on this most recently.

**Environment**:

- Dask version: master (895c9541044e16809151f0e3db95c48ea338c6d3)
- Python version: 3.8.7

Discussion:

Patch:
diff --git a/.github/workflows/ci-additional.yml b/.github/workflows/ci-additional.yml
index 0e067dd8e..15ba4d835 100644
--- a/.github/workflows/ci-additional.yml
+++ b/.github/workflows/ci-additional.yml
@@ -79,7 +79,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.6", "3.7", "3.8"]
+        python-version: ["3.7", "3.8", "3.9"]
     steps:
       - name: Checkout source
         uses: actions/checkout@v2
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 915b50261..8616cc0b8 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -9,7 +9,7 @@ jobs:
       fail-fast: false
       matrix:
         os: ["windows-latest", "ubuntu-latest", "macos-latest"]
-        python-version: ["3.6", "3.7", "3.8", "3.9"]
+        python-version: ["3.7", "3.8", "3.9"]
 
     env:
       PYTHON_VERSION: ${{ matrix.python-version }}
diff --git a/continuous_integration/environment-3.6.yaml b/continuous_integration/environment-3.6.yaml
deleted file mode 100644
index 927d95267..000000000
--- a/continuous_integration/environment-3.6.yaml
+++ /dev/null
@@ -1,63 +0,0 @@
-name: test-environment
-channels:
-  - conda-forge
-dependencies:
-  # required dependencies
-  - python=3.6
-  - numpy=1.15
-  - pandas=0.25
-  # test dependencies
-  - pytest
-  - pytest-cov
-  - pytest-rerunfailures
-  - pytest-xdist
-  - moto
-  - flask
-  - coverage
-  - jsonschema
-  # Issue with conda package for NumPy 1.15?
-  # See fastparquet-444
-  # - fastparquet
-  - h5py
-  - pytables
-  - fsspec
-  - zarr
-  - tiledb-py
-  - xarray
-  - sqlalchemy
-  - pyarrow=0.14
-  # other -- IO
-  - bcolz
-  - blosc
-  - s3fs=0.4
-  - boto3
-  - botocore
-  - aiohttp
-  - bokeh
-  - httpretty
-  - chest
-  - cloudpickle
-  - crick
-  - cytoolz
-  - distributed
-  - graphviz
-  - ipython
-  - lz4
-  - numba
-  - partd
-  - psutil
-  - requests
-  - scikit-image
-  - scikit-learn
-  - scipy
-  - toolz
-  - python-snappy
-  - sparse
-  - cachey
-  - python-graphviz
-  - pandas-datareader
-  - python-xxhash
-  - mmh3
-  - pip
-  - pip:
-    - git+https://github.com/dask/distributed
diff --git a/continuous_integration/environment-3.7.yaml b/continuous_integration/environment-3.7.yaml
index e2bc9112d..ded4a5083 100644
--- a/continuous_integration/environment-3.7.yaml
+++ b/continuous_integration/environment-3.7.yaml
@@ -5,8 +5,8 @@ channels:
 dependencies:
   # required dependencies
   - python=3.7
-  - numpy=1.17
-  - pandas=1.1
+  - numpy=1.15
+  - pandas=0.25
   # test dependencies
   - pytest
   - pytest-cov
@@ -14,7 +14,9 @@ dependencies:
   - pytest-xdist
   - moto
   - flask
-  - fastparquet
+  # Issue with conda package for NumPy 1.15
+  # See fastparquet-444
+  # - fastparquet
   - h5py
   - pytables
   - zarr
@@ -22,7 +24,7 @@ dependencies:
   - xarray
   - fsspec
   - sqlalchemy
-  - pyarrow=1.0
+  - pyarrow=0.14
   - coverage
   - jsonschema
   # other -- IO
diff --git a/continuous_integration/environment-3.8.yaml b/continuous_integration/environment-3.8.yaml
index a654fd7ff..e92aca546 100644
--- a/continuous_integration/environment-3.8.yaml
+++ b/continuous_integration/environment-3.8.yaml
@@ -5,8 +5,8 @@ channels:
 dependencies:
   # required dependencies
   - python=3.8
-  - numpy
-  - pandas
+  - numpy=1.17
+  - pandas=1.1
   # test dependencies
   - pytest
   - pytest-cov
@@ -22,7 +22,7 @@ dependencies:
   - xarray
   - fsspec
   - sqlalchemy
-  - pyarrow
+  - pyarrow=1.0
   - coverage
   - jsonschema
   # other -- IO
diff --git a/dask/base.py b/dask/base.py
index 52c1e5fcc..1d5ad89b8 100644
--- a/dask/base.py
+++ b/dask/base.py
@@ -1,5 +1,6 @@
 from collections import OrderedDict
 from contextlib import contextmanager
+from dataclasses import is_dataclass, fields
 from functools import partial
 from hashlib import md5
 from numbers import Number
@@ -14,7 +15,6 @@ import uuid
 from tlz import merge, groupby, curry, identity
 from tlz.functoolz import Compose
 
-from .compatibility import is_dataclass, dataclass_fields
 from .context import thread_state
 from .core import flatten, quote, get as simple_get, literal
 from .hashing import hash_buffer_hex
@@ -428,7 +428,7 @@ def unpack_collections(*args, **kwargs):
                         dict,
                         [
                             [f.name, _unpack(getattr(expr, f.name))]
-                            for f in dataclass_fields(expr)
+                            for f in fields(expr)
                         ],
                     ),
                 )
diff --git a/dask/blockwise.py b/dask/blockwise.py
index b2ce9e1a0..440b3a455 100644
--- a/dask/blockwise.py
+++ b/dask/blockwise.py
@@ -285,20 +285,13 @@ class Blockwise(Layer):
     def __iter__(self):
         return iter(self._dict)
 
-    def __len__(self):
-        return int(prod(self._out_numblocks().values()))
-
-    def _out_numblocks(self):
-        d = {}
-        out_d = {}
-        indices = {k: v for k, v in self.indices if v is not None}
-        for k, v in self.numblocks.items():
-            for a, b in zip(indices[k], v):
-                d[a] = max(d.get(a, 0), b)
-                if a in self.output_indices:
-                    out_d[a] = d[a]
-
-        return out_d
+    def __len__(self) -> int:
+        # same method as `get_output_keys`, without manifesting the keys themselves
+        return (
+            len(self.output_blocks)
+            if self.output_blocks
+            else prod(self.dims[i] for i in self.output_indices)
+        )
 
     def is_materialized(self):
         return hasattr(self, "_cached_dict")
diff --git a/dask/compatibility.py b/dask/compatibility.py
index f4e0a304a..756f00fe1 100644
--- a/dask/compatibility.py
+++ b/dask/compatibility.py
@@ -7,17 +7,6 @@ from .utils import apply  # noqa
 # TODO: remove this once dask requires distributed >= 2.2.0
 unicode = str  # noqa
 
-try:
-    from dataclasses import is_dataclass, fields as dataclass_fields
-except ImportError:
-    # Python < 3.7
-    def is_dataclass(x):
-        return False
-
-    def dataclass_fields(x):
-        return []
-
-
 try:
     from math import prod
 except ImportError:
diff --git a/dask/dataframe/io/parquet/arrow.py b/dask/dataframe/io/parquet/arrow.py
index 2abec49f6..d8d5aa4c3 100644
--- a/dask/dataframe/io/parquet/arrow.py
+++ b/dask/dataframe/io/parquet/arrow.py
@@ -1778,6 +1778,7 @@ class ArrowLegacyEngine(ArrowDatasetEngine):
                     root_dir=base,
                     engine=cls,
                     out_dir=False,
+                    fs=fs,
                 )
                 if schema is None:
                     schema = metadata.schema.to_arrow_schema()
diff --git a/dask/dataframe/io/parquet/core.py b/dask/dataframe/io/parquet/core.py
index a9194e295..1274d4d0d 100644
--- a/dask/dataframe/io/parquet/core.py
+++ b/dask/dataframe/io/parquet/core.py
@@ -681,6 +681,7 @@ def create_metadata_file(
     split_every=32,
     compute=True,
     compute_kwargs=None,
+    fs=None,
 ):
     """Construct a global _metadata file from a list of parquet files.
 
@@ -722,6 +723,12 @@ def create_metadata_file(
         then a ``dask.delayed`` object is returned for future computation.
     compute_kwargs : dict, optional
         Options to be passed in to the compute method
+    fs : fsspec object, optional
+        File-system instance to use for file handling. If prefixes have
+        been removed from the elements of ``paths`` before calling this
+        function, an ``fs`` argument must be provided to ensure correct
+        behavior on remote file systems ("naked" paths cannot be used
+        to infer file-system information).
     """
 
     # Get engine.
@@ -735,7 +742,12 @@ def create_metadata_file(
         engine = get_engine(engine)
 
     # Process input path list
-    fs, _, paths = get_fs_token_paths(paths, mode="rb", storage_options=storage_options)
+    if fs is None:
+        # Only do this if an fsspec file-system object is not
+        # already defined. The prefixes may already be stripped.
+        fs, _, paths = get_fs_token_paths(
+            paths, mode="rb", storage_options=storage_options
+        )
     paths = sorted(paths, key=natural_sort_key)  # numeric rather than glob ordering
     ap_kwargs = {"root": root_dir} if root_dir else {}
     root_dir, fns = _analyze_paths(paths, fs, **ap_kwargs)
diff --git a/dask/delayed.py b/dask/delayed.py
index 8f1127826..5537aabb9 100644
--- a/dask/delayed.py
+++ b/dask/delayed.py
@@ -1,3 +1,4 @@
+from dataclasses import is_dataclass, fields
 import operator
 import types
 import uuid
@@ -9,7 +10,6 @@ from tlz import curry, concat, unique, merge
 from . import config, threaded
 from .base import is_dask_collection, dont_optimize, DaskMethodsMixin
 from .base import replace_name_in_key, tokenize as _tokenize
-from .compatibility import is_dataclass, dataclass_fields
 
 from .core import quote
 from .context import globalmethod
@@ -105,7 +105,7 @@ def unpack_collections(expr):
 
     if is_dataclass(expr):
         args, collections = unpack_collections(
-            [[f.name, getattr(expr, f.name)] for f in dataclass_fields(expr)]
+            [[f.name, getattr(expr, f.name)] for f in fields(expr)]
         )
 
         return (apply, typ, (), (dict, args)), collections
@@ -183,7 +183,7 @@ def to_task_dask(expr):
 
     if is_dataclass(expr):
         args, dsk = to_task_dask(
-            [[f.name, getattr(expr, f.name)] for f in dataclass_fields(expr)]
+            [[f.name, getattr(expr, f.name)] for f in fields(expr)]
         )
 
         return (apply, typ, (), (dict, args)), dsk
diff --git a/dask/highlevelgraph.py b/dask/highlevelgraph.py
index 7939b41aa..227c033d5 100644
--- a/dask/highlevelgraph.py
+++ b/dask/highlevelgraph.py
@@ -653,8 +653,13 @@ class HighLevelGraph(Mapping):
 
         raise KeyError(key)
 
-    def __len__(self):
-        return len(self.to_dict())
+    def __len__(self) -> int:
+        # NOTE: this will double-count keys that are duplicated between layers, so it's
+        # possible that `len(hlg) > len(hlg.to_dict())`. However, duplicate keys should
+        # not occur through normal use, and their existence would usually be a bug.
+        # So we ignore this case in favor of better performance.
+        # https://github.com/dask/dask/issues/7271
+        return sum(len(layer) for layer in self.layers.values())
 
     def __iter__(self):
         return iter(self.to_dict())
diff --git a/docs/source/develop.rst b/docs/source/develop.rst
index 04e2767b1..609d74b7d 100644
--- a/docs/source/develop.rst
+++ b/docs/source/develop.rst
@@ -130,7 +130,7 @@ language support, testing, documentation, and style.
 Python Versions
 ~~~~~~~~~~~~~~~
 
-Dask supports Python versions 3.6, 3.7, and 3.8.
+Dask supports Python versions 3.7, 3.8, and 3.9.
 Name changes are handled by the :file:`dask/compatibility.py` file.
 
 Test
@@ -243,7 +243,7 @@ after the line.
 
 .. _numpydoc: https://numpydoc.readthedocs.io/en/latest/format.html#docstring-standard
 
-Docstrings are currently tested under Python 3.6 on Travis.ci.  You can test
+Docstrings are tested under Python 3.8 on GitHub Actions. You can test
 docstrings with pytest as follows::
 
    py.test dask --doctest-modules
diff --git a/setup.py b/setup.py
index c57d267cc..fd481dd6a 100755
--- a/setup.py
+++ b/setup.py
@@ -61,14 +61,13 @@ setup(
     keywords="task-scheduling parallel numpy pandas pydata",
     classifiers=[
         "Programming Language :: Python :: 3",
-        "Programming Language :: Python :: 3.6",
         "Programming Language :: Python :: 3.7",
         "Programming Language :: Python :: 3.8",
         "Programming Language :: Python :: 3.9",
     ],
     packages=packages + tests,
     long_description=open("README.rst").read() if exists("README.rst") else "",
-    python_requires=">=3.6",
+    python_requires=">=3.7",
     install_requires=install_requires,
     setup_requires=setup_requires,
     tests_require=["pytest"],


Test patch:
diff --git a/dask/bytes/tests/test_local.py b/dask/bytes/tests/test_local.py
index 3d48186f6..2b166a596 100644
--- a/dask/bytes/tests/test_local.py
+++ b/dask/bytes/tests/test_local.py
@@ -125,13 +125,6 @@ def test_read_bytes_include_path():
         assert {os.path.split(path)[1] for path in paths} == files.keys()
 
 
-@pytest.mark.xfail(
-    os.environ.get("GITHUB_ACTIONS")
-    and sys.platform == "win32"
-    and sys.version_info[:2] == (3, 6),
-    reason="TODO: Fails on GitHub Actions when running Python 3.6 on Windows."
-    "See https://github.com/dask/dask/pull/5862.",
-)
 def test_with_urls():
     with filetexts(files, mode="b"):
         # OS-independent file:// URI with glob *
diff --git a/dask/bytes/tests/test_s3.py b/dask/bytes/tests/test_s3.py
index 44408874a..115462b2e 100644
--- a/dask/bytes/tests/test_s3.py
+++ b/dask/bytes/tests/test_s3.py
@@ -428,7 +428,8 @@ def test_modification_time_read_bytes(s3, s3so):
 
 
 @pytest.mark.parametrize("engine", ["pyarrow", "fastparquet"])
-def test_parquet(s3, engine, s3so):
+@pytest.mark.parametrize("metadata_file", [True, False])
+def test_parquet(s3, engine, s3so, metadata_file):
     import s3fs
 
     dd = pytest.importorskip("dask.dataframe")
@@ -461,13 +462,19 @@ def test_parquet(s3, engine, s3so):
         index=pd.Index(np.arange(1000), name="foo"),
     )
     df = dd.from_pandas(data, chunksize=500)
-    df.to_parquet(url, engine=engine, storage_options=s3so)
+    df.to_parquet(
+        url, engine=engine, storage_options=s3so, write_metadata_file=metadata_file
+    )
 
     files = [f.split("/")[-1] for f in s3.ls(url)]
-    assert "_common_metadata" in files
+    if metadata_file:
+        assert "_common_metadata" in files
+        assert "_metadata" in files
     assert "part.0.parquet" in files
 
-    df2 = dd.read_parquet(url, index="foo", engine=engine, storage_options=s3so)
+    df2 = dd.read_parquet(
+        url, index="foo", gather_statistics=True, engine=engine, storage_options=s3so
+    )
     assert len(df2.divisions) > 1
 
     tm.assert_frame_equal(data, df2.compute())
diff --git a/dask/tests/test_highgraph.py b/dask/tests/test_highgraph.py
index 772b779b2..fbc7cb55c 100644
--- a/dask/tests/test_highgraph.py
+++ b/dask/tests/test_highgraph.py
@@ -1,3 +1,4 @@
+from dask.blockwise import Blockwise, blockwise_token
 import os
 from collections.abc import Set
 
@@ -215,3 +216,27 @@ def test_highlevelgraph_dicts_deprecation():
         layers = {"a": MaterializedLayer({"x": 1, "y": (inc, "x")})}
         hg = HighLevelGraph(layers, {"a": set()})
         assert hg.dicts == layers
+
+
+def test_len_does_not_materialize():
+    a = {"x": 1}
+    b = Blockwise(
+        output="b",
+        output_indices=tuple("ij"),
+        dsk={"b": [[blockwise_token(0)]]},
+        indices=(),
+        numblocks={},
+        new_axes={"i": (1, 1, 1), "j": (1, 1)},
+    )
+    assert len(b) == len(b.get_output_keys())
+
+    layers = {"a": a, "b": b}
+    dependencies = {"a": set(), "b": {"a"}}
+    hg = HighLevelGraph(layers, dependencies)
+
+    assert hg.layers["a"].is_materialized()
+    assert not hg.layers["b"].is_materialized()
+
+    assert len(hg) == len(a) + len(b) == 7
+
+    assert not hg.layers["b"].is_materialized()