We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Encoding error : `UnicodeDecodeError: 'utf-8' codec can't decode byte 0xe9 in position 2: invalid continuation byte`
<!--
General questions should be asked on the mailing list modin-dev@googlegroups.com.

Before submitting an issue, please fill out the following form.
-->

### System information
- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**: ubuntu 18.04
- **Modin installed from : pip install modin[ray]
- **Modin version**: 0.6.3
- **Python version**: 3.7.3
<!--
You can obtain the Modin version with

python -c "import modin; print(modin.__version__)"
-->

### Describe the problem
<!-- Describe the problem clearly here. -->
Hello, 
i'm trying to use modin to reduce the memory peak due the volum of the data, so i change the pandas with modin.pandas, i try to do a simple read of a file but encoded in 'latin-1' (french) . With pandas all goes smoothly but using modin i got an error of encoding as follow : 
`UnicodeDecodeError: 'utf-8' codec can't decode byte 0xe9 in position 2: invalid continuation byte`

the script used (which works fine on pandas but not in modin )  : 
`caract = pd.read_csv(path, sep="\t", encoding = "ISO-8859-1")`


ps ::  i tried other encoding and the same remark : works on pandas and not on modin (backed by ray) : ISO-8859-1, ISO-8859-9, latin-1

any solution ??

thanks 
### Source code / logs
`RayTaskError: ray_worker (pid=10815, host=ubuntu)
  File "pandas/_libs/parsers.pyx", line 1297, in pandas._libs.parsers.TextReader._string_convert
  File "pandas/_libs/parsers.pyx", line 1520, in pandas._libs.parsers._string_box_utf8
UnicodeDecodeError: 'utf-8' codec can't decode byte 0xe9 in position 2: invalid continuation byte

During handling of the above exception, another exception occurred:

ray_worker (pid=10815, host=ubuntu)
  File "/home/lasngd/.conda/envs/pytorch/lib/python3.7/site-packages/modin/engines/ray/task_wrapper.py", line 8, in deploy_ray_func
    return func(**args)
  File "/home/lasngd/.conda/envs/pytorch/lib/python3.7/site-packages/modin/backends/pandas/parsers.py", line 69, in parse
    pandas_df = pandas.read_csv(BytesIO(to_read), **kwargs)
  File "/home/lasngd/.conda/envs/pytorch/lib/python3.7/site-packages/pandas/io/parsers.py", line 685, in parser_f
    return _read(filepath_or_buffer, kwds)
  File "/home/lasngd/.conda/envs/pytorch/lib/python3.7/site-packages/pandas/io/parsers.py", line 463, in _read
    data = parser.read(nrows)
  File "/home/lasngd/.conda/envs/pytorch/lib/python3.7/site-packages/pandas/io/parsers.py", line 1154, in read
    ret = self._engine.read(nrows)
  File "/home/lasngd/.conda/envs/pytorch/lib/python3.7/site-packages/pandas/io/parsers.py", line 2059, in read
    data = self._reader.read(nrows)
  File "pandas/_libs/parsers.pyx", line 881, in pandas._libs.parsers.TextReader.read
  File "pandas/_libs/parsers.pyx", line 896, in pandas._libs.parsers.TextReader._read_low_memory
  File "pandas/_libs/parsers.pyx", line 973, in pandas._libs.parsers.TextReader._read_rows
  File "pandas/_libs/parsers.pyx", line 1105, in pandas._libs.parsers.TextReader._convert_column_data
  File "pandas/_libs/parsers.pyx", line 1158, in pandas._libs.parsers.TextReader._convert_tokens
  File "pandas/_libs/parsers.pyx", line 1281, in pandas._libs.parsers.TextReader._convert_with_dtype
  File "pandas/_libs/parsers.pyx", line 1297, in pandas._libs.parsers.TextReader._string_convert
  File "pandas/_libs/parsers.pyx", line 1520, in pandas._libs.parsers._string_box_utf8
UnicodeDecodeError: 'utf-8' codec can't decode byte 0xe9 in position 2: invalid continuation byte`
<!-- Include any logs or source code that would be helpful to diagnose the problem. If including tracebacks, please include the full traceback. Large logs and files should be attached. Try to provide a reproducible test case that is the bare minimum necessary to generate the problem. -->


