We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Row index does not match pandas on read_csv with skiprows and names on certain files
pandas has a very strange edge case here. The following conditions must be met:

* `skiprows` must be > 1
* `names` must not be `None`
* `index_col` must be `None`
* File must have different data types in each line above `skiprows` number

Since this is data dependent, the best course of action is to hand the decision of what number to start to pandas.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `modin/engines/base/io/text/csv_reader.py`
Content:
```
1 # Licensed to Modin Development Team under one or more contributor license agreements.
2 # See the NOTICE file distributed with this work for additional information regarding
3 # copyright ownership.  The Modin Development Team licenses this file to you under the
4 # Apache License, Version 2.0 (the "License"); you may not use this file except in
5 # compliance with the License.  You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software distributed under
10 # the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
11 # ANY KIND, either express or implied. See the License for the specific language
12 # governing permissions and limitations under the License.
13 
14 from modin.engines.base.io.text.text_file_reader import TextFileReader
15 from modin.data_management.utils import compute_chunksize
16 from pandas.io.parsers import _validate_usecols_arg
17 import pandas
18 import csv
19 import sys
20 
21 
22 class CSVReader(TextFileReader):
23     @classmethod
24     def _read(cls, filepath_or_buffer, **kwargs):
25         if isinstance(filepath_or_buffer, str):
26             if not cls.file_exists(filepath_or_buffer):
27                 return cls.single_worker_read(filepath_or_buffer, **kwargs)
28             filepath_or_buffer = cls.get_path(filepath_or_buffer)
29         elif not cls.pathlib_or_pypath(filepath_or_buffer):
30             return cls.single_worker_read(filepath_or_buffer, **kwargs)
31         compression_type = cls.infer_compression(
32             filepath_or_buffer, kwargs.get("compression")
33         )
34         if compression_type is not None:
35             if (
36                 compression_type == "gzip"
37                 or compression_type == "bz2"
38                 or compression_type == "xz"
39             ):
40                 kwargs["compression"] = compression_type
41             elif (
42                 compression_type == "zip"
43                 and sys.version_info[0] == 3
44                 and sys.version_info[1] >= 7
45             ):
46                 # need python3.7 to .seek and .tell ZipExtFile
47                 kwargs["compression"] = compression_type
48             else:
49                 return cls.single_worker_read(filepath_or_buffer, **kwargs)
50 
51         chunksize = kwargs.get("chunksize")
52         if chunksize is not None:
53             return cls.single_worker_read(filepath_or_buffer, **kwargs)
54 
55         skiprows = kwargs.get("skiprows")
56         if skiprows is not None and not isinstance(skiprows, int):
57             return cls.single_worker_read(filepath_or_buffer, **kwargs)
58         nrows = kwargs.pop("nrows", None)
59         names = kwargs.get("names", None)
60         index_col = kwargs.get("index_col", None)
61         usecols = kwargs.get("usecols", None)
62         if names is None:
63             # For the sake of the empty df, we assume no `index_col` to get the correct
64             # column names before we build the index. Because we pass `names` in, this
65             # step has to happen without removing the `index_col` otherwise it will not
66             # be assigned correctly
67             names = pandas.read_csv(
68                 filepath_or_buffer,
69                 **dict(kwargs, usecols=None, nrows=0, skipfooter=0, index_col=None),
70             ).columns
71         elif index_col is None and not usecols:
72             # When names is set to some list that is smaller than the number of columns
73             # in the file, the first columns are built as a hierarchical index.
74             empty_pd_df = pandas.read_csv(filepath_or_buffer, nrows=0)
75             num_cols = len(empty_pd_df.columns)
76             if num_cols > len(names):
77                 index_col = list(range(num_cols - len(names)))
78                 if len(index_col) == 1:
79                     index_col = index_col[0]
80                 kwargs["index_col"] = index_col
81         empty_pd_df = pandas.read_csv(
82             filepath_or_buffer, **dict(kwargs, nrows=0, skipfooter=0)
83         )
84         column_names = empty_pd_df.columns
85         skipfooter = kwargs.get("skipfooter", None)
86         skiprows = kwargs.pop("skiprows", None)
87         usecols_md = _validate_usecols_arg(usecols)
88         if usecols is not None and usecols_md[1] != "integer":
89             del kwargs["usecols"]
90             all_cols = pandas.read_csv(
91                 cls.file_open(filepath_or_buffer, "rb"),
92                 **dict(kwargs, nrows=0, skipfooter=0),
93             ).columns
94             usecols = all_cols.get_indexer_for(list(usecols_md[0]))
95         parse_dates = kwargs.pop("parse_dates", False)
96         partition_kwargs = dict(
97             kwargs,
98             header=None,
99             names=names,
100             skipfooter=0,
101             skiprows=None,
102             parse_dates=parse_dates,
103             usecols=usecols,
104         )
105         encoding = kwargs.get("encoding", None)
106         quotechar = kwargs.get("quotechar", '"').encode(
107             encoding if encoding is not None else "UTF-8"
108         )
109         is_quoting = kwargs.get("quoting", "") != csv.QUOTE_NONE
110         with cls.file_open(filepath_or_buffer, "rb", compression_type) as f:
111             # Skip the header since we already have the header information and skip the
112             # rows we are told to skip.
113             if isinstance(skiprows, int) or skiprows is None:
114                 if skiprows is None:
115                     skiprows = 0
116                 header = kwargs.get("header", "infer")
117                 if header == "infer" and kwargs.get("names", None) is None:
118                     skiprows += 1
119                 elif isinstance(header, int):
120                     skiprows += header + 1
121                 elif hasattr(header, "__iter__") and not isinstance(header, str):
122                     skiprows += max(header) + 1
123                 cls.offset(
124                     f,
125                     nrows=skiprows,
126                     quotechar=quotechar,
127                     is_quoting=is_quoting,
128                 )
129             if kwargs.get("encoding", None) is not None:
130                 partition_kwargs["skiprows"] = 1
131             # Launch tasks to read partitions
132             partition_ids = []
133             index_ids = []
134             dtypes_ids = []
135             # Max number of partitions available
136             from modin.pandas import DEFAULT_NPARTITIONS
137 
138             num_partitions = DEFAULT_NPARTITIONS
139             # This is the number of splits for the columns
140             num_splits = min(len(column_names), num_partitions)
141             # Metadata
142             column_chunksize = compute_chunksize(empty_pd_df, num_splits, axis=1)
143             if column_chunksize > len(column_names):
144                 column_widths = [len(column_names)]
145                 # This prevents us from unnecessarily serializing a bunch of empty
146                 # objects.
147                 num_splits = 1
148             else:
149                 column_widths = [
150                     column_chunksize
151                     if len(column_names) > (column_chunksize * (i + 1))
152                     else 0
153                     if len(column_names) < (column_chunksize * i)
154                     else len(column_names) - (column_chunksize * i)
155                     for i in range(num_splits)
156                 ]
157 
158             args = {
159                 "fname": filepath_or_buffer,
160                 "num_splits": num_splits,
161                 **partition_kwargs,
162             }
163 
164             splits = cls.partitioned_file(
165                 f,
166                 nrows=nrows,
167                 num_partitions=num_partitions,
168                 quotechar=quotechar,
169                 is_quoting=is_quoting,
170             )
171             for start, end in splits:
172                 args.update({"start": start, "end": end})
173                 partition_id = cls.deploy(cls.parse, num_splits + 2, args)
174                 partition_ids.append(partition_id[:-2])
175                 index_ids.append(partition_id[-2])
176                 dtypes_ids.append(partition_id[-1])
177 
178         # Compute the index based on a sum of the lengths of each partition (by default)
179         # or based on the column(s) that were requested.
180         if index_col is None:
181             row_lengths = cls.materialize(index_ids)
182             new_index = pandas.RangeIndex(sum(row_lengths))
183             # pandas has a really weird edge case here.
184             if kwargs.get("names", None) is not None and skiprows > 1:
185                 new_index = pandas.RangeIndex(
186                     skiprows - 1, new_index.stop + skiprows - 1
187                 )
188         else:
189             index_objs = cls.materialize(index_ids)
190             row_lengths = [len(o) for o in index_objs]
191             new_index = index_objs[0].append(index_objs[1:])
192             new_index.name = empty_pd_df.index.name
193 
194         # Compute dtypes by getting collecting and combining all of the partitions. The
195         # reported dtypes from differing rows can be different based on the inference in
196         # the limited data seen by each worker. We use pandas to compute the exact dtype
197         # over the whole column for each column. The index is set below.
198         dtypes = cls.get_dtypes(dtypes_ids) if len(dtypes_ids) > 0 else None
199 
200         partition_ids = cls.build_partition(partition_ids, row_lengths, column_widths)
201         # If parse_dates is present, the column names that we have might not be
202         # the same length as the returned column names. If we do need to modify
203         # the column names, we remove the old names from the column names and
204         # insert the new one at the front of the Index.
205         if parse_dates is not None:
206             # We have to recompute the column widths if `parse_dates` is set because
207             # we are not guaranteed to have the correct information regarding how many
208             # columns are on each partition.
209             column_widths = None
210             # Check if is list of lists
211             if isinstance(parse_dates, list) and isinstance(parse_dates[0], list):
212                 for group in parse_dates:
213                     new_col_name = "_".join(group)
214                     column_names = column_names.drop(group).insert(0, new_col_name)
215             # Check if it is a dictionary
216             elif isinstance(parse_dates, dict):
217                 for new_col_name, group in parse_dates.items():
218                     column_names = column_names.drop(group).insert(0, new_col_name)
219         # Set the index for the dtypes to the column names
220         if isinstance(dtypes, pandas.Series):
221             dtypes.index = column_names
222         else:
223             dtypes = pandas.Series(dtypes, index=column_names)
224         new_frame = cls.frame_cls(
225             partition_ids,
226             new_index,
227             column_names,
228             row_lengths,
229             column_widths,
230             dtypes=dtypes,
231         )
232         new_query_compiler = cls.query_compiler_cls(new_frame)
233 
234         if skipfooter:
235             new_query_compiler = new_query_compiler.drop(
236                 new_query_compiler.index[-skipfooter:]
237             )
238         if kwargs.get("squeeze", False) and len(new_query_compiler.columns) == 1:
239             return new_query_compiler[new_query_compiler.columns[0]]
240         if index_col is None:
241             new_query_compiler._modin_frame._apply_index_objs(axis=0)
242         return new_query_compiler
243
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/modin/engines/base/io/text/csv_reader.py b/modin/engines/base/io/text/csv_reader.py
--- a/modin/engines/base/io/text/csv_reader.py
+++ b/modin/engines/base/io/text/csv_reader.py
@@ -180,11 +180,6 @@
         if index_col is None:
             row_lengths = cls.materialize(index_ids)
             new_index = pandas.RangeIndex(sum(row_lengths))
-            # pandas has a really weird edge case here.
-            if kwargs.get("names", None) is not None and skiprows > 1:
-                new_index = pandas.RangeIndex(
-                    skiprows - 1, new_index.stop + skiprows - 1
-                )
         else:
             index_objs = cls.materialize(index_ids)
             row_lengths = [len(o) for o in index_objs]

{"golden_diff": "diff --git a/modin/engines/base/io/text/csv_reader.py b/modin/engines/base/io/text/csv_reader.py\n--- a/modin/engines/base/io/text/csv_reader.py\n+++ b/modin/engines/base/io/text/csv_reader.py\n@@ -180,11 +180,6 @@\n         if index_col is None:\n             row_lengths = cls.materialize(index_ids)\n             new_index = pandas.RangeIndex(sum(row_lengths))\n-            # pandas has a really weird edge case here.\n-            if kwargs.get(\"names\", None) is not None and skiprows > 1:\n-                new_index = pandas.RangeIndex(\n-                    skiprows - 1, new_index.stop + skiprows - 1\n-                )\n         else:\n             index_objs = cls.materialize(index_ids)\n             row_lengths = [len(o) for o in index_objs]\n", "issue": "Row index does not match pandas on read_csv with skiprows and names on certain files\npandas has a very strange edge case here. The following conditions must be met:\r\n\r\n* `skiprows` must be > 1\r\n* `names` must not be `None`\r\n* `index_col` must be `None`\r\n* File must have different data types in each line above `skiprows` number\r\n\r\nSince this is data dependent, the best course of action is to hand the decision of what number to start to pandas.\n", "before_files": [{"content": "# Licensed to Modin Development Team under one or more contributor license agreements.\n# See the NOTICE file distributed with this work for additional information regarding\n# copyright ownership.  The Modin Development Team licenses this file to you under the\n# Apache License, Version 2.0 (the \"License\"); you may not use this file except in\n# compliance with the License.  You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software distributed under\n# the License is distributed on an \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF\n# ANY KIND, either express or implied. See the License for the specific language\n# governing permissions and limitations under the License.\n\nfrom modin.engines.base.io.text.text_file_reader import TextFileReader\nfrom modin.data_management.utils import compute_chunksize\nfrom pandas.io.parsers import _validate_usecols_arg\nimport pandas\nimport csv\nimport sys\n\n\nclass CSVReader(TextFileReader):\n    @classmethod\n    def _read(cls, filepath_or_buffer, **kwargs):\n        if isinstance(filepath_or_buffer, str):\n            if not cls.file_exists(filepath_or_buffer):\n                return cls.single_worker_read(filepath_or_buffer, **kwargs)\n            filepath_or_buffer = cls.get_path(filepath_or_buffer)\n        elif not cls.pathlib_or_pypath(filepath_or_buffer):\n            return cls.single_worker_read(filepath_or_buffer, **kwargs)\n        compression_type = cls.infer_compression(\n            filepath_or_buffer, kwargs.get(\"compression\")\n        )\n        if compression_type is not None:\n            if (\n                compression_type == \"gzip\"\n                or compression_type == \"bz2\"\n                or compression_type == \"xz\"\n            ):\n                kwargs[\"compression\"] = compression_type\n            elif (\n                compression_type == \"zip\"\n                and sys.version_info[0] == 3\n                and sys.version_info[1] >= 7\n            ):\n                # need python3.7 to .seek and .tell ZipExtFile\n                kwargs[\"compression\"] = compression_type\n            else:\n                return cls.single_worker_read(filepath_or_buffer, **kwargs)\n\n        chunksize = kwargs.get(\"chunksize\")\n        if chunksize is not None:\n            return cls.single_worker_read(filepath_or_buffer, **kwargs)\n\n        skiprows = kwargs.get(\"skiprows\")\n        if skiprows is not None and not isinstance(skiprows, int):\n            return cls.single_worker_read(filepath_or_buffer, **kwargs)\n        nrows = kwargs.pop(\"nrows\", None)\n        names = kwargs.get(\"names\", None)\n        index_col = kwargs.get(\"index_col\", None)\n        usecols = kwargs.get(\"usecols\", None)\n        if names is None:\n            # For the sake of the empty df, we assume no `index_col` to get the correct\n            # column names before we build the index. Because we pass `names` in, this\n            # step has to happen without removing the `index_col` otherwise it will not\n            # be assigned correctly\n            names = pandas.read_csv(\n                filepath_or_buffer,\n                **dict(kwargs, usecols=None, nrows=0, skipfooter=0, index_col=None),\n            ).columns\n        elif index_col is None and not usecols:\n            # When names is set to some list that is smaller than the number of columns\n            # in the file, the first columns are built as a hierarchical index.\n            empty_pd_df = pandas.read_csv(filepath_or_buffer, nrows=0)\n            num_cols = len(empty_pd_df.columns)\n            if num_cols > len(names):\n                index_col = list(range(num_cols - len(names)))\n                if len(index_col) == 1:\n                    index_col = index_col[0]\n                kwargs[\"index_col\"] = index_col\n        empty_pd_df = pandas.read_csv(\n            filepath_or_buffer, **dict(kwargs, nrows=0, skipfooter=0)\n        )\n        column_names = empty_pd_df.columns\n        skipfooter = kwargs.get(\"skipfooter\", None)\n        skiprows = kwargs.pop(\"skiprows\", None)\n        usecols_md = _validate_usecols_arg(usecols)\n        if usecols is not None and usecols_md[1] != \"integer\":\n            del kwargs[\"usecols\"]\n            all_cols = pandas.read_csv(\n                cls.file_open(filepath_or_buffer, \"rb\"),\n                **dict(kwargs, nrows=0, skipfooter=0),\n            ).columns\n            usecols = all_cols.get_indexer_for(list(usecols_md[0]))\n        parse_dates = kwargs.pop(\"parse_dates\", False)\n        partition_kwargs = dict(\n            kwargs,\n            header=None,\n            names=names,\n            skipfooter=0,\n            skiprows=None,\n            parse_dates=parse_dates,\n            usecols=usecols,\n        )\n        encoding = kwargs.get(\"encoding\", None)\n        quotechar = kwargs.get(\"quotechar\", '\"').encode(\n            encoding if encoding is not None else \"UTF-8\"\n        )\n        is_quoting = kwargs.get(\"quoting\", \"\") != csv.QUOTE_NONE\n        with cls.file_open(filepath_or_buffer, \"rb\", compression_type) as f:\n            # Skip the header since we already have the header information and skip the\n            # rows we are told to skip.\n            if isinstance(skiprows, int) or skiprows is None:\n                if skiprows is None:\n                    skiprows = 0\n                header = kwargs.get(\"header\", \"infer\")\n                if header == \"infer\" and kwargs.get(\"names\", None) is None:\n                    skiprows += 1\n                elif isinstance(header, int):\n                    skiprows += header + 1\n                elif hasattr(header, \"__iter__\") and not isinstance(header, str):\n                    skiprows += max(header) + 1\n                cls.offset(\n                    f,\n                    nrows=skiprows,\n                    quotechar=quotechar,\n                    is_quoting=is_quoting,\n                )\n            if kwargs.get(\"encoding\", None) is not None:\n                partition_kwargs[\"skiprows\"] = 1\n            # Launch tasks to read partitions\n            partition_ids = []\n            index_ids = []\n            dtypes_ids = []\n            # Max number of partitions available\n            from modin.pandas import DEFAULT_NPARTITIONS\n\n            num_partitions = DEFAULT_NPARTITIONS\n            # This is the number of splits for the columns\n            num_splits = min(len(column_names), num_partitions)\n            # Metadata\n            column_chunksize = compute_chunksize(empty_pd_df, num_splits, axis=1)\n            if column_chunksize > len(column_names):\n                column_widths = [len(column_names)]\n                # This prevents us from unnecessarily serializing a bunch of empty\n                # objects.\n                num_splits = 1\n            else:\n                column_widths = [\n                    column_chunksize\n                    if len(column_names) > (column_chunksize * (i + 1))\n                    else 0\n                    if len(column_names) < (column_chunksize * i)\n                    else len(column_names) - (column_chunksize * i)\n                    for i in range(num_splits)\n                ]\n\n            args = {\n                \"fname\": filepath_or_buffer,\n                \"num_splits\": num_splits,\n                **partition_kwargs,\n            }\n\n            splits = cls.partitioned_file(\n                f,\n                nrows=nrows,\n                num_partitions=num_partitions,\n                quotechar=quotechar,\n                is_quoting=is_quoting,\n            )\n            for start, end in splits:\n                args.update({\"start\": start, \"end\": end})\n                partition_id = cls.deploy(cls.parse, num_splits + 2, args)\n                partition_ids.append(partition_id[:-2])\n                index_ids.append(partition_id[-2])\n                dtypes_ids.append(partition_id[-1])\n\n        # Compute the index based on a sum of the lengths of each partition (by default)\n        # or based on the column(s) that were requested.\n        if index_col is None:\n            row_lengths = cls.materialize(index_ids)\n            new_index = pandas.RangeIndex(sum(row_lengths))\n            # pandas has a really weird edge case here.\n            if kwargs.get(\"names\", None) is not None and skiprows > 1:\n                new_index = pandas.RangeIndex(\n                    skiprows - 1, new_index.stop + skiprows - 1\n                )\n        else:\n            index_objs = cls.materialize(index_ids)\n            row_lengths = [len(o) for o in index_objs]\n            new_index = index_objs[0].append(index_objs[1:])\n            new_index.name = empty_pd_df.index.name\n\n        # Compute dtypes by getting collecting and combining all of the partitions. The\n        # reported dtypes from differing rows can be different based on the inference in\n        # the limited data seen by each worker. We use pandas to compute the exact dtype\n        # over the whole column for each column. The index is set below.\n        dtypes = cls.get_dtypes(dtypes_ids) if len(dtypes_ids) > 0 else None\n\n        partition_ids = cls.build_partition(partition_ids, row_lengths, column_widths)\n        # If parse_dates is present, the column names that we have might not be\n        # the same length as the returned column names. If we do need to modify\n        # the column names, we remove the old names from the column names and\n        # insert the new one at the front of the Index.\n        if parse_dates is not None:\n            # We have to recompute the column widths if `parse_dates` is set because\n            # we are not guaranteed to have the correct information regarding how many\n            # columns are on each partition.\n            column_widths = None\n            # Check if is list of lists\n            if isinstance(parse_dates, list) and isinstance(parse_dates[0], list):\n                for group in parse_dates:\n                    new_col_name = \"_\".join(group)\n                    column_names = column_names.drop(group).insert(0, new_col_name)\n            # Check if it is a dictionary\n            elif isinstance(parse_dates, dict):\n                for new_col_name, group in parse_dates.items():\n                    column_names = column_names.drop(group).insert(0, new_col_name)\n        # Set the index for the dtypes to the column names\n        if isinstance(dtypes, pandas.Series):\n            dtypes.index = column_names\n        else:\n            dtypes = pandas.Series(dtypes, index=column_names)\n        new_frame = cls.frame_cls(\n            partition_ids,\n            new_index,\n            column_names,\n            row_lengths,\n            column_widths,\n            dtypes=dtypes,\n        )\n        new_query_compiler = cls.query_compiler_cls(new_frame)\n\n        if skipfooter:\n            new_query_compiler = new_query_compiler.drop(\n                new_query_compiler.index[-skipfooter:]\n            )\n        if kwargs.get(\"squeeze\", False) and len(new_query_compiler.columns) == 1:\n            return new_query_compiler[new_query_compiler.columns[0]]\n        if index_col is None:\n            new_query_compiler._modin_frame._apply_index_objs(axis=0)\n        return new_query_compiler\n", "path": "modin/engines/base/io/text/csv_reader.py"}], "after_files": [{"content": "# Licensed to Modin Development Team under one or more contributor license agreements.\n# See the NOTICE file distributed with this work for additional information regarding\n# copyright ownership.  The Modin Development Team licenses this file to you under the\n# Apache License, Version 2.0 (the \"License\"); you may not use this file except in\n# compliance with the License.  You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software distributed under\n# the License is distributed on an \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF\n# ANY KIND, either express or implied. See the License for the specific language\n# governing permissions and limitations under the License.\n\nfrom modin.engines.base.io.text.text_file_reader import TextFileReader\nfrom modin.data_management.utils import compute_chunksize\nfrom pandas.io.parsers import _validate_usecols_arg\nimport pandas\nimport csv\nimport sys\n\n\nclass CSVReader(TextFileReader):\n    @classmethod\n    def _read(cls, filepath_or_buffer, **kwargs):\n        if isinstance(filepath_or_buffer, str):\n            if not cls.file_exists(filepath_or_buffer):\n                return cls.single_worker_read(filepath_or_buffer, **kwargs)\n            filepath_or_buffer = cls.get_path(filepath_or_buffer)\n        elif not cls.pathlib_or_pypath(filepath_or_buffer):\n            return cls.single_worker_read(filepath_or_buffer, **kwargs)\n        compression_type = cls.infer_compression(\n            filepath_or_buffer, kwargs.get(\"compression\")\n        )\n        if compression_type is not None:\n            if (\n                compression_type == \"gzip\"\n                or compression_type == \"bz2\"\n                or compression_type == \"xz\"\n            ):\n                kwargs[\"compression\"] = compression_type\n            elif (\n                compression_type == \"zip\"\n                and sys.version_info[0] == 3\n                and sys.version_info[1] >= 7\n            ):\n                # need python3.7 to .seek and .tell ZipExtFile\n                kwargs[\"compression\"] = compression_type\n            else:\n                return cls.single_worker_read(filepath_or_buffer, **kwargs)\n\n        chunksize = kwargs.get(\"chunksize\")\n        if chunksize is not None:\n            return cls.single_worker_read(filepath_or_buffer, **kwargs)\n\n        skiprows = kwargs.get(\"skiprows\")\n        if skiprows is not None and not isinstance(skiprows, int):\n            return cls.single_worker_read(filepath_or_buffer, **kwargs)\n        nrows = kwargs.pop(\"nrows\", None)\n        names = kwargs.get(\"names\", None)\n        index_col = kwargs.get(\"index_col\", None)\n        usecols = kwargs.get(\"usecols\", None)\n        if names is None:\n            # For the sake of the empty df, we assume no `index_col` to get the correct\n            # column names before we build the index. Because we pass `names` in, this\n            # step has to happen without removing the `index_col` otherwise it will not\n            # be assigned correctly\n            names = pandas.read_csv(\n                filepath_or_buffer,\n                **dict(kwargs, usecols=None, nrows=0, skipfooter=0, index_col=None),\n            ).columns\n        elif index_col is None and not usecols:\n            # When names is set to some list that is smaller than the number of columns\n            # in the file, the first columns are built as a hierarchical index.\n            empty_pd_df = pandas.read_csv(filepath_or_buffer, nrows=0)\n            num_cols = len(empty_pd_df.columns)\n            if num_cols > len(names):\n                index_col = list(range(num_cols - len(names)))\n                if len(index_col) == 1:\n                    index_col = index_col[0]\n                kwargs[\"index_col\"] = index_col\n        empty_pd_df = pandas.read_csv(\n            filepath_or_buffer, **dict(kwargs, nrows=0, skipfooter=0)\n        )\n        column_names = empty_pd_df.columns\n        skipfooter = kwargs.get(\"skipfooter\", None)\n        skiprows = kwargs.pop(\"skiprows\", None)\n        usecols_md = _validate_usecols_arg(usecols)\n        if usecols is not None and usecols_md[1] != \"integer\":\n            del kwargs[\"usecols\"]\n            all_cols = pandas.read_csv(\n                cls.file_open(filepath_or_buffer, \"rb\"),\n                **dict(kwargs, nrows=0, skipfooter=0),\n            ).columns\n            usecols = all_cols.get_indexer_for(list(usecols_md[0]))\n        parse_dates = kwargs.pop(\"parse_dates\", False)\n        partition_kwargs = dict(\n            kwargs,\n            header=None,\n            names=names,\n            skipfooter=0,\n            skiprows=None,\n            parse_dates=parse_dates,\n            usecols=usecols,\n        )\n        encoding = kwargs.get(\"encoding\", None)\n        quotechar = kwargs.get(\"quotechar\", '\"').encode(\n            encoding if encoding is not None else \"UTF-8\"\n        )\n        is_quoting = kwargs.get(\"quoting\", \"\") != csv.QUOTE_NONE\n        with cls.file_open(filepath_or_buffer, \"rb\", compression_type) as f:\n            # Skip the header since we already have the header information and skip the\n            # rows we are told to skip.\n            if isinstance(skiprows, int) or skiprows is None:\n                if skiprows is None:\n                    skiprows = 0\n                header = kwargs.get(\"header\", \"infer\")\n                if header == \"infer\" and kwargs.get(\"names\", None) is None:\n                    skiprows += 1\n                elif isinstance(header, int):\n                    skiprows += header + 1\n                elif hasattr(header, \"__iter__\") and not isinstance(header, str):\n                    skiprows += max(header) + 1\n                cls.offset(\n                    f,\n                    nrows=skiprows,\n                    quotechar=quotechar,\n                    is_quoting=is_quoting,\n                )\n            if kwargs.get(\"encoding\", None) is not None:\n                partition_kwargs[\"skiprows\"] = 1\n            # Launch tasks to read partitions\n            partition_ids = []\n            index_ids = []\n            dtypes_ids = []\n            # Max number of partitions available\n            from modin.pandas import DEFAULT_NPARTITIONS\n\n            num_partitions = DEFAULT_NPARTITIONS\n            # This is the number of splits for the columns\n            num_splits = min(len(column_names), num_partitions)\n            # Metadata\n            column_chunksize = compute_chunksize(empty_pd_df, num_splits, axis=1)\n            if column_chunksize > len(column_names):\n                column_widths = [len(column_names)]\n                # This prevents us from unnecessarily serializing a bunch of empty\n                # objects.\n                num_splits = 1\n            else:\n                column_widths = [\n                    column_chunksize\n                    if len(column_names) > (column_chunksize * (i + 1))\n                    else 0\n                    if len(column_names) < (column_chunksize * i)\n                    else len(column_names) - (column_chunksize * i)\n                    for i in range(num_splits)\n                ]\n\n            args = {\n                \"fname\": filepath_or_buffer,\n                \"num_splits\": num_splits,\n                **partition_kwargs,\n            }\n\n            splits = cls.partitioned_file(\n                f,\n                nrows=nrows,\n                num_partitions=num_partitions,\n                quotechar=quotechar,\n                is_quoting=is_quoting,\n            )\n            for start, end in splits:\n                args.update({\"start\": start, \"end\": end})\n                partition_id = cls.deploy(cls.parse, num_splits + 2, args)\n                partition_ids.append(partition_id[:-2])\n                index_ids.append(partition_id[-2])\n                dtypes_ids.append(partition_id[-1])\n\n        # Compute the index based on a sum of the lengths of each partition (by default)\n        # or based on the column(s) that were requested.\n        if index_col is None:\n            row_lengths = cls.materialize(index_ids)\n            new_index = pandas.RangeIndex(sum(row_lengths))\n        else:\n            index_objs = cls.materialize(index_ids)\n            row_lengths = [len(o) for o in index_objs]\n            new_index = index_objs[0].append(index_objs[1:])\n            new_index.name = empty_pd_df.index.name\n\n        # Compute dtypes by getting collecting and combining all of the partitions. The\n        # reported dtypes from differing rows can be different based on the inference in\n        # the limited data seen by each worker. We use pandas to compute the exact dtype\n        # over the whole column for each column. The index is set below.\n        dtypes = cls.get_dtypes(dtypes_ids) if len(dtypes_ids) > 0 else None\n\n        partition_ids = cls.build_partition(partition_ids, row_lengths, column_widths)\n        # If parse_dates is present, the column names that we have might not be\n        # the same length as the returned column names. If we do need to modify\n        # the column names, we remove the old names from the column names and\n        # insert the new one at the front of the Index.\n        if parse_dates is not None:\n            # We have to recompute the column widths if `parse_dates` is set because\n            # we are not guaranteed to have the correct information regarding how many\n            # columns are on each partition.\n            column_widths = None\n            # Check if is list of lists\n            if isinstance(parse_dates, list) and isinstance(parse_dates[0], list):\n                for group in parse_dates:\n                    new_col_name = \"_\".join(group)\n                    column_names = column_names.drop(group).insert(0, new_col_name)\n            # Check if it is a dictionary\n            elif isinstance(parse_dates, dict):\n                for new_col_name, group in parse_dates.items():\n                    column_names = column_names.drop(group).insert(0, new_col_name)\n        # Set the index for the dtypes to the column names\n        if isinstance(dtypes, pandas.Series):\n            dtypes.index = column_names\n        else:\n            dtypes = pandas.Series(dtypes, index=column_names)\n        new_frame = cls.frame_cls(\n            partition_ids,\n            new_index,\n            column_names,\n            row_lengths,\n            column_widths,\n            dtypes=dtypes,\n        )\n        new_query_compiler = cls.query_compiler_cls(new_frame)\n\n        if skipfooter:\n            new_query_compiler = new_query_compiler.drop(\n                new_query_compiler.index[-skipfooter:]\n            )\n        if kwargs.get(\"squeeze\", False) and len(new_query_compiler.columns) == 1:\n            return new_query_compiler[new_query_compiler.columns[0]]\n        if index_col is None:\n            new_query_compiler._modin_frame._apply_index_objs(axis=0)\n        return new_query_compiler\n", "path": "modin/engines/base/io/text/csv_reader.py"}]}