FIX-#3417: Fix read_csv with skiprows and header parameters

modin/core/io/text/fwf_dispatcher.py

@@ -15,6 +15,7 @@
 
 import pandas
 from pandas._typing import FilePathOrBuffer
+from typing import Union, Sequence
 
 from modin.core.io.text.text_file_dispatcher import TextFileDispatcher
 
@@ -29,6 +30,8 @@ def check_parameters_support(
         cls,
         filepath_or_buffer: FilePathOrBuffer,
         read_kwargs: dict,
+        skiprows_md: Union[Sequence, callable, int],
+        header_size: int,
     ):
         """
         Check support of parameters of `read_fwf` function.
@@ -39,6 +42,10 @@ def check_parameters_support(
             `filepath_or_buffer` parameter of `read_fwf` function.
         read_kwargs : dict
             Parameters of `read_fwf` function.
+        skiprows_md : int, array or callable
+            `skiprows` parameter modified for easier handling by Modin.
+        header_size : int
+            Number of rows that are used by header.
 
         Returns
         -------
@@ -49,4 +56,6 @@ def check_parameters_support(
             # If infer_nrows is a significant portion of the number of rows, pandas may be
             # faster.
             return False
-        return super().check_parameters_support(filepath_or_buffer, read_kwargs)
+        return super().check_parameters_support(
+            filepath_or_buffer, read_kwargs, skiprows_md, header_size
+        )

modin/core/io/text/text_file_dispatcher.py

@@ -35,6 +35,7 @@
 from modin.utils import _inherit_docstrings
 from modin.core.io.text.utils import CustomNewlineIterator
 from modin.config import NPartitions
+from modin.error_message import ErrorMessage
 
 ColumnNamesTypes = Tuple[Union[pandas.Index, pandas.MultiIndex, pandas.Int64Index]]
 IndexColType = Union[int, str, bool, Sequence[int], Sequence[str], None]
@@ -616,6 +617,8 @@ def check_parameters_support(
         cls,
         filepath_or_buffer: FilePathOrBuffer,
         read_kwargs: dict,
+        skiprows_md: Union[Sequence, callable, int],
+        header_size: int,
     ) -> bool:
         """
         Check support of only general parameters of `read_*` function.
@@ -626,12 +629,17 @@ def check_parameters_support(
             `filepath_or_buffer` parameter of `read_*` function.
         read_kwargs : dict
             Parameters of `read_*` function.
+        skiprows_md : int, array or callable
+            `skiprows` parameter modified for easier handling by Modin.
+        header_size : int
+            Number of rows that are used by header.
 
         Returns
         -------
         bool
             Whether passed parameters are supported or not.
         """
+        skiprows = read_kwargs.get("skiprows")
         if isinstance(filepath_or_buffer, str):
             if not cls.file_exists(filepath_or_buffer):
                 return False
@@ -641,6 +649,24 @@ def check_parameters_support(
         if read_kwargs["chunksize"] is not None:
             return False
 
+        skiprows_unsupported = (
+            is_list_like(skiprows_md) and skiprows_md[0] < header_size
+        ) or (
+            callable(skiprows)
+            and any(
+                map(
+                    bool,
+                    cls._get_skip_mask(pandas.RangeIndex(header_size), skiprows),
+                )
+            )
+        )
+        if skiprows_unsupported:
+            ErrorMessage.single_warning(
+                "Values of `header` and `skiprows` parameters have intersections. "
+                "This case doesn't supported by Modin, so pandas implementation will be used!"
+            )
+            return False
+
         return True
 
     @classmethod
@@ -759,14 +785,19 @@ def _manage_skiprows_parameter(
                 )
                 skiprows_partitioning = len(skiprows_md)
                 skiprows_md = 0
-            else:
+            elif skiprows_md[0] > header_size:
                 skiprows_md = skiprows_md - header_size
+
         elif callable(skiprows):
 
             def skiprows_func(x):
                 return skiprows(x + header_size)
 
             skiprows_md = skiprows_func
+        elif skiprows is not None:
+            raise TypeError(
+                f"Not acceptable type of `skiprows` parameter: {type(skiprows)}"
+            )
 
         return skiprows_md, pre_reading, skiprows_partitioning
 
@@ -894,10 +925,6 @@ def _get_new_qc(
                     skip_mask = skip_mask.to_numpy("bool")
                 view_idx = index_range[~skip_mask]
                 new_query_compiler = new_query_compiler.view(index=view_idx)
-            else:
-                raise TypeError(
-                    f"Not acceptable type of `skiprows` parameter: {type(skiprows_md)}"
-                )
 
             if not isinstance(new_query_compiler.index, pandas.MultiIndex):
                 new_query_compiler = new_query_compiler.reset_index(drop=True)
@@ -964,6 +991,8 @@ def _read(cls, filepath_or_buffer: FilePathOrBuffer, **kwargs):
         use_modin_impl = cls.check_parameters_support(
             filepath_or_buffer,
             kwargs,
+            skiprows_md,
+            header_size,
         )
         if not use_modin_impl:
             return cls.single_worker_read(

modin/pandas/test/test_io.py

@@ -1135,7 +1135,13 @@ def test_read_csv_empty_frame(self):
     @pytest.mark.parametrize(
         "skiprows",
         [
+            [x for x in range(10)],
+            [x + 5 for x in range(15)],
+            [x for x in range(10) if x % 2 == 0],
+            [x + 5 for x in range(15) if x % 2 == 0],
+            lambda x: x % 2,
             lambda x: x > 20,
+            lambda x: x < 20,
             lambda x: True,
             lambda x: x in [10, 20],
             pytest.param(
@@ -1147,13 +1153,16 @@ def test_read_csv_empty_frame(self):
             ),
         ],
     )
-    def test_read_csv_skiprows_corner_cases(self, skiprows):
+    @pytest.mark.parametrize("header", ["infer", None, 0, 1, 150])
+    def test_read_csv_skiprows_corner_cases(self, skiprows, header):
         eval_io(
             fn_name="read_csv",
             check_kwargs_callable=not callable(skiprows),
             # read_csv kwargs
             filepath_or_buffer=pytest.csvs_names["test_read_csv_regular"],
             skiprows=skiprows,
+            header=header,
+            dtype="str",  # to avoid issues with heterogeneous data
         )