Repository: modin-project/modin
Pull request: Fix case when Index column named 'Unnamed:' after reading DataFrame via read_csv
<!--
Thank you for your contribution!
Please review the contributing docs: https://modin.readthedocs.io/en/latest/CONTRIBUTING.html
if you have questions about contributing.
-->

## What do these changes do?

<!-- Please give a short brief about these changes. -->

- [x] commit message follows format outlined [here](https://modin.readthedocs.io/en/latest/contributing.html)
- [x] passes `flake8 modin`
- [x] passes `black --check modin`
- [x] signed commit with `git commit -s` <!-- you can amend your commit with a signature via `git commit -amend -s` -->
- [x] Resolves #3119 <!-- issue must be created for each patch -->
- [x] tests added and passing
- [x] module layout described at `docs/developer/architecture.rst` is up-to-date <!-- if you have added, renamed or removed files or directories please update the documentation accordingly -->


Problem statement:
Index column named 'Unnamed: 0' on `.to_csv()` after reading DataFrame with `.read_csv()`
### System information
- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**: Ubuntu 20.10
- **Modin version** (`modin.__version__`): 0.9.1
- **Python version**: 3.7.4
- **Code we can use to reproduce**:
```
import numpy as np
import pandas as pd
import os
os.environ["MODIN_ENGINE"] = "ray" 
import modin.pandas as pd2

df1 = pd.DataFrame(np.arange(12).reshape(4, 3), columns=list('abc'),
                   index=[f'i{i}' for i in range(1, 5)])
df2 = pd2.DataFrame(np.arange(12).reshape(4, 3), columns=list('abc'),
                    index=[f'i{i}' for i in range(1, 5)])

df1.to_csv('df1.csv')
df2.to_csv('df2.csv')

df1_re = pd.read_csv('df1.csv', index_col=0)
df2_re = pd2.read_csv('df2.csv', index_col=0)

df1_re.to_csv('df1_re.csv')
df2_re.to_csv('df2_re.csv')
```
This gives 
```
$ head *re*csv
==> df1_re.csv <==
,a,b,c
i1,0,1,2
i2,3,4,5
i3,6,7,8
i4,9,10,11

==> df2_re.csv <==
Unnamed: 0,a,b,c
i1,0,1,2
i2,3,4,5
i3,6,7,8
i4,9,10,11
```

### Describe the problem
After reading a CSV and writing it again the resulting file has `Unnamed: 0` as header of the index column, whereas a file that has been written with pandas does not (compare `df1_re.csv` and `df2_re.csv` after running the example). This only occurs with Ray, while Dask gives ``UserWarning: `to_csv` defaulting to pandas implementation.`` and writes the file as expected.

Hints from issue discussion:
Thanks @julibeg for the report!

cc @anmyachev 
Some details.

After creating `df2` by the constructor,  `df2` has 1 partition:
```python
    a   b   c
i1  0   1   2
i2  3   4   5
i3  6   7   8
i4  9  10  11
```
After creating `df2_re` by `read_csv`, `df2_re` has 2 partitions:
```python
(Pdb) df2_re._query_compiler._modin_frame._partitions[0][0].to_pandas()
            a  b  c
Unnamed: 0
i1          0  1  2
i2          3  4  5
(Pdb) df2_re._query_compiler._modin_frame._partitions[1][0].to_pandas()
            a   b   c
Unnamed: 0
i3          6   7   8
i4          9  10  11
```

It turns out in some cases that a dataframe created from the same data but in different ways has a different internal representation. Need to understand why.

Commit message: FIX-#3119: handle files with BOM

Signed-off-by: Alexander Myskov <alexander.myskov@intel.com>

Patch to review:
diff --git a/modin/backends/pandas/parsers.py b/modin/backends/pandas/parsers.py
index cb9d4185d86..56f9a16f670 100644
--- a/modin/backends/pandas/parsers.py
+++ b/modin/backends/pandas/parsers.py
@@ -210,16 +210,22 @@ def parse(fname, **kwargs):
         num_splits = kwargs.pop("num_splits", None)
         start = kwargs.pop("start", None)
         end = kwargs.pop("end", None)
-        index_col = kwargs.get("index_col", None)
+        header_size = kwargs.pop("header_size", None)
         if start is not None and end is not None:
             # pop "compression" from kwargs because bio is uncompressed
             bio = FileDispatcher.file_open(
                 fname, "rb", kwargs.pop("compression", "infer")
             )
-            if kwargs.get("encoding", None) is not None:
-                header = b"" + bio.readline()
-            else:
-                header = b""
+            header = b""
+            # In this case we beware that fisrt line can contain BOM, so
+            # adding this line to the `header` for reading and then skip it
+            if kwargs.get("encoding", None) is not None and header_size == 0:
+                header += bio.readline()
+                # `skiprows` can be only None here, so don't check it's type
+                # and just set to 1
+                kwargs["skiprows"] = 1
+            for _ in range(header_size):
+                header += bio.readline()
             bio.seek(start)
             to_read = header + bio.read(end - start)
             bio.close()
@@ -227,11 +233,11 @@ def parse(fname, **kwargs):
         else:
             # This only happens when we are reading with only one worker (Default)
             return pandas.read_csv(fname, **kwargs)
-        if index_col is not None:
-            index = pandas_df.index
-        else:
-            # The lengths will become the RangeIndex
-            index = len(pandas_df)
+        index = (
+            pandas_df.index
+            if not isinstance(pandas_df.index, pandas.RangeIndex)
+            else len(pandas_df)
+        )
         return _split_result_for_readers(1, num_splits, pandas_df) + [
             index,
             pandas_df.dtypes,
diff --git a/modin/engines/base/io/text/csv_dispatcher.py b/modin/engines/base/io/text/csv_dispatcher.py
index 27a55fabc44..d165ea739cb 100644
--- a/modin/engines/base/io/text/csv_dispatcher.py
+++ b/modin/engines/base/io/text/csv_dispatcher.py
@@ -74,12 +74,17 @@ def _read(cls, filepath_or_buffer, **kwargs):
         names = kwargs.get("names", lib.no_default)
         index_col = kwargs.get("index_col", None)
         encoding = kwargs.get("encoding", None)
-        skiprows = kwargs.get("skiprows")
+        skiprows = kwargs.get("skiprows", None)
 
         is_quoting = kwargs.get("quoting", "") != QUOTE_NONE
         quotechar = kwargs.get("quotechar", '"').encode(
             encoding if encoding is not None else "UTF-8"
         )
+        # In these cases we should pass additional metadata
+        # to the workers to match pandas output
+        pass_names = names in [None, lib.no_default] and (
+            skiprows is not None or kwargs.get("skipfooter", 0) != 0
+        )
 
         # Define header size for further skipping (Header can be skipped because header
         # information will be obtained further from empty_df, so no need to handle it
@@ -93,28 +98,6 @@ def _read(cls, filepath_or_buffer, **kwargs):
         # simultaneously
         skiprows = skiprows + header_size if skiprows else header_size
 
-        # Now we need to define parameters, which are common for all partitions. These
-        # parameters can be `sniffed` from empty dataframes created further
-        if names in [lib.no_default, None]:
-            # When reading the one row df, we assume no `index_col` to get the correct
-            # column names before we build the index. Because we pass `names` in, this
-            # step has to happen without removing the `index_col` otherwise it will not
-            # be assigned correctly
-            names = pandas.read_csv(
-                filepath_or_buffer,
-                **dict(kwargs, usecols=None, nrows=1, skipfooter=0, index_col=None),
-            ).columns
-        elif index_col is None and not kwargs.get("usecols", None):
-            # When names is set to some list that is smaller than the number of columns
-            # in the file, the first columns are built as a hierarchical index.
-            empty_pd_df = pandas.read_csv(
-                filepath_or_buffer, nrows=0, encoding=encoding
-            )
-            num_cols = len(empty_pd_df.columns)
-            if num_cols > len(names):
-                index_col = list(range(num_cols - len(names)))
-                if len(index_col) == 1:
-                    index_col = index_col[0]
         pd_df_metadata = pandas.read_csv(
             filepath_or_buffer,
             **dict(kwargs, nrows=1, skipfooter=0, index_col=index_col),
@@ -135,13 +118,12 @@ def _read(cls, filepath_or_buffer, **kwargs):
             kwargs,
             fname=filepath_or_buffer_md,
             num_splits=num_splits,
-            header=None,
-            names=names,
+            header_size=header_size if not pass_names else 0,
+            names=names if not pass_names else column_names,
             skipfooter=0,
-            skiprows=1 if encoding is not None else None,
+            skiprows=None,
             nrows=None,
             compression=compression_infered,
-            index_col=index_col,
         )
 
         with cls.file_open(filepath_or_buffer_md, "rb", compression_infered) as f:
@@ -162,7 +144,7 @@ def _read(cls, filepath_or_buffer, **kwargs):
             partition_ids=partition_ids,
             index_ids=index_ids,
             dtypes_ids=dtypes_ids,
-            index_col_md=index_col,
+            index_col=index_col,
             index_name=pd_df_metadata.index.name,
             column_widths=column_widths,
             column_names=column_names,
@@ -220,13 +202,16 @@ def _read_csv_check_support(
         if skiprows is not None and not isinstance(skiprows, int):
             return False
 
+        # can't pickle dialect object to the worker process
+        if read_csv_kwargs.get("dialect", None) is not None:
+            return False
+
         return True
 
     @classmethod
     def _define_index(
         cls,
         index_ids: list,
-        index_col: IndexColType,
         index_name: str,
     ) -> Tuple[IndexColType, list]:
         """
@@ -236,8 +221,6 @@ def _define_index(
         ----------
         index_ids : list
             Array with references to the partitions index objects.
-        index_col : IndexColType
-            `index_col` parameter of read_csv function.
         index_name : str
             Name that should be assigned to the index if `index_col`
             is not provided.
@@ -249,11 +232,12 @@ def _define_index(
         row_lengths : list
             Partitions rows lengths.
         """
-        if index_col is None:
-            row_lengths = cls.materialize(index_ids)
-            new_index = pandas.RangeIndex(sum(row_lengths))
+        index_objs = cls.materialize(index_ids)
+        simple_idx = isinstance(index_objs[0], int) if len(index_objs) > 0 else True
+        if simple_idx:
+            row_lengths = index_objs
+            new_index = pandas.RangeIndex(sum(index_objs))
         else:
-            index_objs = cls.materialize(index_ids)
             row_lengths = [len(o) for o in index_objs]
             new_index = index_objs[0].append(index_objs[1:])
             new_index.name = index_name
@@ -266,7 +250,7 @@ def _get_new_qc(
         partition_ids: list,
         index_ids: list,
         dtypes_ids: list,
-        index_col_md: IndexColType,
+        index_col: IndexColType,
         index_name: str,
         column_widths: list,
         column_names: ColumnNamesTypes,
@@ -283,8 +267,8 @@ def _get_new_qc(
             Array with references to the partitions index objects.
         dtypes_ids : list
             Array with references to the partitions dtypes objects.
-        index_col_md : IndexColType
-            `index_col` parameter passed to the workers.
+        index_col : IndexColType
+            `index_col` parameter of `read_csv` function.
         index_name : str
             Name that should be assigned to the index if `index_col`
             is not provided.
@@ -300,7 +284,7 @@ def _get_new_qc(
         new_query_compiler : BaseQueryCompiler
             New query compiler, created from `new_frame`.
         """
-        new_index, row_lengths = cls._define_index(index_ids, index_col_md, index_name)
+        new_index, row_lengths = cls._define_index(index_ids, index_name)
         # Compute dtypes by getting collecting and combining all of the partitions. The
         # reported dtypes from differing rows can be different based on the inference in
         # the limited data seen by each worker. We use pandas to compute the exact dtype
@@ -331,7 +315,7 @@ def _get_new_qc(
             )
         if kwargs.get("squeeze", False) and len(new_query_compiler.columns) == 1:
             return new_query_compiler[new_query_compiler.columns[0]]
-        if index_col_md is None:
+        if index_col is None:
             new_query_compiler._modin_frame.synchronize_labels(axis=0)
 
         return new_query_compiler
diff --git a/modin/pandas/test/data/issue_3119.csv b/modin/pandas/test/data/issue_3119.csv
new file mode 100644
index 00000000000..ff323915339
--- /dev/null
+++ b/modin/pandas/test/data/issue_3119.csv
@@ -0,0 +1,5 @@
+,a,b,c
+i1,0,1,2
+i2,3,4,5
+i3,6,7,8
+i4,9,10,11
diff --git a/modin/pandas/test/test_io.py b/modin/pandas/test/test_io.py
index 0dc26af6f12..96669df2140 100644
--- a/modin/pandas/test/test_io.py
+++ b/modin/pandas/test/test_io.py
@@ -15,6 +15,7 @@
 import numpy as np
 import pandas
 from pandas.errors import ParserWarning
+import pandas._libs.lib as lib
 from collections import OrderedDict
 from modin.config import TestDatasetSize, Engine, Backend, IsExperimental
 from modin.utils import to_pandas
@@ -250,15 +251,11 @@ def test_read_csv_delimiters(
         )
 
     # Column and Index Locations and Names tests
-    @pytest.mark.skipif(
-        Engine.get() != "Python" and Backend.get() != "Omnisci",
-        reason="many parameters combiantions fails: issue #2312, #2307",
-    )
     @pytest.mark.parametrize("header", ["infer", None, 0])
     @pytest.mark.parametrize("index_col", [None, "col1"])
     @pytest.mark.parametrize("prefix", [None, "_", "col"])
     @pytest.mark.parametrize(
-        "names", [None, ["col1"], ["c1", "c2", "c3", "c4", "c5", "c6", "c7"]]
+        "names", [lib.no_default, ["col1"], ["c1", "c2", "c3", "c4", "c5", "c6", "c7"]]
     )
     @pytest.mark.parametrize(
         "usecols", [None, ["col1"], ["col1", "col2", "col6"], [0, 1, 5]]
@@ -278,6 +275,8 @@ def test_read_csv_col_handling(
             pytest.xfail(
                 reason="The reason of tests fail in `cloud` mode is unknown for now - issue #2340"
             )
+        if names is lib.no_default:
+            pytest.skip("some parameters combiantions fails: issue #2312")
         eval_io(
             fn_name="read_csv",
             # read_csv kwargs
@@ -642,10 +641,6 @@ def test_read_csv_file_format(
             pytest.xfail(
                 "read_csv with Ray engine fails with some 'escapechar' parameters - issue #2494"
             )
-        elif Engine.get() != "Python" and dialect:
-            pytest.xfail(
-                "read_csv with Ray engine fails with `dialect` parameter - issue #2508"
-            )
 
         unique_filename = get_unique_filename()
         if dialect:
@@ -773,12 +768,6 @@ def test_read_csv_internal(
         memory_map,
         float_precision,
     ):
-        if Engine.get() != "Python" and delimiter == " ":
-            pytest.xfail(
-                "read_csv with Ray engine doesn't \
-                raise exceptions while Pandas raises - issue #2320"
-            )
-
         # In this case raised TypeError: cannot use a string pattern on a bytes-like object,
         # so TypeError should be excluded from raising_exceptions list in order to check, that
         # the same exceptions are raised by Pandas and Modin
@@ -1110,6 +1099,18 @@ def test_read_csv_file_handle(self, request, read_mode, make_csv_file):
         finally:
             teardown_test_files([unique_filename])
 
+    def test_unnamed_index(self):
+        def get_internal_df(df):
+            partition = read_df._query_compiler._modin_frame._partitions[0][0]
+            return partition.to_pandas()
+
+        path = "modin/pandas/test/data/issue_3119.csv"
+        read_df = pd.read_csv(path, index_col=0)
+        assert get_internal_df(read_df).index.name is None
+        read_df = pd.read_csv(path, index_col=[0, 1])
+        for name1, name2 in zip(get_internal_df(read_df).index.names, [None, "a"]):
+            assert name1 == name2
+
 
 class TestTable:
     def test_read_table(self, make_csv_file):


Code review comments:
diff_hunk: @@ -220,13 +202,16 @@ def _read_csv_check_support(
         if skiprows is not None and not isinstance(skiprows, int):
             return False
 
+        # can't pickle dialect object to the worker process
+        if read_csv_kwargs.get("dialect", None) is not None:
+            return False
path: modin/engines/base/io/text/csv_dispatcher.py
text: @user1:
are you sure this is relevant to the issue you're stating as being closed by this PR?

@user2:
Not sure) change was reverted.

diff_hunk: @@ -249,11 +232,12 @@ def _define_index(
         row_lengths : list
             Partitions rows lengths.
         """
-        if index_col is None:
-            row_lengths = cls.materialize(index_ids)
-            new_index = pandas.RangeIndex(sum(row_lengths))
+        index_objs = cls.materialize(index_ids)
+        simple_idx = isinstance(index_objs[0], int) if len(index_objs) > 0 else True
+        if simple_idx:
path: modin/engines/base/io/text/csv_dispatcher.py
text: @user1:
```suggestion
        if len(index_objs) == 0 or isinstance(index_objs[0], int):
```