Repository: dask/dask

Issue:
Wrong path when reading empty csv file
```python

import dask.dataframe as dd

for k, content in enumerate(["0, 1, 2", "3, 4, 5", "6, 7, 8"]):
    with open(str(k) + ".csv", "w") as file:
        file.write(content)

print(dd.read_csv("*.csv",  include_path_column=True, names=["A", "B", "C"],).compute())
print()

# The second file is empty
for k, content in enumerate(["0, 1, 2", "", "6, 7, 8"]):
    with open(str(k) + ".csv", "w") as file:
        file.write(content)

print(dd.read_csv("*.csv",  include_path_column=True, names=["A", "B", "C"],).compute())

```

returns:

```
   A  B  C    path
0  0  1  2    0.csv
0  3  4  5    1.csv
0  6  7  8    2.csv

   A  B  C    path
0  0  1  2    0.csv
0  6  7  8    1.csv  # <- This is wrong and should be "2.csv"

```


**Environment**:

- Dask version: 2.19.0
- Python version: 3.8.3
- Operating System: Windows 10
- Install method (conda, pip, source): Anaconda


Discussion:

Patch:
diff --git a/dask/array/core.py b/dask/array/core.py
index b2583bbae..a47254ece 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -3903,7 +3903,7 @@ def elemwise(op, *args, **kwargs):
     for arg in args:
         shape = getattr(arg, "shape", ())
         if any(is_dask_collection(x) for x in shape):
-            # Want to excluded Delayed shapes and dd.Scalar
+            # Want to exclude Delayed shapes and dd.Scalar
             shape = ()
         shapes.append(shape)
 
diff --git a/dask/array/reshape.py b/dask/array/reshape.py
index 18cef220e..ac99db2e7 100644
--- a/dask/array/reshape.py
+++ b/dask/array/reshape.py
@@ -159,7 +159,7 @@ def reshape(x, shape):
     shape = tuple(map(sanitize_index, shape))
     known_sizes = [s for s in shape if s != -1]
     if len(known_sizes) < len(shape):
-        if len(known_sizes) - len(shape) > 1:
+        if len(shape) - len(known_sizes) > 1:
             raise ValueError("can only specify one unknown dimension")
         # Fastpath for x.reshape(-1) on 1D arrays, allows unknown shape in x
         # for this case only.
diff --git a/dask/array/routines.py b/dask/array/routines.py
index 8c3ebce2d..d57ae6c09 100644
--- a/dask/array/routines.py
+++ b/dask/array/routines.py
@@ -5,6 +5,7 @@ from collections.abc import Iterable
 from functools import wraps, partial
 from numbers import Real, Integral
 from distutils.version import LooseVersion
+from typing import List
 
 import numpy as np
 from tlz import concat, sliding_window, interleave
@@ -1403,17 +1404,76 @@ def piecewise(x, condlist, funclist, *args, **kw):
     )
 
 
+def aligned_coarsen_chunks(chunks: List[int], multiple: int) -> List[int]:
+    """
+    Returns a new chunking aligned with the coarsening multiple.
+    Any excess is at the end of the array.
+
+    Examples
+    --------
+    >>> aligned_coarsen_chunks(chunks=(1, 2, 3), multiple=4)
+    (4, 2)
+    >>> aligned_coarsen_chunks(chunks=(1, 20, 3, 4), multiple=4)
+    (20, 4, 4)
+    >>> aligned_coarsen_chunks(chunks=(20, 10, 15, 23, 24), multiple=10)
+    (20, 10, 20, 20, 20, 2)
+    """
+
+    def choose_new_size(multiple, q, left):
+        """
+        See if multiple * q is a good choice when 'left' elements are remaining.
+        Else return multiple * (q-1)
+        """
+        possible = multiple * q
+        if (left - possible) > 0:
+            return possible
+        else:
+            return multiple * (q - 1)
+
+    newchunks = []
+    left = sum(chunks) - sum(newchunks)
+    chunkgen = (c for c in chunks)
+    while left > 0:
+        if left < multiple:
+            newchunks.append(left)
+            break
+
+        chunk_size = next(chunkgen, 0)
+        if chunk_size == 0:
+            chunk_size = multiple
+
+        q, r = divmod(chunk_size, multiple)
+        if q == 0:
+            continue
+        elif r == 0:
+            newchunks.append(chunk_size)
+        elif r >= 5:
+            newchunks.append(choose_new_size(multiple, q + 1, left))
+        else:
+            newchunks.append(choose_new_size(multiple, q, left))
+
+        left = sum(chunks) - sum(newchunks)
+
+    return tuple(newchunks)
+
+
 @wraps(chunk.coarsen)
 def coarsen(reduction, x, axes, trim_excess=False, **kwargs):
