Repository: dask/dask

Issue:
IndexError when supplying multiple paths to dask.dataframe.read_csv() and including a path column
I fixed my installation (dask version 2.30) locally by adding
` and i < len(self.paths)` to this line
https://github.com/dask/dask/blob/fbccc4ef3e1974da2b4a9cb61aa83c1e6d61efba/dask/dataframe/io/csv.py#L82

I bet there is a deeper reason for why `i` is exceeding the length of paths but I could not identify it with a quick look through the neighboring code.

Minimal example to reproduce, assuming multiple csv files in the CWD:
```python
import dask.dataframe as dd
dd.read_csv("*.csv", include_path_column=True).compute() 
```

Discussion:
jsignell: I'll merge this if it passes CI since it only affects when `include_path_column==True` and the behavior was definitely wrong before. 

Patch:
diff --git a/dask/array/overlap.py b/dask/array/overlap.py
index 8b400ca9a..9170ec909 100644
--- a/dask/array/overlap.py
+++ b/dask/array/overlap.py
@@ -820,6 +820,15 @@ def coerce_depth(ndim, depth):
         for i in range(ndim):
             if i not in depth:
                 depth[i] = 0
+    return coerce_depth_type(ndim, depth)
+
+
+def coerce_depth_type(ndim, depth):
+    for i in range(ndim):
+        if isinstance(depth[i], tuple):
+            depth[i] = tuple(int(d) for d in depth[i])
+        else:
+            depth[i] = int(depth[i])
     return depth
 
 
diff --git a/dask/dataframe/io/csv.py b/dask/dataframe/io/csv.py
index a93235c68..dbf1b9a5d 100644
--- a/dask/dataframe/io/csv.py
+++ b/dask/dataframe/io/csv.py
@@ -1,4 +1,3 @@
-from os.path import basename
 from collections.abc import Mapping
 from io import BytesIO
 from warnings import warn, catch_warnings, simplefilter
@@ -78,9 +77,12 @@ class CSVSubgraph(Mapping):
             raise KeyError(key)
 
         block = self.blocks[i]
-
         if self.paths is not None:
-            path_info = (self.colname, self.paths[i], self.paths)
+            path_info = (
+                self.colname,
+                self.paths[i],
+                list(self.head[self.colname].cat.categories),
+            )
         else:
             path_info = None
 
@@ -190,7 +192,7 @@ def pandas_read_text(
     dtypes : dict
         DTypes to assign to columns
     path : tuple
-        A tuple containing path column name, path to file, and all paths.
+        A tuple containing path column name, path to file, and an ordered list of paths.
 
     See Also
     --------
@@ -341,7 +343,7 @@ def text_blocks_to_pandas(
     kwargs : dict
         Keyword arguments to pass down to ``reader``
     path : tuple, optional
-        A tuple containing column name for path and list of all paths
+        A tuple containing column name for path and the path_converter if provided
 
     Returns
     -------
@@ -385,20 +387,19 @@ def text_blocks_to_pandas(
     name = "read-csv-" + tokenize(reader, columns, enforce, head, blocksize)
 
     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
+        colname, path_converter = path
+        paths = [b[1].path for b in blocks]
+        if path_converter:
+            paths = [path_converter(p) for p in paths]
         head = head.assign(
             **{
                 colname: pd.Categorical.from_codes(
-                    np.zeros(len(head), dtype=int), paths
+                    np.zeros(len(head), dtype=int), set(paths)
                 )
             }
         )
+        path = (colname, paths)
+
     if len(unknown_categoricals):
         head = clear_known_categories(head, cols=unknown_categoricals)
 
@@ -543,9 +544,7 @@ def read_pandas(
 
     if include_path_column:
         b_sample, values, paths = b_out
-        if path_converter:
-            paths = [path_converter(path) for path in paths]
-        path = (include_path_column, paths)
+        path = (include_path_column, path_converter)
     else:
         b_sample, values = b_out
         path = None
diff --git a/docs/source/high-level-graphs.rst b/docs/source/high-level-graphs.rst
index 5be2fed1f..65264bcd0 100644
--- a/docs/source/high-level-graphs.rst
+++ b/docs/source/high-level-graphs.rst
@@ -157,7 +157,7 @@ constructor:
       'add': {('add', 0): (operator.add, ('read-csv', 0), 100),
               ('add', 1): (operator.add, ('read-csv', 1), 100),
               ('add', 2): (operator.add, ('read-csv', 2), 100),
-              ('add', 3): (operator.add, ('read-csv', 3), 100)}
+              ('add', 3): (operator.add, ('read-csv', 3), 100)},
 
       'filter': {('filter', 0): (lambda part: part[part.name == 'Alice'], ('add', 0)),
                  ('filter', 1): (lambda part: part[part.name == 'Alice'], ('add', 1)),


Test patch:
diff --git a/dask/array/tests/test_overlap.py b/dask/array/tests/test_overlap.py
index cf830dd51..712fbe844 100644
--- a/dask/array/tests/test_overlap.py
+++ b/dask/array/tests/test_overlap.py
@@ -246,6 +246,13 @@ def test_overlap():
     assert_eq(g, expected)
     assert same_keys(g, overlap(d, depth={0: 2, 1: 1}, boundary={0: 100, 1: "reflect"}))
 
+    u_depth = np.uint16([2, 1])
+    u_depth = {k: v for k, v in enumerate(u_depth)}
+    g = overlap(d, depth=u_depth, boundary={0: 100, 1: "reflect"})
+    assert g.chunks == ((8, 8), (6, 6))
+    assert_eq(g, expected)
+    assert same_keys(g, overlap(d, depth={0: 2, 1: 1}, boundary={0: 100, 1: "reflect"}))
+
     g = overlap(d, depth={0: 2, 1: 1}, boundary={0: 100, 1: "none"})
     expected = np.array(
         [
@@ -270,6 +277,12 @@ def test_overlap():
     assert_eq(g, expected)
     assert g.chunks == ((8, 8), (5, 5))
 
+    u_depth = np.uint16([2, 1])
+    u_depth = {k: v for k, v in enumerate(u_depth)}
+    g = overlap(d, depth=u_depth, boundary={0: 100, 1: "none"})
+    assert_eq(g, expected)
+    assert g.chunks == ((8, 8), (5, 5))
+
 
 def test_asymmetric_overlap_boundary_exception():
     x = da.arange(10, chunks=5)
diff --git a/dask/dataframe/io/tests/test_csv.py b/dask/dataframe/io/tests/test_csv.py
index 1e482bf58..7a6aa8ef4 100644
--- a/dask/dataframe/io/tests/test_csv.py
+++ b/dask/dataframe/io/tests/test_csv.py
@@ -458,6 +458,23 @@ def test_read_csv_include_path_column_is_dtype_category(dd_read, files):
         assert has_known_categories(result.path)
 
 
+@pytest.mark.parametrize(
+    "dd_read,files", [(dd.read_csv, csv_files), (dd.read_table, tsv_files)]
+)
+@read_table_mark
+def test_read_csv_include_path_column_with_multiple_partitions_per_file(dd_read, files):
+    with filetexts(files, mode="b"):
+        df = dd_read("2014-01-*.csv", blocksize="10B", include_path_column=True)
+        assert df.npartitions > 3
+        assert df.path.dtype == "category"
+        assert has_known_categories(df.path)
+
+        dfs = dd_read("2014-01-*.csv", blocksize="10B", include_path_column=True)
+        result = dfs.compute()
+        assert result.path.dtype == "category"
+        assert has_known_categories(result.path)
+
+
 # After this point, we test just using read_csv, as all functionality
 # for both is implemented using the same code.