--- 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/backends/pandas/parsers.py`
Content:
```
1 import numpy as np
2 import pandas
3 from pandas.core.dtypes.cast import find_common_type
4 from pandas.core.dtypes.concat import union_categoricals
5 from pandas.io.common import _infer_compression
6 from modin.engines.base.io import FileReader
7 from modin.data_management.utils import split_result_of_axis_func_pandas
8 from modin.error_message import ErrorMessage
9 from io import BytesIO
10 
11 
12 def _split_result_for_readers(axis, num_splits, df):  # pragma: no cover
13     """Splits the DataFrame read into smaller DataFrames and handles all edge cases.
14 
15     Args:
16         axis: Which axis to split over.
17         num_splits: The number of splits to create.
18         df: The DataFrame after it has been read.
19 
20     Returns:
21         A list of pandas DataFrames.
22     """
23     splits = split_result_of_axis_func_pandas(axis, num_splits, df)
24     if not isinstance(splits, list):
25         splits = [splits]
26     return splits
27 
28 
29 def find_common_type_cat(types):
30     if all(isinstance(t, pandas.CategoricalDtype) for t in types):
31         if all(t.ordered for t in types):
32             return pandas.CategoricalDtype(
33                 np.sort(np.unique([c for t in types for c in t.categories])[0]),
34                 ordered=True,
35             )
36         return union_categoricals(
37             [pandas.Categorical([], dtype=t) for t in types],
38             sort_categories=all(t.ordered for t in types),
39         ).dtype
40     else:
41         return find_common_type(types)
42 
43 
44 class PandasParser(object):
45     @classmethod
46     def get_dtypes(cls, dtypes_ids):
47         return (
48             pandas.concat(cls.materialize(dtypes_ids), axis=1)
49             .apply(lambda row: find_common_type_cat(row.values), axis=1)
50             .squeeze(axis=0)
51         )
52 
53     @classmethod
54     def single_worker_read(cls, fname, **kwargs):
55         ErrorMessage.default_to_pandas("Parameters provided")
56         # Use default args for everything
57         pandas_frame = cls.parse(fname, **kwargs)
58         if isinstance(pandas_frame, pandas.io.parsers.TextFileReader):
59             pd_read = pandas_frame.read
60             pandas_frame.read = lambda *args, **kwargs: cls.query_compiler_cls.from_pandas(
61                 pd_read(*args, **kwargs), cls.frame_cls
62             )
63             return pandas_frame
64         return cls.query_compiler_cls.from_pandas(pandas_frame, cls.frame_cls)
65 
66     infer_compression = _infer_compression
67 
68 
69 class PandasCSVParser(PandasParser):
70     @staticmethod
71     def parse(fname, **kwargs):
72         num_splits = kwargs.pop("num_splits", None)
73         start = kwargs.pop("start", None)
74         end = kwargs.pop("end", None)
75         index_col = kwargs.get("index_col", None)
76         if start is not None and end is not None:
77             # pop "compression" from kwargs because bio is uncompressed
78             bio = FileReader.file_open(fname, "rb", kwargs.pop("compression", "infer"))
79             if kwargs.pop("encoding", False):
80                 header = b"" + bio.readline()
81             else:
82                 header = b""
83             bio.seek(start)
84             to_read = header + bio.read(end - start)
85             bio.close()
86             pandas_df = pandas.read_csv(BytesIO(to_read), **kwargs)
87         else:
88             # This only happens when we are reading with only one worker (Default)
89             return pandas.read_csv(fname, **kwargs)
90         if index_col is not None:
91             index = pandas_df.index
92         else:
93             # The lengths will become the RangeIndex
94             index = len(pandas_df)
95         return _split_result_for_readers(1, num_splits, pandas_df) + [
96             index,
97             pandas_df.dtypes,
98         ]
99 
100 
101 class PandasJSONParser(PandasParser):
102     @staticmethod
103     def parse(fname, **kwargs):
104         num_splits = kwargs.pop("num_splits", None)
105         start = kwargs.pop("start", None)
106         end = kwargs.pop("end", None)
107         if start is not None and end is not None:
108             # pop "compression" from kwargs because bio is uncompressed
109             bio = FileReader.file_open(fname, "rb", kwargs.pop("compression", "infer"))
110             bio.seek(start)
111             to_read = b"" + bio.read(end - start)
112             bio.close()
113             columns = kwargs.pop("columns")
114             pandas_df = pandas.read_json(BytesIO(to_read), **kwargs)
115         else:
116             # This only happens when we are reading with only one worker (Default)
117             return pandas.read_json(fname, **kwargs)
118         if not pandas_df.columns.equals(columns):
119             raise NotImplementedError("Columns must be the same across all rows.")
120         partition_columns = pandas_df.columns
121         return _split_result_for_readers(1, num_splits, pandas_df) + [
122             len(pandas_df),
123             pandas_df.dtypes,
124             partition_columns,
125         ]
126 
127 
128 class PandasParquetParser(PandasParser):
129     @staticmethod
130     def parse(fname, **kwargs):
131         import pyarrow.parquet as pq
132 
133         num_splits = kwargs.pop("num_splits", None)
134         columns = kwargs.get("columns", None)
135         if num_splits is None:
136             return pandas.read_parquet(fname, **kwargs)
137         kwargs["use_pandas_metadata"] = True
138         df = pq.read_table(fname, **kwargs).to_pandas()
139         if columns is not None:
140             df = df[columns]
141         # Append the length of the index here to build it externally
142         return _split_result_for_readers(0, num_splits, df) + [len(df.index), df.dtypes]
143 
144 
145 class PandasHDFParser(PandasParser):  # pragma: no cover
146     @staticmethod
147     def parse(fname, **kwargs):
148         kwargs["key"] = kwargs.pop("_key", None)
149         num_splits = kwargs.pop("num_splits", None)
150         if num_splits is None:
151             return pandas.read_hdf(fname, **kwargs)
152         df = pandas.read_hdf(fname, **kwargs)
153         # Append the length of the index here to build it externally
154         return _split_result_for_readers(0, num_splits, df) + [len(df.index), df.dtypes]
155 
156 
157 class PandasFeatherParser(PandasParser):
158     @staticmethod
159     def parse(fname, **kwargs):
160         from pyarrow import feather
161 
162         num_splits = kwargs.pop("num_splits", None)
163         if num_splits is None:
164             return pandas.read_feather(fname, **kwargs)
165         df = feather.read_feather(fname, **kwargs)
166         # Append the length of the index here to build it externally
167         return _split_result_for_readers(0, num_splits, df) + [len(df.index), df.dtypes]
168 
169 
170 class PandasSQLParser(PandasParser):
171     @staticmethod
172     def parse(sql, con, index_col, **kwargs):
173         num_splits = kwargs.pop("num_splits", None)
174         if num_splits is None:
175             return pandas.read_sql(sql, con, index_col=index_col, **kwargs)
176         df = pandas.read_sql(sql, con, index_col=index_col, **kwargs)
177         if index_col is None:
178             index = len(df)
179         else:
180             index = df.index
181         return _split_result_for_readers(1, num_splits, df) + [index, df.dtypes]
182
```
--- 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/backends/pandas/parsers.py b/modin/backends/pandas/parsers.py
--- a/modin/backends/pandas/parsers.py
+++ b/modin/backends/pandas/parsers.py
@@ -76,7 +76,7 @@
         if start is not None and end is not None:
             # pop "compression" from kwargs because bio is uncompressed
             bio = FileReader.file_open(fname, "rb", kwargs.pop("compression", "infer"))
-            if kwargs.pop("encoding", False):
+            if kwargs.get("encoding", None) is not None:
                 header = b"" + bio.readline()
             else:
                 header = b""

{"golden_diff": "diff --git a/modin/backends/pandas/parsers.py b/modin/backends/pandas/parsers.py\n--- a/modin/backends/pandas/parsers.py\n+++ b/modin/backends/pandas/parsers.py\n@@ -76,7 +76,7 @@\n         if start is not None and end is not None:\n             # pop \"compression\" from kwargs because bio is uncompressed\n             bio = FileReader.file_open(fname, \"rb\", kwargs.pop(\"compression\", \"infer\"))\n-            if kwargs.pop(\"encoding\", False):\n+            if kwargs.get(\"encoding\", None) is not None:\n                 header = b\"\" + bio.readline()\n             else:\n                 header = b\"\"\n", "issue": "Encoding error : `UnicodeDecodeError: 'utf-8' codec can't decode byte 0xe9 in position 2: invalid continuation byte`\n<!--\r\nGeneral questions should be asked on the mailing list modin-dev@googlegroups.com.\r\n\r\nBefore submitting an issue, please fill out the following form.\r\n-->\r\n\r\n### System information\r\n- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**: ubuntu 18.04\r\n- **Modin installed from : pip install modin[ray]\r\n- **Modin version**: 0.6.3\r\n- **Python version**: 3.7.3\r\n<!--\r\nYou can obtain the Modin version with\r\n\r\npython -c \"import modin; print(modin.__version__)\"\r\n-->\r\n\r\n### Describe the problem\r\n<!-- Describe the problem clearly here. -->\r\nHello, \r\ni'm trying to use modin to reduce the memory peak due the volum of the data, so i change the pandas with modin.pandas, i try to do a simple read of a file but encoded in 'latin-1' (french) . With pandas all goes smoothly but using modin i got an error of encoding as follow : \r\n`UnicodeDecodeError: 'utf-8' codec can't decode byte 0xe9 in position 2: invalid continuation byte`\r\n\r\nthe script used (which works fine on pandas but not in modin )  : \r\n`caract = pd.read_csv(path, sep=\"\\t\", encoding = \"ISO-8859-1\")`\r\n\r\n\r\nps ::  i tried other encoding and the same remark : works on pandas and not on modin (backed by ray) : ISO-8859-1, ISO-8859-9, latin-1\r\n\r\nany solution ??\r\n\r\nthanks \r\n### Source code / logs\r\n`RayTaskError: ray_worker (pid=10815, host=ubuntu)\r\n  File \"pandas/_libs/parsers.pyx\", line 1297, in pandas._libs.parsers.TextReader._string_convert\r\n  File \"pandas/_libs/parsers.pyx\", line 1520, in pandas._libs.parsers._string_box_utf8\r\nUnicodeDecodeError: 'utf-8' codec can't decode byte 0xe9 in position 2: invalid continuation byte\r\n\r\nDuring handling of the above exception, another exception occurred:\r\n\r\nray_worker (pid=10815, host=ubuntu)\r\n  File \"/home/lasngd/.conda/envs/pytorch/lib/python3.7/site-packages/modin/engines/ray/task_wrapper.py\", line 8, in deploy_ray_func\r\n    return func(**args)\r\n  File \"/home/lasngd/.conda/envs/pytorch/lib/python3.7/site-packages/modin/backends/pandas/parsers.py\", line 69, in parse\r\n    pandas_df = pandas.read_csv(BytesIO(to_read), **kwargs)\r\n  File \"/home/lasngd/.conda/envs/pytorch/lib/python3.7/site-packages/pandas/io/parsers.py\", line 685, in parser_f\r\n    return _read(filepath_or_buffer, kwds)\r\n  File \"/home/lasngd/.conda/envs/pytorch/lib/python3.7/site-packages/pandas/io/parsers.py\", line 463, in _read\r\n    data = parser.read(nrows)\r\n  File \"/home/lasngd/.conda/envs/pytorch/lib/python3.7/site-packages/pandas/io/parsers.py\", line 1154, in read\r\n    ret = self._engine.read(nrows)\r\n  File \"/home/lasngd/.conda/envs/pytorch/lib/python3.7/site-packages/pandas/io/parsers.py\", line 2059, in read\r\n    data = self._reader.read(nrows)\r\n  File \"pandas/_libs/parsers.pyx\", line 881, in pandas._libs.parsers.TextReader.read\r\n  File \"pandas/_libs/parsers.pyx\", line 896, in pandas._libs.parsers.TextReader._read_low_memory\r\n  File \"pandas/_libs/parsers.pyx\", line 973, in pandas._libs.parsers.TextReader._read_rows\r\n  File \"pandas/_libs/parsers.pyx\", line 1105, in pandas._libs.parsers.TextReader._convert_column_data\r\n  File \"pandas/_libs/parsers.pyx\", line 1158, in pandas._libs.parsers.TextReader._convert_tokens\r\n  File \"pandas/_libs/parsers.pyx\", line 1281, in pandas._libs.parsers.TextReader._convert_with_dtype\r\n  File \"pandas/_libs/parsers.pyx\", line 1297, in pandas._libs.parsers.TextReader._string_convert\r\n  File \"pandas/_libs/parsers.pyx\", line 1520, in pandas._libs.parsers._string_box_utf8\r\nUnicodeDecodeError: 'utf-8' codec can't decode byte 0xe9 in position 2: invalid continuation byte`\r\n<!-- Include any logs or source code that would be helpful to diagnose the problem. If including tracebacks, please include the full traceback. Large logs and files should be attached. Try to provide a reproducible test case that is the bare minimum necessary to generate the problem. -->\r\n\n", "before_files": [{"content": "import numpy as np\nimport pandas\nfrom pandas.core.dtypes.cast import find_common_type\nfrom pandas.core.dtypes.concat import union_categoricals\nfrom pandas.io.common import _infer_compression\nfrom modin.engines.base.io import FileReader\nfrom modin.data_management.utils import split_result_of_axis_func_pandas\nfrom modin.error_message import ErrorMessage\nfrom io import BytesIO\n\n\ndef _split_result_for_readers(axis, num_splits, df):  # pragma: no cover\n    \"\"\"Splits the DataFrame read into smaller DataFrames and handles all edge cases.\n\n    Args:\n        axis: Which axis to split over.\n        num_splits: The number of splits to create.\n        df: The DataFrame after it has been read.\n\n    Returns:\n        A list of pandas DataFrames.\n    \"\"\"\n    splits = split_result_of_axis_func_pandas(axis, num_splits, df)\n    if not isinstance(splits, list):\n        splits = [splits]\n    return splits\n\n\ndef find_common_type_cat(types):\n    if all(isinstance(t, pandas.CategoricalDtype) for t in types):\n        if all(t.ordered for t in types):\n            return pandas.CategoricalDtype(\n                np.sort(np.unique([c for t in types for c in t.categories])[0]),\n                ordered=True,\n            )\n        return union_categoricals(\n            [pandas.Categorical([], dtype=t) for t in types],\n            sort_categories=all(t.ordered for t in types),\n        ).dtype\n    else:\n        return find_common_type(types)\n\n\nclass PandasParser(object):\n    @classmethod\n    def get_dtypes(cls, dtypes_ids):\n        return (\n            pandas.concat(cls.materialize(dtypes_ids), axis=1)\n            .apply(lambda row: find_common_type_cat(row.values), axis=1)\n            .squeeze(axis=0)\n        )\n\n    @classmethod\n    def single_worker_read(cls, fname, **kwargs):\n        ErrorMessage.default_to_pandas(\"Parameters provided\")\n        # Use default args for everything\n        pandas_frame = cls.parse(fname, **kwargs)\n        if isinstance(pandas_frame, pandas.io.parsers.TextFileReader):\n            pd_read = pandas_frame.read\n            pandas_frame.read = lambda *args, **kwargs: cls.query_compiler_cls.from_pandas(\n                pd_read(*args, **kwargs), cls.frame_cls\n            )\n            return pandas_frame\n        return cls.query_compiler_cls.from_pandas(pandas_frame, cls.frame_cls)\n\n    infer_compression = _infer_compression\n\n\nclass PandasCSVParser(PandasParser):\n    @staticmethod\n    def parse(fname, **kwargs):\n        num_splits = kwargs.pop(\"num_splits\", None)\n        start = kwargs.pop(\"start\", None)\n        end = kwargs.pop(\"end\", None)\n        index_col = kwargs.get(\"index_col\", None)\n        if start is not None and end is not None:\n            # pop \"compression\" from kwargs because bio is uncompressed\n            bio = FileReader.file_open(fname, \"rb\", kwargs.pop(\"compression\", \"infer\"))\n            if kwargs.pop(\"encoding\", False):\n                header = b\"\" + bio.readline()\n            else:\n                header = b\"\"\n            bio.seek(start)\n            to_read = header + bio.read(end - start)\n            bio.close()\n            pandas_df = pandas.read_csv(BytesIO(to_read), **kwargs)\n        else:\n            # This only happens when we are reading with only one worker (Default)\n            return pandas.read_csv(fname, **kwargs)\n        if index_col is not None:\n            index = pandas_df.index\n        else:\n            # The lengths will become the RangeIndex\n            index = len(pandas_df)\n        return _split_result_for_readers(1, num_splits, pandas_df) + [\n            index,\n            pandas_df.dtypes,\n        ]\n\n\nclass PandasJSONParser(PandasParser):\n    @staticmethod\n    def parse(fname, **kwargs):\n        num_splits = kwargs.pop(\"num_splits\", None)\n        start = kwargs.pop(\"start\", None)\n        end = kwargs.pop(\"end\", None)\n        if start is not None and end is not None:\n            # pop \"compression\" from kwargs because bio is uncompressed\n            bio = FileReader.file_open(fname, \"rb\", kwargs.pop(\"compression\", \"infer\"))\n            bio.seek(start)\n            to_read = b\"\" + bio.read(end - start)\n            bio.close()\n            columns = kwargs.pop(\"columns\")\n            pandas_df = pandas.read_json(BytesIO(to_read), **kwargs)\n        else:\n            # This only happens when we are reading with only one worker (Default)\n            return pandas.read_json(fname, **kwargs)\n        if not pandas_df.columns.equals(columns):\n            raise NotImplementedError(\"Columns must be the same across all rows.\")\n        partition_columns = pandas_df.columns\n        return _split_result_for_readers(1, num_splits, pandas_df) + [\n            len(pandas_df),\n            pandas_df.dtypes,\n            partition_columns,\n        ]\n\n\nclass PandasParquetParser(PandasParser):\n    @staticmethod\n    def parse(fname, **kwargs):\n        import pyarrow.parquet as pq\n\n        num_splits = kwargs.pop(\"num_splits\", None)\n        columns = kwargs.get(\"columns\", None)\n        if num_splits is None:\n            return pandas.read_parquet(fname, **kwargs)\n        kwargs[\"use_pandas_metadata\"] = True\n        df = pq.read_table(fname, **kwargs).to_pandas()\n        if columns is not None:\n            df = df[columns]\n        # Append the length of the index here to build it externally\n        return _split_result_for_readers(0, num_splits, df) + [len(df.index), df.dtypes]\n\n\nclass PandasHDFParser(PandasParser):  # pragma: no cover\n    @staticmethod\n    def parse(fname, **kwargs):\n        kwargs[\"key\"] = kwargs.pop(\"_key\", None)\n        num_splits = kwargs.pop(\"num_splits\", None)\n        if num_splits is None:\n            return pandas.read_hdf(fname, **kwargs)\n        df = pandas.read_hdf(fname, **kwargs)\n        # Append the length of the index here to build it externally\n        return _split_result_for_readers(0, num_splits, df) + [len(df.index), df.dtypes]\n\n\nclass PandasFeatherParser(PandasParser):\n    @staticmethod\n    def parse(fname, **kwargs):\n        from pyarrow import feather\n\n        num_splits = kwargs.pop(\"num_splits\", None)\n        if num_splits is None:\n            return pandas.read_feather(fname, **kwargs)\n        df = feather.read_feather(fname, **kwargs)\n        # Append the length of the index here to build it externally\n        return _split_result_for_readers(0, num_splits, df) + [len(df.index), df.dtypes]\n\n\nclass PandasSQLParser(PandasParser):\n    @staticmethod\n    def parse(sql, con, index_col, **kwargs):\n        num_splits = kwargs.pop(\"num_splits\", None)\n        if num_splits is None:\n            return pandas.read_sql(sql, con, index_col=index_col, **kwargs)\n        df = pandas.read_sql(sql, con, index_col=index_col, **kwargs)\n        if index_col is None:\n            index = len(df)\n        else:\n            index = df.index\n        return _split_result_for_readers(1, num_splits, df) + [index, df.dtypes]\n", "path": "modin/backends/pandas/parsers.py"}], "after_files": [{"content": "import numpy as np\nimport pandas\nfrom pandas.core.dtypes.cast import find_common_type\nfrom pandas.core.dtypes.concat import union_categoricals\nfrom pandas.io.common import _infer_compression\nfrom modin.engines.base.io import FileReader\nfrom modin.data_management.utils import split_result_of_axis_func_pandas\nfrom modin.error_message import ErrorMessage\nfrom io import BytesIO\n\n\ndef _split_result_for_readers(axis, num_splits, df):  # pragma: no cover\n    \"\"\"Splits the DataFrame read into smaller DataFrames and handles all edge cases.\n\n    Args:\n        axis: Which axis to split over.\n        num_splits: The number of splits to create.\n        df: The DataFrame after it has been read.\n\n    Returns:\n        A list of pandas DataFrames.\n    \"\"\"\n    splits = split_result_of_axis_func_pandas(axis, num_splits, df)\n    if not isinstance(splits, list):\n        splits = [splits]\n    return splits\n\n\ndef find_common_type_cat(types):\n    if all(isinstance(t, pandas.CategoricalDtype) for t in types):\n        if all(t.ordered for t in types):\n            return pandas.CategoricalDtype(\n                np.sort(np.unique([c for t in types for c in t.categories])[0]),\n                ordered=True,\n            )\n        return union_categoricals(\n            [pandas.Categorical([], dtype=t) for t in types],\n            sort_categories=all(t.ordered for t in types),\n        ).dtype\n    else:\n        return find_common_type(types)\n\n\nclass PandasParser(object):\n    @classmethod\n    def get_dtypes(cls, dtypes_ids):\n        return (\n            pandas.concat(cls.materialize(dtypes_ids), axis=1)\n            .apply(lambda row: find_common_type_cat(row.values), axis=1)\n            .squeeze(axis=0)\n        )\n\n    @classmethod\n    def single_worker_read(cls, fname, **kwargs):\n        ErrorMessage.default_to_pandas(\"Parameters provided\")\n        # Use default args for everything\n        pandas_frame = cls.parse(fname, **kwargs)\n        if isinstance(pandas_frame, pandas.io.parsers.TextFileReader):\n            pd_read = pandas_frame.read\n            pandas_frame.read = lambda *args, **kwargs: cls.query_compiler_cls.from_pandas(\n                pd_read(*args, **kwargs), cls.frame_cls\n            )\n            return pandas_frame\n        return cls.query_compiler_cls.from_pandas(pandas_frame, cls.frame_cls)\n\n    infer_compression = _infer_compression\n\n\nclass PandasCSVParser(PandasParser):\n    @staticmethod\n    def parse(fname, **kwargs):\n        num_splits = kwargs.pop(\"num_splits\", None)\n        start = kwargs.pop(\"start\", None)\n        end = kwargs.pop(\"end\", None)\n        index_col = kwargs.get(\"index_col\", None)\n        if start is not None and end is not None:\n            # pop \"compression\" from kwargs because bio is uncompressed\n            bio = FileReader.file_open(fname, \"rb\", kwargs.pop(\"compression\", \"infer\"))\n            if kwargs.get(\"encoding\", None) is not None:\n                header = b\"\" + bio.readline()\n            else:\n                header = b\"\"\n            bio.seek(start)\n            to_read = header + bio.read(end - start)\n            bio.close()\n            pandas_df = pandas.read_csv(BytesIO(to_read), **kwargs)\n        else:\n            # This only happens when we are reading with only one worker (Default)\n            return pandas.read_csv(fname, **kwargs)\n        if index_col is not None:\n            index = pandas_df.index\n        else:\n            # The lengths will become the RangeIndex\n            index = len(pandas_df)\n        return _split_result_for_readers(1, num_splits, pandas_df) + [\n            index,\n            pandas_df.dtypes,\n        ]\n\n\nclass PandasJSONParser(PandasParser):\n    @staticmethod\n    def parse(fname, **kwargs):\n        num_splits = kwargs.pop(\"num_splits\", None)\n        start = kwargs.pop(\"start\", None)\n        end = kwargs.pop(\"end\", None)\n        if start is not None and end is not None:\n            # pop \"compression\" from kwargs because bio is uncompressed\n            bio = FileReader.file_open(fname, \"rb\", kwargs.pop(\"compression\", \"infer\"))\n            bio.seek(start)\n            to_read = b\"\" + bio.read(end - start)\n            bio.close()\n            columns = kwargs.pop(\"columns\")\n            pandas_df = pandas.read_json(BytesIO(to_read), **kwargs)\n        else:\n            # This only happens when we are reading with only one worker (Default)\n            return pandas.read_json(fname, **kwargs)\n        if not pandas_df.columns.equals(columns):\n            raise NotImplementedError(\"Columns must be the same across all rows.\")\n        partition_columns = pandas_df.columns\n        return _split_result_for_readers(1, num_splits, pandas_df) + [\n            len(pandas_df),\n            pandas_df.dtypes,\n            partition_columns,\n        ]\n\n\nclass PandasParquetParser(PandasParser):\n    @staticmethod\n    def parse(fname, **kwargs):\n        import pyarrow.parquet as pq\n\n        num_splits = kwargs.pop(\"num_splits\", None)\n        columns = kwargs.get(\"columns\", None)\n        if num_splits is None:\n            return pandas.read_parquet(fname, **kwargs)\n        kwargs[\"use_pandas_metadata\"] = True\n        df = pq.read_table(fname, **kwargs).to_pandas()\n        if columns is not None:\n            df = df[columns]\n        # Append the length of the index here to build it externally\n        return _split_result_for_readers(0, num_splits, df) + [len(df.index), df.dtypes]\n\n\nclass PandasHDFParser(PandasParser):  # pragma: no cover\n    @staticmethod\n    def parse(fname, **kwargs):\n        kwargs[\"key\"] = kwargs.pop(\"_key\", None)\n        num_splits = kwargs.pop(\"num_splits\", None)\n        if num_splits is None:\n            return pandas.read_hdf(fname, **kwargs)\n        df = pandas.read_hdf(fname, **kwargs)\n        # Append the length of the index here to build it externally\n        return _split_result_for_readers(0, num_splits, df) + [len(df.index), df.dtypes]\n\n\nclass PandasFeatherParser(PandasParser):\n    @staticmethod\n    def parse(fname, **kwargs):\n        from pyarrow import feather\n\n        num_splits = kwargs.pop(\"num_splits\", None)\n        if num_splits is None:\n            return pandas.read_feather(fname, **kwargs)\n        df = feather.read_feather(fname, **kwargs)\n        # Append the length of the index here to build it externally\n        return _split_result_for_readers(0, num_splits, df) + [len(df.index), df.dtypes]\n\n\nclass PandasSQLParser(PandasParser):\n    @staticmethod\n    def parse(sql, con, index_col, **kwargs):\n        num_splits = kwargs.pop(\"num_splits\", None)\n        if num_splits is None:\n            return pandas.read_sql(sql, con, index_col=index_col, **kwargs)\n        df = pandas.read_sql(sql, con, index_col=index_col, **kwargs)\n        if index_col is None:\n            index = len(df)\n        else:\n            index = df.index\n        return _split_result_for_readers(1, num_splits, df) + [index, df.dtypes]\n", "path": "modin/backends/pandas/parsers.py"}]}