-    if not trim_excess and not all(
-        bd % div == 0 for i, div in axes.items() for bd in x.chunks[i]
-    ):
+    if not trim_excess and not all(x.shape[i] % div == 0 for i, div in axes.items()):
         msg = "Coarsening factor does not align with block dimensions"
         raise ValueError(msg)
 
     if "dask" in inspect.getfile(reduction):
         reduction = getattr(np, reduction.__name__)
 
+    new_chunks = {}
+    for i, div in axes.items():
+        aligned = aligned_coarsen_chunks(x.chunks[i], div)
+        if aligned != x.chunks[i]:
+            new_chunks[i] = aligned
+    if new_chunks:
+        x = x.rechunk(new_chunks)
+
     name = "coarsen-" + tokenize(reduction, x, axes, trim_excess)
     dsk = {
         (name,)
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index c93a153b3..60c3bab6b 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -1141,8 +1141,8 @@ Dask Name: {name}, {task} tasks"""
             ``'12h'`` or ``pd.Timedelta(hours=12)``.  Assumes a datetime index.
         force : bool, default False
             Allows the expansion of the existing divisions.
-            If False then the new divisions lower and upper bounds must be
-            the same as the old divisions.
+            If False then the new divisions' lower and upper bounds must be
+            the same as the old divisions'.
 
         Notes
         -----
@@ -2801,7 +2801,7 @@ Dask Name: {name}, {task} tasks""".format(
             from the input series and the transformation. Ignored for
             non-callable/dict-like ``index`` or when the input series has
             unknown divisions. Note that this may only be set to ``True`` if
-            you know that the transformed index is monotonicly increasing. Dask
+            you know that the transformed index is monotonically increasing. Dask
             will check that transformed divisions are monotonic, but cannot
             check all the values between divisions, so incorrectly setting this
             can result in bugs.
@@ -3535,7 +3535,10 @@ class DataFrame(_Frame):
             # Slicing with integer labels is always iloc based except for a
             # float indexer for some reason
             if is_integer_slice and not is_float_dtype(self.index.dtype):
-                self.iloc[key]
+                # NOTE: this always fails currently, as iloc is mostly
+                # unsupported, but we call it anyway here for future-proofing
+                # and error-attribution purposes
+                return self.iloc[key]
             else:
                 return self.loc[key]
 
@@ -5336,7 +5339,7 @@ def _rename_dask(df, names):
     Destructively rename columns of dd.DataFrame or name of dd.Series.
     Not for pd.DataFrame or pd.Series.
 
-    Internaly used to overwrite dd.DataFrame.columns and dd.Series.name
+    Internally used to overwrite dd.DataFrame.columns and dd.Series.name
     We can't use map_partition because it applies function then rename
 
     Parameters
diff --git a/dask/dataframe/io/csv.py b/dask/dataframe/io/csv.py
index 769e50b9a..9c6b4e805 100644
--- a/dask/dataframe/io/csv.py
+++ b/dask/dataframe/io/csv.py
@@ -1,3 +1,4 @@
+from os.path import basename
 from collections.abc import Mapping
 from io import BytesIO
 from warnings import warn, catch_warnings, simplefilter
@@ -329,6 +330,12 @@ def text_blocks_to_pandas(
     name = "read-csv-" + tokenize(reader, columns, enforce, head)
 
     if path:
+        block_file_names = [basename(b[1].path) for b in blocks]
+        path = (
+            path[0],
+            [p for p in path[1] if basename(p) in block_file_names],
+        )
+
         colname, paths = path
         head = head.assign(
             **{
diff --git a/dask/dataframe/multi.py b/dask/dataframe/multi.py
index becd0e001..ca635fa4b 100644
--- a/dask/dataframe/multi.py
+++ b/dask/dataframe/multi.py
@@ -11,7 +11,7 @@ There are two important cases:
 2.  We combine along an unpartitioned index or other column
 
 In the first case we know which partitions of each dataframe interact with
-which others.  This lets uss be significantly more clever and efficient.
+which others.  This lets us be significantly more clever and efficient.
 
 In the second case each partition from one dataset interacts with all
 partitions from the other.  We handle this through a shuffle operation.
diff --git a/docs/source/setup/kubernetes-helm.rst b/docs/source/setup/kubernetes-helm.rst
index 40892515f..a9edab600 100644
--- a/docs/source/setup/kubernetes-helm.rst
+++ b/docs/source/setup/kubernetes-helm.rst
@@ -1,11 +1,12 @@
 Kubernetes and Helm
 ===================
 
-It is easy to launch a Dask cluster and a Jupyter notebook server on cloud
+It is easy to launch a Dask cluster and a Jupyter_ notebook server on cloud
 resources using Kubernetes_ and Helm_.
 
 .. _Kubernetes: https://kubernetes.io/
 .. _Helm: https://helm.sh/
+.. _Jupyter: https://jupyter.org/
 
 This is particularly useful when you want to deploy a fresh Python environment
 on Cloud services like Amazon Web Services, Google Compute Engine, or
@@ -15,7 +16,6 @@ If you already have Python environments running in a pre-existing Kubernetes
 cluster, then you may prefer the :doc:`Kubernetes native<kubernetes-native>`
 documentation, which is a bit lighter weight.
 
-
 Launch Kubernetes Cluster
 -------------------------
 
@@ -26,7 +26,7 @@ on one of the common cloud providers like Google, Amazon, or
 Microsoft.  We recommend the first part of the documentation in the guide
 `Zero to JupyterHub <https://zero-to-jupyterhub.readthedocs.io/en/latest/>`_
 that focuses on Kubernetes and Helm (you do not need to follow all of these
-instructions).  Also, JupyterHub is not necessary to deploy Dask:
+instructions). In particular, you don't need to install JupyterHub.
 
 - `Creating a Kubernetes Cluster <https://zero-to-jupyterhub.readthedocs.io/en/latest/create-k8s-cluster.html>`_
 - `Setting up Helm <https://zero-to-jupyterhub.readthedocs.io/en/latest/setup-helm.html>`_
@@ -34,9 +34,8 @@ instructions).  Also, JupyterHub is not necessary to deploy Dask:
 Alternatively, you may want to experiment with Kubernetes locally using
 `Minikube <https://kubernetes.io/docs/getting-started-guides/minikube/>`_.
 
-
-Helm Install Dask
------------------
+Which Chart is Right for You?
+-----------------------------
 
 Dask maintains a Helm chart repository containing various charts for the Dask community
 https://helm.dask.org/ .
@@ -45,7 +44,29 @@ You will need to add this to your known channels and update your local charts::
    helm repo add dask https://helm.dask.org/
    helm repo update
 
-Now, you can launch Dask on your Kubernetes cluster using the Dask Helm_ chart::
+We provides two Helm charts. The right one to choose depends on whether you're
+deploying Dask for a single user or for many users.
+
+
+================  =====================================================================
+Helm Chart        Use Case
+================  =====================================================================
+``dask/dask``     Single-user deployment with one notebook server and one Dask Cluster.
+``dask/daskhub``  Multi-user deployment with JupyterHub and Dask Gateway.
+================  =====================================================================
+
+See :ref:`kubernetes-helm.single` or :ref:`kubernetes-helm.multi` for detailed
+instructions on deploying either of these.
+As you might suspect, deploying ``dask/daskhub`` is a bit more complicated since
+there are more components. If you're just deploying for a single user we'd recommend
+using ``dask/dask``.
+
+.. _kubernetes-helm.single:
+
+Helm Install Dask for a Single User
+-----------------------------------
+
+Once your Kubernetes cluster is ready, you can deploy dask using the Dask Helm_ chart::
 
    helm install my-dask dask/dask
 
@@ -54,7 +75,7 @@ also an optional Jupyter server.
 
 
 Verify Deployment
------------------
+^^^^^^^^^^^^^^^^^
 
 This might take a minute to deploy.  You can check its status with
 ``kubectl``::
@@ -93,7 +114,7 @@ Additionally, you can list all active helm deployments with::
 
 
 Connect to Dask and Jupyter
----------------------------
+^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
 When we ran ``kubectl get services``, we saw some externally visible IPs:
 
@@ -129,7 +150,7 @@ variable automatically.
 
 
 Configure Environment
----------------------
+^^^^^^^^^^^^^^^^^^^^^
 
 By default, the Helm deployment launches three workers using one core each and
 a standard conda environment. We can customize this environment by creating a
@@ -186,7 +207,7 @@ list``
 
 
 Check status and logs
----------------------
+^^^^^^^^^^^^^^^^^^^^^
 
 For standard issues, you should be able to see the worker status and logs using the
 Dask dashboard (in particular, you can see the worker links from the ``info/`` page).
@@ -229,7 +250,7 @@ their logs with the following commands:
 
 
 Delete a Helm deployment
-------------------------
+^^^^^^^^^^^^^^^^^^^^^^^^
 
 You can always delete a helm deployment using its name::
 
@@ -240,7 +261,7 @@ Cloud service (you will need to delete those explicitly).
 
 
 Avoid the Jupyter Server
-------------------------
+^^^^^^^^^^^^^^^^^^^^^^^^
 
 Sometimes you do not need to run a Jupyter server alongside your Dask cluster.
 
@@ -248,3 +269,139 @@ Sometimes you do not need to run a Jupyter server alongside your Dask cluster.
 
    jupyter:
      enabled: false
+
+.. _kubernetes-helm.multi:
+
+Helm Install Dask for Mulitple Users
+------------------------------------
+
+The ``dask/daskhub`` Helm Chart deploys JupyterHub_, `Dask Gateway`_, and configures
+the two to work well together. In particular, Dask Gateway is registered as
+a JupyterHub service so that Dask Gateway can re-use JupyterHub's authentication,
+and the JupyterHub environment is configured to connect to the Dask Gateway
+without any arguments.
+
+.. note::
+
+   The ``dask/daskhub`` helm chart came out of the `Pangeo`_ project, a community
+   platform for big data geoscience.
+   
+.. _Pangeo: http://pangeo.io/
+.. _Dask Gateway: https://gateway.dask.org/
+.. _JupyterHub: https://jupyterhub.readthedocs.io/en/stable/
+
+The ``dask/daskhub`` helm chart uses the JupyterHub and Dask-Gateway helm charts.
+You'll want to consult the `JupyterHub helm documentation <https://zero-to-jupyterhub.readthedocs.io/en/latest/setup-jupyterhub/setup-jupyterhub.html>`_ and
+and `Dask Gateway helm documentation <https://gateway.dask.org/install-kube.html>`_ for further customization. The default values
+are at https://github.com/dask/helm-chart/blob/master/daskhub/values.yaml.
+
+Verify that you've set up a Kubernetes cluster and added Dask's helm charts:
+
+.. code-block:: console
+
+   $ helm repo add dask https://helm.dask.org/
+   $ helm repo update
+
+JupyterHub and Dask Gateway require a few secret tokens. We'll generate them
+on the command line and insert the tokens in a ``secrets.yaml`` file that will
+be passed to Helm.
+
+Run the following command, and copy the output. This is our `token-1`.
+
+.. code-block:: console
+
+   $ openssl rand -hex 32  # generate token-1
+
+Run command again and copy the output again. This is our `token-2`.
+
+.. code-block:: console
+
+   $ openssl rand -hex 32  # generate token-2
+
+Now substitute those two values for ``<token-1>`` and ``<token-2>`` below.
+Note that ``<token-2>`` is used twice, once for ``jupyterhub.hub.services.dask-gateway.apiToken``, and a second time for ``dask-gateway.gateway.auth.jupyterhub.apiToken``.
+
+.. code-block:: yaml
+
+   # file: secrets.yaml
+   jupyterhub:
+     proxy:
+       secretToken: "<token-1>"
+     hub:
+       services:
+         dask-gateway:
+           apiToken: "<token-2>"
+
+   dask-gateway:
+     gateway:
+       auth:
+         jupyterhub:
+           apiToken: "<token-2>"
+
+Now we're ready to install DaskHub
+
+.. code-block:: console
+
+   $ helm upgrade --wait --install --render-subchart-notes \
+       dhub dask/daskhub \
+       --values=secrets.yaml
+
+
+The output explains how to find the IPs for your JupyterHub depoyment.
+
+.. code-block:: console
+
+   $ kubectl get service proxy-public
+   NAME           TYPE           CLUSTER-IP      EXTERNAL-IP      PORT(S)                      AGE
+   proxy-public   LoadBalancer   10.43.249.239   35.202.158.223   443:31587/TCP,80:30500/TCP   2m40s
+
+
+Creating a Dask Cluster
+^^^^^^^^^^^^^^^^^^^^^^^
+
+To create a Dask cluster on this deployment, users need to connect to the Dask Gateway
+
+.. code-block:: python
+
+   >>> from dask_gateway import GatewayCluster
+   >>> cluster = GatewayCluster()
+   >>> client = cluster.get_client()
+   >>> cluster
+
+Depending on the configuration, users may need to ``cluster.scale(n)`` to
+get workers. See https://gateway.dask.org/ for more on Dask Gateway.
+
+Matching the User Environment
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Dask Clients will be running the JupyterHub's singleuser environment. To ensure
+that the same environment is used for the scheduler and workers, you can provide
+it as a Gateway option and configure the ``singleuser`` environment to default
+to the value set by JupyterHub.
+
+.. code-block:: yaml
+
+   # config.yaml
+   jupyterhub:
+     singleuser:
+       extraEnv:
+         DASK_GATEWAY__CLUSTER__OPTIONS__IMAGE: '{JUPYTER_IMAGE_SPEC}'
+
+   dask-gateway:
+     extraConfig:
+       optionHandler: |
+         from dask_gateway_server.options import Options, Integer, Float, String
+         def option_handler(options):
+             if ":" not in options.image:
+                 raise ValueError("When specifying an image you must also provide a tag")
+             return {
+                 "image": options.image,
+             }
+         c.Backend.cluster_options = Options(
+             String("image", default="pangeo/base-notebook:2020.07.28", label="Image"),
+             handler=option_handler,
+         )
+
+The user environment will need to include ``dask-gateway``. Any packages installed
+manually after the ``singleuser`` pod started will not be included in the worker
+environment.
diff --git a/docs/source/setup/kubernetes.rst b/docs/source/setup/kubernetes.rst
index 2ab9fae15..9d561cb9c 100644
--- a/docs/source/setup/kubernetes.rst
+++ b/docs/source/setup/kubernetes.rst
@@ -13,23 +13,28 @@ particularly in the cloud.  You can use Kubernetes to launch Dask workers in the
 following two ways:
 
 1.  **Helm**:
-    You can launch a Dask scheduler, several workers, and an optional Jupyter Notebook server
-    on a Kubernetes easily using Helm_
+
+    You can deploy Dask and (optionally) Jupyter or JupyterHub on Kubernetes
+    easily using Helm_
 
     .. code-block:: bash
 
        helm repo add dask https://helm.dask.org/    # add the Dask Helm chart repository
        helm repo update                             # get latest Helm charts
-       helm install dask/dask                     # deploy standard Dask chart
+       # For single-user deployments, use dask/dask
+       helm install dask/dask                       # deploy standard Dask chart
+       # For multi-user deployments, use dask/daskhub
+       helm install dask/daskhub                    # deploy JupyterHub & Dask
 
     This is a good choice if you want to do the following:
 
     1.  Run a managed Dask cluster for a long period of time
-    2.  Also deploy a Jupyter server from which to run code
+    2.  Also deploy a Jupyter / JupyterHub server from which to run code
     3.  Share the same Dask cluster between many automated services
     4.  Try out Dask for the first time on a cloud-based system
-        like Amazon, Google, or Microsoft Azure
-        (see also our :doc:`Cloud documentation <cloud>`)
+        like Amazon, Google, or Microsoft Azure where you already have
+        a Kubernetes cluster. If you don't already have Kubernetes deployed,
+        see our :doc:`Cloud documentation <cloud>`.
 
     .. note::
 


Test patch:
diff --git a/dask/array/tests/test_array_core.py b/dask/array/tests/test_array_core.py
index ec65561d2..197159398 100644
--- a/dask/array/tests/test_array_core.py
+++ b/dask/array/tests/test_array_core.py
@@ -4138,7 +4138,7 @@ def test_dask_array_holds_scipy_sparse_containers():
 
     z = x.T.map_blocks(scipy.sparse.csr_matrix)
     zz = z.compute(scheduler="single-threaded")
-    assert isinstance(yy, scipy.sparse.spmatrix)
+    assert isinstance(zz, scipy.sparse.spmatrix)
     assert (zz == xx.T).all()
 
 
diff --git a/dask/array/tests/test_reshape.py b/dask/array/tests/test_reshape.py
index c453f28a6..3bdceb86a 100644
--- a/dask/array/tests/test_reshape.py
+++ b/dask/array/tests/test_reshape.py
@@ -1,6 +1,8 @@
 import pytest
 import numpy as np
+import dask.array as da
 from dask.array.reshape import reshape_rechunk, expand_tuple, contract_tuple
+from dask.array.utils import assert_eq
 
 
 @pytest.mark.parametrize(
@@ -61,3 +63,19 @@ def test_contract_tuple():
     assert contract_tuple((1, 1, 2, 5, 1), 2) == (2, 2, 4, 2)
     assert contract_tuple((2, 4), 2) == (2, 4)
     assert contract_tuple((2, 4), 3) == (6,)
+
+
+def test_reshape_unknown_sizes():
+    a = np.random.random((10, 6, 6))
+    A = da.from_array(a, chunks=(5, 2, 3))
+
+    a2 = a.reshape((60, -1))
+    A2 = A.reshape((60, -1))
+
+    assert A2.shape == (60, 6)
+    assert_eq(A2, a2)
+
+    with pytest.raises(ValueError):
+        a.reshape((60, -1, -1))
+    with pytest.raises(ValueError):
+        A.reshape((60, -1, -1))
diff --git a/dask/array/tests/test_routines.py b/dask/array/tests/test_routines.py
index b657d9098..532262bb9 100644
--- a/dask/array/tests/test_routines.py
+++ b/dask/array/tests/test_routines.py
@@ -1521,13 +1521,31 @@ def test_coarsen():
 
 def test_coarsen_with_excess():
     x = da.arange(10, chunks=5)
-    assert_eq(da.coarsen(np.min, x, {0: 3}, trim_excess=True), np.array([0, 5]))
+    assert_eq(da.coarsen(np.min, x, {0: 5}, trim_excess=True), np.array([0, 5]))
     assert_eq(
         da.coarsen(np.sum, x, {0: 3}, trim_excess=True),
-        np.array([0 + 1 + 2, 5 + 6 + 7]),
+        np.array([0 + 1 + 2, 3 + 4 + 5, 6 + 7 + 8]),
     )
 
 
+def test_coarsen_bad_chunks():
+
+    x1 = da.arange(10, chunks=5)
+    x2 = x1.rechunk((1, 2, 3, 4))
+    assert_eq(da.coarsen(np.sum, x1, {0: 5}), da.coarsen(np.sum, x2, {0: 5}))
+
+
+def test_aligned_coarsen_chunks():
+
+    from ..routines import aligned_coarsen_chunks as acc
+
+    assert acc((20, 10, 15, 23, 24), 10) == (20, 10, 20, 20, 20, 2)
+    assert acc((20, 10, 15, 42, 23, 24), 10) == (20, 10, 20, 40, 20, 20, 4)
+    assert acc((20, 10, 15, 47, 23, 24), 10) == (20, 10, 20, 50, 20, 10, 9)
+    assert acc((2, 10, 15, 47, 23, 24), 10) == (10, 20, 50, 20, 20, 1)
+    assert acc((10, 20, 30, 40, 2), 10) == (10, 20, 30, 40, 2)
+
+
 def test_insert():
     x = np.random.randint(10, size=(10, 10))
     a = da.from_array(x, chunks=(5, 5))
diff --git a/dask/dataframe/io/tests/test_csv.py b/dask/dataframe/io/tests/test_csv.py
index dd2a667d7..75045e51e 100644
--- a/dask/dataframe/io/tests/test_csv.py
+++ b/dask/dataframe/io/tests/test_csv.py
@@ -1564,3 +1564,26 @@ def test_to_csv_line_ending():
 def test_block_mask(block_lists):
     mask = list(block_mask(block_lists))
     assert len(mask) == len(list(flatten(block_lists)))
+
+
+def test_reading_empty_csv_files_with_path():
+    with tmpdir() as tdir:
+        for k, content in enumerate(["0, 1, 2", "", "6, 7, 8"]):
+            with open(os.path.join(tdir, str(k) + ".csv"), "w") as file:
+                file.write(content)
+        result = dd.read_csv(
+            os.path.join(tdir, "*.csv"),
+            include_path_column=True,
+            converters={"path": parse_filename},
+            names=["A", "B", "C"],
+        ).compute()
+        df = pd.DataFrame(
+            {
+                "A": [0, 6],
+                "B": [1, 7],
+                "C": [2, 8],
+                "path": ["0.csv", "2.csv"],
+            }
+        )
+        df["path"] = df["path"].astype("category")
+        assert_eq(result, df, check_index=False)
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index 28a0beb22..8b2aca252 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -2609,7 +2609,7 @@ def test_to_dask_array_raises(as_frame):
         a.to_dask_array(5)
 
 
-@pytest.mark.parametrize("as_frame", [False, False])
+@pytest.mark.parametrize("as_frame", [False, True])
 def test_to_dask_array_unknown(as_frame):
     s = pd.Series([1, 2, 3, 4, 5], name="foo")
     a = dd.from_pandas(s, chunksize=2)
@@ -2622,11 +2622,12 @@ def test_to_dask_array_unknown(as_frame):
     result = result.chunks
 
     if as_frame:
+        assert len(result) == 2
         assert result[1] == (1,)
+    else:
+        assert len(result) == 1
 
-    assert len(result) == 1
     result = result[0]
-
     assert len(result) == 2
     assert all(np.isnan(x) for x in result)