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:
Would you mind expanding your example code to generate some files too, and then include the traceback, so that we can be sure talk about exactly the same thing, and help with diagnosis?
( @jsignell , for the include_path_column logic)
I am not sure if this is a regression or not, but I can reproduce. Thanks for bringking this up @Sturla22. The issue is that the number of blocks does not neccessarily correspond to the number of files. I'll have a PR shortly. 

Patch:
diff --git a/dask/dataframe/io/csv.py b/dask/dataframe/io/csv.py
--- 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 @@ def __getitem__(self, key):
             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


Test patch:
diff --git a/dask/dataframe/io/tests/test_csv.py b/dask/dataframe/io/tests/test_csv.py
--- 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.