Repository: modin-project/modin

Issue:
JSON dispatcher data file split correction
Originated from https://github.com/modin-project/modin/pull/2607#discussion_r571989125.

Discussion:

Patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index c01fdd63..c0c12d4b 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -65,7 +65,7 @@ jobs:
             modin/pandas/series_utils.py modin/pandas/utils.py \
             modin/pandas/base.py \
             modin/pandas/io.py modin/engines/base/io/io.py \
-            modin/engines/base/frame asv_bench/benchmarks/utils.py \
+            modin/engines/base/frame asv_bench/benchmarks/utils \
             asv_bench/benchmarks/__init__.py asv_bench/benchmarks/io/__init__.py \
             asv_bench/benchmarks/scalability/__init__.py \
             modin/engines/base/io/column_stores \
@@ -247,6 +247,7 @@ jobs:
       - run: python -m pytest modin/test/test_envvar_npartitions.py
       - run: python -m pytest -n 2 modin/test/test_partition_api.py
       - run: python -m pytest modin/test/test_utils.py
+      - run: python -m pytest asv_bench/test/test_utils.py
 
   test-defaults:
     needs: [lint-commit, lint-flake8, lint-black, test-api, test-headers]
diff --git a/asv_bench/benchmarks/benchmarks.py b/asv_bench/benchmarks/benchmarks.py
index a26f3076..0c81c2ae 100644
--- a/asv_bench/benchmarks/benchmarks.py
+++ b/asv_bench/benchmarks/benchmarks.py
@@ -31,14 +31,11 @@ from .utils import (
     random_columns,
     random_booleans,
     ASV_USE_IMPL,
-    ASV_DATASET_SIZE,
-    BINARY_OP_DATA_SIZE,
-    UNARY_OP_DATA_SIZE,
-    SERIES_DATA_SIZE,
     GROUPBY_NGROUPS,
     IMPL,
     execute,
     translator_groupby_ngroups,
+    get_benchmark_shapes,
 )
 
 
@@ -59,8 +56,8 @@ class BaseTimeGroupBy:
 class TimeGroupByMultiColumn(BaseTimeGroupBy):
     param_names = ["shape", "ngroups", "groupby_ncols"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
-        GROUPBY_NGROUPS[ASV_DATASET_SIZE],
+        get_benchmark_shapes("TimeGroupByMultiColumn"),
+        GROUPBY_NGROUPS,
         [6],
     ]
 
@@ -74,8 +71,8 @@ class TimeGroupByMultiColumn(BaseTimeGroupBy):
 class TimeGroupByDefaultAggregations(BaseTimeGroupBy):
     param_names = ["shape", "ngroups"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
-        GROUPBY_NGROUPS[ASV_DATASET_SIZE],
+        get_benchmark_shapes("TimeGroupByDefaultAggregations"),
+        GROUPBY_NGROUPS,
     ]
 
     def time_groupby_count(self, *args, **kwargs):
@@ -94,8 +91,8 @@ class TimeGroupByDefaultAggregations(BaseTimeGroupBy):
 class TimeGroupByDictionaryAggregation(BaseTimeGroupBy):
     param_names = ["shape", "ngroups", "operation_type"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
-        GROUPBY_NGROUPS[ASV_DATASET_SIZE],
+        get_benchmark_shapes("TimeGroupByDictionaryAggregation"),
+        GROUPBY_NGROUPS,
         ["reduction", "aggregation"],
     ]
     operations = {
@@ -118,7 +115,7 @@ class TimeGroupByDictionaryAggregation(BaseTimeGroupBy):
 class TimeJoin:
     param_names = ["shapes", "how", "sort"]
     params = [
-        BINARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("TimeJoin"),
         ["left", "inner"],
         [False],
     ]
@@ -139,7 +136,7 @@ class TimeJoin:
 class TimeMerge:
     param_names = ["shapes", "how", "sort"]
     params = [
-        BINARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("TimeMerge"),
         ["left", "inner"],
         [False],
     ]
@@ -164,7 +161,7 @@ class TimeMerge:
 class TimeConcat:
     param_names = ["shapes", "how", "axis"]
     params = [
-        BINARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("TimeConcat"),
         ["inner"],
         [0, 1],
     ]
@@ -184,7 +181,7 @@ class TimeConcat:
 class TimeAppend:
     param_names = ["shapes", "sort"]
     params = [
-        BINARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("TimeAppend"),
         [False, True],
     ]
 
@@ -205,7 +202,7 @@ class TimeAppend:
 class TimeBinaryOp:
     param_names = ["shapes", "binary_op", "axis"]
     params = [
-        BINARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("TimeBinaryOp"),
         ["mul"],
         [0, 1],
     ]
@@ -259,7 +256,7 @@ class BaseTimeSetItem:
 
 class TimeSetItem(BaseTimeSetItem):
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("TimeSetItem"),
         [1],
         ["zero", "middle", "last"],
         [True, False],
@@ -276,7 +273,7 @@ class TimeSetItem(BaseTimeSetItem):
 
 class TimeInsert(BaseTimeSetItem):
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("TimeInsert"),
         [1],
         ["zero", "middle", "last"],
         [True, False],
@@ -294,7 +291,7 @@ class TimeInsert(BaseTimeSetItem):
 class TimeArithmetic:
     param_names = ["shape", "axis"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("TimeArithmetic"),
         [0, 1],
     ]
 
@@ -320,7 +317,7 @@ class TimeArithmetic:
 class TimeSortValues:
     param_names = ["shape", "columns_number", "ascending_list"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("TimeSortValues"),
         [1, 2, 10, 100],
         [False, True],
     ]
@@ -341,7 +338,7 @@ class TimeSortValues:
 class TimeDrop:
     param_names = ["shape", "axis", "drop_ncols"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("TimeDrop"),
         [0, 1],
         [1, 0.8],
     ]
@@ -362,7 +359,7 @@ class TimeDrop:
 class TimeHead:
     param_names = ["shape", "head_count"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("TimeHead"),
         [5, 0.8],
     ]
 
@@ -382,7 +379,7 @@ class TimeFillnaSeries:
     param_names = ["value_type", "shape", "limit"]
     params = [
         ["scalar", "dict", "Series"],
-        SERIES_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("TimeFillnaSeries"),
         [None, 0.8],
     ]
 
@@ -415,7 +412,7 @@ class TimeFillnaDataFrame:
     param_names = ["value_type", "shape", "limit"]
     params = [
         ["scalar", "dict", "DataFrame", "Series"],
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("TimeFillnaDataFrame"),
         [None, 0.8],
     ]
 
@@ -471,8 +468,8 @@ class BaseTimeValueCounts:
 class TimeValueCountsFrame(BaseTimeValueCounts):
     param_names = ["shape", "ngroups", "subset"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
-        GROUPBY_NGROUPS[ASV_DATASET_SIZE],
+        get_benchmark_shapes("TimeValueCountsFrame"),
+        GROUPBY_NGROUPS,
         [2, 10],
     ]
 
@@ -483,8 +480,8 @@ class TimeValueCountsFrame(BaseTimeValueCounts):
 class TimeValueCountsSeries(BaseTimeValueCounts):
     param_names = ["shape", "ngroups", "bins"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
-        GROUPBY_NGROUPS[ASV_DATASET_SIZE],
+        get_benchmark_shapes("TimeValueCountsSeries"),
+        GROUPBY_NGROUPS,
         [None, 3],
     ]
 
@@ -499,7 +496,7 @@ class TimeValueCountsSeries(BaseTimeValueCounts):
 class TimeIndexing:
     param_names = ["shape", "indexer_type"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("TimeIndexing"),
         [
             "scalar",
             "bool",
@@ -531,7 +528,7 @@ class TimeIndexing:
 
 class TimeMultiIndexing:
     param_names = ["shape"]
-    params = [UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE]]
+    params = [get_benchmark_shapes("TimeMultiIndexing")]
 
     def setup(self, shape):
         df = generate_dataframe(ASV_USE_IMPL, "int", *shape, RAND_LOW, RAND_HIGH)
@@ -557,7 +554,11 @@ class TimeMultiIndexing:
 
 class TimeResetIndex:
     param_names = ["shape", "drop", "level"]
-    params = [UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE], [False, True], [None, "level_1"]]
+    params = [
+        get_benchmark_shapes("TimeResetIndex"),
+        [False, True],
+        [None, "level_1"],
+    ]
 
     def setup(self, shape, drop, level):
         self.df = generate_dataframe(ASV_USE_IMPL, "int", *shape, RAND_LOW, RAND_HIGH)
@@ -576,7 +577,7 @@ class TimeResetIndex:
 class TimeAstype:
     param_names = ["shape", "dtype", "astype_ncolumns"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("TimeAstype"),
         ["float64", "category"],
         ["one", "all"],
     ]
@@ -597,7 +598,7 @@ class TimeAstype:
 class TimeDescribe:
     param_names = ["shape"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("TimeDescribe"),
     ]
 
     def setup(self, shape):
@@ -610,7 +611,7 @@ class TimeDescribe:
 class TimeProperties:
     param_names = ["shape"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("TimeProperties"),
     ]
 
     def setup(self, shape):
diff --git a/asv_bench/benchmarks/io/csv.py b/asv_bench/benchmarks/io/csv.py
index a7421630..70a64e53 100644
--- a/asv_bench/benchmarks/io/csv.py
+++ b/asv_bench/benchmarks/io/csv.py
@@ -19,21 +19,20 @@ from ..utils import (
     RAND_LOW,
     RAND_HIGH,
     ASV_USE_IMPL,
-    ASV_DATASET_SIZE,
     ASV_USE_BACKEND,
-    UNARY_OP_DATA_SIZE,
     IMPL,
     execute,
     get_shape_id,
     prepare_io_data,
+    get_benchmark_shapes,
 )
 
 
 class BaseReadCsv:
-    # test data file can de created only once
+    # test data file should be created only once
     def setup_cache(self, test_filename="io_test_file"):
         test_filenames = prepare_io_data(
-            test_filename, "str_int", UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE]
+            test_filename, self.data_type, get_benchmark_shapes(self.__class__.__name__)
         )
         return test_filenames
 
@@ -45,15 +44,17 @@ class BaseReadCsv:
 
 
 class TimeReadCsvSkiprows(BaseReadCsv):
+    shapes = get_benchmark_shapes("TimeReadCsvSkiprows")
     skiprows_mapping = {
         "lambda_even_rows": lambda x: x % 2,
-        "range_uniform": np.arange(1, UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE][0][0] // 10),
-        "range_step2": np.arange(1, UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE][0][0], 2),
+        "range_uniform": np.arange(1, shapes[0][0] // 10),
+        "range_step2": np.arange(1, shapes[0][0], 2),
     }
+    data_type = "str_int"
 
     param_names = ["shape", "skiprows"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        shapes,
         [
             None,
             "lambda_even_rows",
@@ -75,17 +76,10 @@ class TimeReadCsvSkiprows(BaseReadCsv):
 
 
 class TimeReadCsvTrueFalseValues(BaseReadCsv):
-    param_names = ["shape"]
-    params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
-    ]
+    data_type = "true_false_int"
 
-    # test data file should be created only once
-    def setup_cache(self, test_filename="io_test_file"):
-        test_filenames = prepare_io_data(
-            test_filename, "true_false_int", UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE]
-        )
-        return test_filenames
+    param_names = ["shape"]
+    params = [get_benchmark_shapes("TimeReadCsvTrueFalseValues")]
 
     def time_true_false_values(self, test_filenames, shape):
         execute(
@@ -99,12 +93,13 @@ class TimeReadCsvTrueFalseValues(BaseReadCsv):
 
 
 class TimeReadCsvNamesDtype:
+    shapes = get_benchmark_shapes("TimeReadCsvNamesDtype")
     _dtypes_params = ["Int64", "Int64_Timestamp"]
     _timestamp_columns = ["col1", "col2"]
 
     param_names = ["shape", "names", "dtype"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        shapes,
         ["array-like"],
         _dtypes_params,
     ]
@@ -126,7 +121,7 @@ class TimeReadCsvNamesDtype:
     def setup_cache(self, test_filename="io_test_file_csv_names_dtype"):
         # filenames with a metadata of saved dataframes
         cache = {}
-        for shape in UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE]:
+        for shape in self.shapes:
             for dtype in self._dtypes_params:
                 df = generate_dataframe("pandas", "int", *shape, RAND_LOW, RAND_HIGH)
                 if dtype == "Int64_Timestamp":
diff --git a/asv_bench/benchmarks/omnisci/benchmarks.py b/asv_bench/benchmarks/omnisci/benchmarks.py
index 39f7e6f8..b3dfd205 100644
--- a/asv_bench/benchmarks/omnisci/benchmarks.py
+++ b/asv_bench/benchmarks/omnisci/benchmarks.py
@@ -16,8 +16,9 @@
 from ..utils import (
     generate_dataframe,
     gen_nan_data,
+    RAND_LOW,
+    RAND_HIGH,
     ASV_USE_IMPL,
-    ASV_DATASET_SIZE,
     GROUPBY_NGROUPS,
     IMPL,
     execute,
@@ -25,21 +26,14 @@ from ..utils import (
     random_columns,
     random_booleans,
     trigger_import,
-)
-
-from .utils import (
-    BINARY_OP_DATA_SIZE,
-    UNARY_OP_DATA_SIZE,
-    SERIES_DATA_SIZE,
-    RAND_LOW,
-    RAND_HIGH,
+    get_benchmark_shapes,
 )
 
 
 class TimeJoin:
     param_names = ["shape", "how"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("omnisci.TimeJoin"),
         ["left", "inner"],
     ]
 
@@ -56,7 +50,7 @@ class TimeJoin:
 class TimeMerge:
     param_names = ["shapes", "how"]
     params = [
-        BINARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("omnisci.TimeMerge"),
         ["left"],
     ]
 
@@ -80,9 +74,7 @@ class TimeMerge:
 
 class TimeAppend:
     param_names = ["shapes"]
-    params = [
-        BINARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
-    ]
+    params = [get_benchmark_shapes("omnisci.TimeAppend")]
 
     def setup(self, shapes):
         self.df1 = generate_dataframe(
@@ -100,7 +92,7 @@ class TimeAppend:
 class TimeBinaryOpDataFrame:
     param_names = ["shape", "binary_op"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("omnisci.TimeBinaryOpDataFrame"),
         ["mul"],
     ]
 
@@ -119,7 +111,7 @@ class TimeBinaryOpDataFrame:
 class TimeBinaryOpSeries:
     param_names = ["shape", "binary_op"]
     params = [
-        SERIES_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("omnisci.TimeBinaryOpSeries"),
         ["mul"],
     ]
 
@@ -136,9 +128,7 @@ class TimeBinaryOpSeries:
 
 class TimeArithmetic:
     param_names = ["shape"]
-    params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
-    ]
+    params = [get_benchmark_shapes("omnisci.TimeArithmetic")]
 
     def setup(self, shape):
         self.df = generate_dataframe(ASV_USE_IMPL, "int", *shape, RAND_LOW, RAND_HIGH)
@@ -163,7 +153,7 @@ class TimeArithmetic:
 class TimeSortValues:
     param_names = ["shape", "columns_number", "ascending_list"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("omnisci.TimeSortValues"),
         [1, 5],
         [False, True],
     ]
@@ -185,7 +175,7 @@ class TimeSortValues:
 class TimeDrop:
     param_names = ["shape", "drop_ncols"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("omnisci.TimeDrop"),
         [1, 0.8],
     ]
 
@@ -206,7 +196,7 @@ class TimeDrop:
 class TimeHead:
     param_names = ["shape", "head_count"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("omnisci.TimeHead"),
         [5, 0.8],
     ]
 
@@ -227,7 +217,7 @@ class TimeFillna:
     param_names = ["value_type", "shape", "limit"]
     params = [
         ["scalar", "dict"],
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("omnisci.TimeFillna"),
         [None],
     ]
 
@@ -257,8 +247,8 @@ class TimeFillna:
 class TimeValueCountsSeries:
     param_names = ["shape", "ngroups"]
     params = [
-        SERIES_DATA_SIZE[ASV_DATASET_SIZE],
-        GROUPBY_NGROUPS[ASV_DATASET_SIZE],
+        get_benchmark_shapes("omnisci.TimeValueCountsSeries"),
+        GROUPBY_NGROUPS,
     ]
 
     def setup(self, shape, ngroups):
@@ -282,7 +272,7 @@ class TimeValueCountsSeries:
 class TimeIndexing:
     param_names = ["shape", "indexer_type"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("omnisci.TimeIndexing"),
         [
             "scalar",
             "bool",
@@ -312,7 +302,11 @@ class TimeIndexing:
 
 class TimeResetIndex:
     param_names = ["shape", "drop", "level"]
-    params = [UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE], [False, True], [None, "level_1"]]
+    params = [
+        get_benchmark_shapes("omnisci.TimeResetIndex"),
+        [False, True],
+        [None, "level_1"],
+    ]
 
     def setup(self, shape, drop, level):
         pd = IMPL[ASV_USE_IMPL]
@@ -335,7 +329,7 @@ class TimeResetIndex:
 class TimeAstype:
     param_names = ["shape", "dtype", "astype_ncolumns"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("omnisci.TimeAstype"),
         ["float64"],
         ["one", "all"],
     ]
@@ -361,9 +355,7 @@ class TimeAstype:
 
 class TimeDescribe:
     param_names = ["shape"]
-    params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
-    ]
+    params = [get_benchmark_shapes("omnisci.TimeDescribe")]
 
     def setup(self, shape):
         self.df = generate_dataframe(ASV_USE_IMPL, "int", *shape, RAND_LOW, RAND_HIGH)
@@ -375,9 +367,7 @@ class TimeDescribe:
 
 class TimeProperties:
     param_names = ["shape"]
-    params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
-    ]
+    params = [get_benchmark_shapes("omnisci.TimeProperties")]
 
     def setup(self, shape):
         self.df = generate_dataframe(ASV_USE_IMPL, "int", *shape, RAND_LOW, RAND_HIGH)
@@ -411,8 +401,8 @@ class BaseTimeGroupBy:
 class TimeGroupByDefaultAggregations(BaseTimeGroupBy):
     param_names = ["shape", "ngroups"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
-        GROUPBY_NGROUPS[ASV_DATASET_SIZE],
+        get_benchmark_shapes("omnisci.TimeGroupByDefaultAggregations"),
+        GROUPBY_NGROUPS,
     ]
 
     def time_groupby_count(self, *args, **kwargs):
@@ -425,8 +415,8 @@ class TimeGroupByDefaultAggregations(BaseTimeGroupBy):
 class TimeGroupByMultiColumn(BaseTimeGroupBy):
     param_names = ["shape", "ngroups", "groupby_ncols"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
-        GROUPBY_NGROUPS[ASV_DATASET_SIZE],
+        get_benchmark_shapes("omnisci.TimeGroupByMultiColumn"),
+        GROUPBY_NGROUPS,
         [6],
     ]
 
diff --git a/asv_bench/benchmarks/omnisci/io.py b/asv_bench/benchmarks/omnisci/io.py
index 713ecf27..300e02c8 100644
--- a/asv_bench/benchmarks/omnisci/io.py
+++ b/asv_bench/benchmarks/omnisci/io.py
@@ -20,27 +20,24 @@ from ..utils import (
     RAND_LOW,
     RAND_HIGH,
     ASV_USE_IMPL,
-    ASV_DATASET_SIZE,
     IMPL,
     get_shape_id,
     trigger_import,
+    get_benchmark_shapes,
 )
 
-from .utils import UNARY_OP_DATA_SIZE
-
 from ..io.csv import TimeReadCsvTrueFalseValues  # noqa: F401
 
 
 class TimeReadCsvNames:
+    shapes = get_benchmark_shapes("omnisci.TimeReadCsvNames")
     param_names = ["shape"]
-    params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
-    ]
+    params = [shapes]
 
     def setup_cache(self, test_filename="io_test_file_csv_names"):
         # filenames with a metadata of saved dataframes
         cache = {}
-        for shape in UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE]:
+        for shape in self.shapes:
             df = generate_dataframe("pandas", "int", *shape, RAND_LOW, RAND_HIGH)
             file_id = get_shape_id(shape)
             cache[file_id] = (
diff --git a/asv_bench/benchmarks/omnisci/utils.py b/asv_bench/benchmarks/omnisci/utils.py
index 12a5d823..83bc1717 100644
--- a/asv_bench/benchmarks/omnisci/utils.py
+++ b/asv_bench/benchmarks/omnisci/utils.py
@@ -12,33 +12,3 @@
 # governing permissions and limitations under the License.
 
 """The module contains the functionality that is used when benchmarking Modin commits on OmniSci backend."""
-
-RAND_LOW = 0
-RAND_HIGH = 1_000_000_000
-
-BINARY_OP_DATA_SIZE = {
-    "big": [
-        ((500_000, 20), (1_000_000, 10)),
-    ],
-    "small": [
-        ((10_000, 20), (25_000, 10)),
-    ],
-}
-
-UNARY_OP_DATA_SIZE = {
-    "big": [
-        (1_000_000, 10),
-    ],
-    "small": [
-        (10_000, 10),
-    ],
-}
-
-SERIES_DATA_SIZE = {
-    "big": [
-        (10_000_000, 1),
-    ],
-    "small": [
-        (100_000, 1),
-    ],
-}
diff --git a/asv_bench/benchmarks/scalability/scalability_benchmarks.py b/asv_bench/benchmarks/scalability/scalability_benchmarks.py
index 7c3e0e62..b0fabf68 100644
--- a/asv_bench/benchmarks/scalability/scalability_benchmarks.py
+++ b/asv_bench/benchmarks/scalability/scalability_benchmarks.py
@@ -26,16 +26,15 @@ from ..utils import (
     generate_dataframe,
     RAND_LOW,
     RAND_HIGH,
-    ASV_DATASET_SIZE,
-    UNARY_OP_DATA_SIZE,
     execute,
+    get_benchmark_shapes,
 )
 
 
 class TimeFromPandas:
     param_names = ["shape", "cpus"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("TimeFromPandas"),
         [4, 16, 32],
     ]
 
@@ -54,7 +53,7 @@ class TimeFromPandas:
 class TimeToPandas:
     param_names = ["shape", "cpus"]
     params = [
-        UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE],
+        get_benchmark_shapes("TimeToPandas"),
         [4, 16, 32],
     ]
 
diff --git a/asv_bench/benchmarks/utils/__init__.py b/asv_bench/benchmarks/utils/__init__.py
new file mode 100644
index 00000000..40b9072a
--- /dev/null
+++ b/asv_bench/benchmarks/utils/__init__.py
@@ -0,0 +1,55 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+"""Modin benchmarks utils."""
+
+from .compatibility import (
+    ASV_USE_IMPL,
+    ASV_USE_BACKEND,
+)
+from .data_shapes import RAND_LOW, RAND_HIGH, GROUPBY_NGROUPS, get_benchmark_shapes
+from .common import (
+    IMPL,
+    execute,
+    get_shape_id,
+    gen_data,
+    gen_nan_data,
+    generate_dataframe,
+    prepare_io_data,
+    random_string,
+    random_columns,
+    random_booleans,
+    translator_groupby_ngroups,
+    trigger_import,
+)
+
+__all__ = [
+    "ASV_USE_IMPL",
+    "ASV_USE_BACKEND",
+    "RAND_LOW",
+    "RAND_HIGH",
+    "GROUPBY_NGROUPS",
+    "get_benchmark_shapes",
+    "IMPL",
+    "execute",
+    "get_shape_id",
+    "gen_data",
+    "gen_nan_data",
+    "generate_dataframe",
+    "prepare_io_data",
+    "random_string",
+    "random_columns",
+    "random_booleans",
+    "translator_groupby_ngroups",
+    "trigger_import",
+]
diff --git a/asv_bench/benchmarks/utils.py b/asv_bench/benchmarks/utils/common.py
similarity index 88%
rename from asv_bench/benchmarks/utils.py
rename to asv_bench/benchmarks/utils/common.py
index 6f893458..6318f08b 100644
--- a/asv_bench/benchmarks/utils.py
+++ b/asv_bench/benchmarks/utils/common.py
@@ -19,7 +19,6 @@ benchmarking old commits, the utilities changed, which in turn can unexpectedly
 the performance results, hence some utility functions are duplicated here.
 """
 
-import os
 import logging
 import modin.pandas as pd
 import pandas
@@ -27,84 +26,15 @@ import numpy as np
 import uuid
 from typing import Optional, Union
 
-RAND_LOW = 0
-RAND_HIGH = 100
-random_state = np.random.RandomState(seed=42)
-
-
-try:
-    from modin.config import NPartitions
-
-    NPARTITIONS = NPartitions.get()
-except ImportError:
-    NPARTITIONS = pd.DEFAULT_NPARTITIONS
-
-try:
-    from modin.config import TestDatasetSize, AsvImplementation, Engine, Backend
-
-    ASV_USE_IMPL = AsvImplementation.get()
-    ASV_DATASET_SIZE = TestDatasetSize.get() or "Small"
-    ASV_USE_ENGINE = Engine.get()
-    ASV_USE_BACKEND = Backend.get()
-except ImportError:
-    # The same benchmarking code can be run for different versions of Modin, so in
-    # case of an error importing important variables, we'll just use predefined values
-    ASV_USE_IMPL = os.environ.get("MODIN_ASV_USE_IMPL", "modin")
-    ASV_DATASET_SIZE = os.environ.get("MODIN_TEST_DATASET_SIZE", "Small")
-    ASV_USE_ENGINE = os.environ.get("MODIN_ENGINE", "Ray")
-    ASV_USE_BACKEND = os.environ.get("MODIN_BACKEND", "Pandas")
-
-ASV_USE_IMPL = ASV_USE_IMPL.lower()
-ASV_DATASET_SIZE = ASV_DATASET_SIZE.lower()
-ASV_USE_ENGINE = ASV_USE_ENGINE.lower()
-ASV_USE_BACKEND = ASV_USE_BACKEND.lower()
-
-assert ASV_USE_IMPL in ("modin", "pandas")
-assert ASV_DATASET_SIZE in ("big", "small")
-assert ASV_USE_ENGINE in ("ray", "dask", "python")
-assert ASV_USE_BACKEND in ("pandas", "omnisci", "pyarrow")
-
-BINARY_OP_DATA_SIZE = {
-    "big": [
-        ((5000, 5000), (5000, 5000)),
-        # the case extremely inefficient
-        # ((20, 500_000), (10, 1_000_000)),
-        ((500_000, 20), (1_000_000, 10)),
-    ],
-    "small": [
-        ((250, 250), (250, 250)),
-        ((20, 10_000), (10, 25_000)),
-        ((10_000, 20), (25_000, 10)),
-    ],
-}
-
-UNARY_OP_DATA_SIZE = {
-    "big": [
-        (5000, 5000),
-        # the case extremely inefficient
-        # (10, 1_000_000),
-        (1_000_000, 10),
-    ],
-    "small": [
-        (250, 250),
-        (10, 10_000),
-        (10_000, 10),
-    ],
-}
-
-SERIES_DATA_SIZE = {
-    "big": [
-        (100_000, 1),
-    ],
-    "small": [
-        (10_000, 1),
-    ],
-}
+from .compatibility import (
+    ASV_USE_IMPL,
+    ASV_DATASET_SIZE,
+    ASV_USE_ENGINE,
+    ASV_USE_BACKEND,
+)
+from .data_shapes import RAND_LOW, RAND_HIGH
 
-GROUPBY_NGROUPS = {
-    "big": [100, "huge_amount_groups"],
-    "small": [5],
-}
+random_state = np.random.RandomState(seed=42)
 
 IMPL = {
     "modin": pd,
diff --git a/asv_bench/benchmarks/utils/compatibility.py b/asv_bench/benchmarks/utils/compatibility.py
new file mode 100644
index 00000000..7b8d7fbe
--- /dev/null
+++ b/asv_bench/benchmarks/utils/compatibility.py
@@ -0,0 +1,49 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+"""Compatibility layer for parameters used by ASV."""
+
+import os
+import modin.pandas as pd
+
+try:
+    from modin.config import NPartitions
+
+    NPARTITIONS = NPartitions.get()
+except ImportError:
+    NPARTITIONS = pd.DEFAULT_NPARTITIONS
+
+try:
+    from modin.config import TestDatasetSize, AsvImplementation, Engine, Backend
+
+    ASV_USE_IMPL = AsvImplementation.get()
+    ASV_DATASET_SIZE = TestDatasetSize.get() or "Small"
+    ASV_USE_ENGINE = Engine.get()
+    ASV_USE_BACKEND = Backend.get()
+except ImportError:
+    # The same benchmarking code can be run for different versions of Modin, so in
+    # case of an error importing important variables, we'll just use predefined values
+    ASV_USE_IMPL = os.environ.get("MODIN_ASV_USE_IMPL", "modin")
+    ASV_DATASET_SIZE = os.environ.get("MODIN_TEST_DATASET_SIZE", "Small")
+    ASV_USE_ENGINE = os.environ.get("MODIN_ENGINE", "Ray")
+    ASV_USE_BACKEND = os.environ.get("MODIN_BACKEND", "Pandas")
+
+ASV_USE_IMPL = ASV_USE_IMPL.lower()
+ASV_DATASET_SIZE = ASV_DATASET_SIZE.lower()
+ASV_USE_ENGINE = ASV_USE_ENGINE.lower()
+ASV_USE_BACKEND = ASV_USE_BACKEND.lower()
+
+assert ASV_USE_IMPL in ("modin", "pandas")
+assert ASV_DATASET_SIZE in ("big", "small")
+assert ASV_USE_ENGINE in ("ray", "dask", "python")
+assert ASV_USE_BACKEND in ("pandas", "omnisci", "pyarrow")
diff --git a/asv_bench/benchmarks/utils/data_shapes.py b/asv_bench/benchmarks/utils/data_shapes.py
new file mode 100644
index 00000000..a5987cf7
--- /dev/null
+++ b/asv_bench/benchmarks/utils/data_shapes.py
@@ -0,0 +1,222 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+"""Define data shapes."""
+
+import os
+import json
+
+from .compatibility import ASV_USE_BACKEND, ASV_DATASET_SIZE
+
+RAND_LOW = 0
+RAND_HIGH = 1_000_000_000 if ASV_USE_BACKEND == "omnisci" else 100
+
+BINARY_OP_DATA_SIZE = {
+    "big": [
+        [[5000, 5000], [5000, 5000]],
+        # the case extremely inefficient
+        # [[20, 500_000], [10, 1_000_000]],
+        [[500_000, 20], [1_000_000, 10]],
+    ],
+    "small": [
+        [[250, 250], [250, 250]],
+        [[20, 10_000], [10, 25_000]],
+        [[10_000, 20], [25_000, 10]],
+    ],
+}
+UNARY_OP_DATA_SIZE = {
+    "big": [
+        [5000, 5000],
+        # the case extremely inefficient
+        # [10, 1_000_000],
+        [1_000_000, 10],
+    ],
+    "small": [
+        [250, 250],
+        [10, 10_000],
+        [10_000, 10],
+    ],
+}
+SERIES_DATA_SIZE = {
+    "big": [
+        (100_000, 1),
+    ],
+    "small": [
+        (10_000, 1),
+    ],
+}
+
+
+OMNISCI_BINARY_OP_DATA_SIZE = {
+    "big": [
+        [[500_000, 20], [1_000_000, 10]],
+    ],
+    "small": [
+        [[10_000, 20], [25_000, 10]],
+    ],
+}
+OMNISCI_UNARY_OP_DATA_SIZE = {
+    "big": [
+        [1_000_000, 10],
+    ],
+    "small": [
+        [10_000, 10],
+    ],
+}
+OMNISCI_SERIES_DATA_SIZE = {
+    "big": [
+        [10_000_000, 1],
+    ],
+    "small": [
+        [100_000, 1],
+    ],
+}
+
+BINARY_SHAPES = (
+    OMNISCI_BINARY_OP_DATA_SIZE[ASV_DATASET_SIZE]
+    if ASV_USE_BACKEND == "omnisci"
+    else BINARY_OP_DATA_SIZE[ASV_DATASET_SIZE]
+)
+UNARY_SHAPES = (
+    OMNISCI_UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE]
+    if ASV_USE_BACKEND == "omnisci"
+    else UNARY_OP_DATA_SIZE[ASV_DATASET_SIZE]
+)
+SERIES_SHAPES = (
+    OMNISCI_SERIES_DATA_SIZE[ASV_DATASET_SIZE]
+    if ASV_USE_BACKEND == "omnisci"
+    else SERIES_DATA_SIZE[ASV_DATASET_SIZE]
+)
+
+DEFAULT_GROUPBY_NGROUPS = {
+    "big": [100, "huge_amount_groups"],
+    "small": [5],
+}
+GROUPBY_NGROUPS = DEFAULT_GROUPBY_NGROUPS[ASV_DATASET_SIZE]
+
+_DEFAULT_CONFIG_T = [
+    (
+        UNARY_SHAPES,
+        [
+            # Pandas backend benchmarks
+            "TimeGroupByMultiColumn",
+            "TimeGroupByDefaultAggregations",
+            "TimeGroupByDictionaryAggregation",
+            "TimeSetItem",
+            "TimeInsert",
+            "TimeArithmetic",
+            "TimeSortValues",
+            "TimeDrop",
+            "TimeHead",
+            "TimeFillna",
+            "TimeFillnaDataFrame",
+            "TimeValueCountsFrame",
+            "TimeValueCountsSeries",
+            "TimeIndexing",
+            "TimeMultiIndexing",
+            "TimeResetIndex",
+            "TimeAstype",
+            "TimeDescribe",
+            "TimeProperties",
+            # IO benchmarks
+            "TimeReadCsvSkiprows",
+            "TimeReadCsvTrueFalseValues",
+            "TimeReadCsvNamesDtype",
+            # Scalability benchmarks
+            "TimeFromPandas",
+            "TimeToPandas",
+            # OmniSci backend benchmarks
+            "omnisci.TimeJoin",
+            "omnisci.TimeBinaryOpDataFrame",
+            "omnisci.TimeArithmetic",
+            "omnisci.TimeSortValues",
+            "omnisci.TimeDrop",
+            "omnisci.TimeHead",
+            "omnisci.TimeFillna",
+            "omnisci.TimeIndexing",
+            "omnisci.TimeResetIndex",
+            "omnisci.TimeAstype",
+            "omnisci.TimeDescribe",
+            "omnisci.TimeProperties",
+            "omnisci.TimeGroupByDefaultAggregations",
+            "omnisci.TimeGroupByMultiColumn",
+            # OmniSci backend IO benchmarks
+            "omnisci.TimeReadCsvNames",
+        ],
+    ),
+    (
+        BINARY_SHAPES,
+        [
+            # Pandas backend benchmarks
+            "TimeJoin",
+            "TimeMerge",
+            "TimeConcat",
+            "TimeAppend",
+            "TimeBinaryOp",
+            # OmniSci backend benchmarks
+            "omnisci.TimeMerge",
+            "omnisci.TimeAppend",
+        ],
+    ),
+    (
+        SERIES_SHAPES,
+        [
+            # Pandas backend benchmarks
+            "TimeFillnaSeries",
+            # OmniSci backend benchmarks
+            "omnisci.TimeBinaryOpSeries",
+            "omnisci.TimeValueCountsSeries",
+        ],
+    ),
+]
+DEFAULT_CONFIG = {}
+for _shape, _names in _DEFAULT_CONFIG_T:
+    DEFAULT_CONFIG.update({_name: _shape for _name in _names})
+
+CONFIG_FROM_FILE = None
+
+
+def get_benchmark_shapes(bench_id: str):
+    """
+    Get custom benchmark shapes from a json file stored in MODIN_ASV_DATASIZE_CONFIG.
+
+    If `bench_id` benchmark is not found in the file, then the default value will
+    be used.
+
+    Parameters
+    ----------
+    bench_id : str
+        Unique benchmark identifier that is used to get shapes.
+
+    Returns
+    -------
+    list
+        Benchmark shapes.
+    """
+    global CONFIG_FROM_FILE
+    if not CONFIG_FROM_FILE:
+        try:
+            from modin.config import AsvDataSizeConfig
+
+            filename = AsvDataSizeConfig.get()
+        except ImportError:
+            filename = os.environ.get("MODIN_ASV_DATASIZE_CONFIG", None)
+        if filename:
+            # should be json
+            with open(filename) as _f:
+                CONFIG_FROM_FILE = json.load(_f)
+
+    if CONFIG_FROM_FILE and bench_id in CONFIG_FROM_FILE:
+        # example: "omnisci.TimeReadCsvNames": [[5555, 55], [3333, 33]]
+        return CONFIG_FROM_FILE[bench_id]
+    return DEFAULT_CONFIG[bench_id]
diff --git a/docs/flow/modin/experimental/xgboost.rst b/docs/flow/modin/experimental/xgboost.rst
index 658b81e5..765fb822 100644
--- a/docs/flow/modin/experimental/xgboost.rst
+++ b/docs/flow/modin/experimental/xgboost.rst
@@ -27,15 +27,13 @@ row partitions using the function :py:func:`~modin.distributed.dataframe.pandas.
   :members:
 
 :py:class:`~modin.experimental.xgboost.Booster` inherits original class ``xgboost.Booster`` and
-overrides method ``predict``. The main differences from original class interface for ``predict``
-method are: (1) changing the type of the `data` parameter to :py:class:`~modin.experimental.xgboost.DMatrix`, and (2)
-a new parameter `num_actors`, which specifies the number of actors to run for prediction.
+overrides method ``predict``. The difference from original class interface for ``predict``
+method is changing the type of the `data` parameter to :py:class:`~modin.experimental.xgboost.DMatrix`.
 
 .. autoclass:: modin.experimental.xgboost.Booster
     :members:
 
-:py:func:`~modin.experimental.xgboost.train` function (similar to ``predict`` method of
-:py:class:`~modin.experimental.xgboost.Booster`) has 2 differences from the original ``train`` function - (1) the
+:py:func:`~modin.experimental.xgboost.train` function has 2 differences from the original ``train`` function - (1) the
 data type of `dtrain` parameter is :py:class:`~modin.experimental.xgboost.DMatrix` and (2) a new parameter `num_actors`.
 
 .. autofunction:: modin.experimental.xgboost.train
@@ -44,10 +42,13 @@ Internal execution flow on Ray backend
 ''''''''''''''''''''''''''''''''''''''
 
 Internal functions :py:func:`~modin.experimental.xgboost.xgboost_ray._train` and
-:py:func:`~modin.experimental.xgboost.xgboost_ray._predict` work similar to xgboost. Approximate execution flow of
-internal implementation is the following:
+:py:func:`~modin.experimental.xgboost.xgboost_ray._predict` work similar to xgboost.
 
-1. The data is passed to :py:func:`~modin.experimental.xgboost.xgboost_ray._train`/:py:func:`~modin.experimental.xgboost.xgboost_ray._predict`
+
+Training
+********
+
+1. The data is passed to :py:func:`~modin.experimental.xgboost.xgboost_ray._train`
    function as a :py:class:`~modin.experimental.xgboost.DMatrix` object. Using an iterator of
    :py:class:`~modin.experimental.xgboost.DMatrix`, lists of ``ray.ObjectRef`` with row partitions of Modin DataFrame are exctracted. Example:
 
@@ -58,54 +59,46 @@ internal implementation is the following:
    ..
 
 2. On this step, the parameter `num_actors` is processed. The internal function :py:func:`~modin.experimental.xgboost.xgboost_ray._get_num_actors`
-   examines the value provided by the user and checks if it fits in the set of expected values
-   (int, "default_train", "default_predict").
-
-   * int - `num_actors` won't be changed. This value will be used.
-   * "default_train" - `num_actors` will be computed using condition that 1 actor should use maximum 2 CPUs.
-     This condition was chosen for using maximum parallel workers with multithreaded XGBoost training (2 threads
-     per worker will be used in this case).
-   * "default_predict" - `num_actors` will be computed using condition that 1 actor should use maximum 8 CPUs.
-     This condition was chosen to combine parallelization techniques: parallel actors and parallel threads.
-
-.. note:: `num_actors` parameter is made available for public functions :py:func:`~modin.experimental.xgboost.train` and ``predict``
-  method of :py:class:`~modin.experimental.xgboost.Booster` class to allow fine-tuning for obtaining the best
-  performance in specific use cases.
-
-3. ``ray.util.placement_group`` is created to reserve all available Ray resources. After that
-   :py:class:`~modin.experimental.xgboost.xgboost_ray.ModinXGBoostActor` objects are created using resources of the
-   previously created placement group.
-
-4. Data (`dtrain` for :py:func:`~modin.experimental.xgboost.xgboost_ray._train`, `data` for
-   :py:func:`~modin.experimental.xgboost.xgboost_ray._predict`) is split between actors evenly. The internal function
+   examines the value provided by the user. In case the value isn't provided, the `num_actors` will be computed using condition that 1 actor should use maximum 2 CPUs.
+   This condition was chosen for using maximum parallel workers with multithreaded XGBoost training (2 threads
+   per worker will be used in this case).
+
+.. note:: `num_actors` parameter is made available for public function :py:func:`~modin.experimental.xgboost.train` to allow
+  fine-tuning for obtaining the best performance in specific use cases.
+
+3. :py:class:`~modin.experimental.xgboost.xgboost_ray.ModinXGBoostActor` objects are created.
+
+4. Data `dtrain` is split between actors evenly. The internal function
    :py:func:`~modin.experimental.xgboost.xgboost_ray._split_data_across_actors` runs assigning row partitions to actors
    using internal function :py:func:`~modin.experimental.xgboost.xgboost_ray._assign_row_partitions_to_actors`.
-   This function creates a dictionary in the form: `{actor_rank: ([part_i0, part_i3, ..], [0, 3, ..]), ..}` for training,
-   `{actor_rank: [part_i0, part_i1, ..], ..}` for prediction.
+   This function creates a dictionary in the form: `{actor_rank: ([part_i0, part_i3, ..], [0, 3, ..]), ..}`.
 
 .. note:: :py:func:`~modin.experimental.xgboost.xgboost_ray._assign_row_partitions_to_actors` takes into account IP
   addresses of row partitions of `dtrain` data to minimize excess data transfer.
 
-1. For each :py:class:`~modin.experimental.xgboost.xgboost_ray.ModinXGBoostActor` the object methods ``set_train_data`` or ``set_predict_data`` are
-   called remotely. Those methods run by loading row partitions in actor according to the dictionary with partitions
+5. For each :py:class:`~modin.experimental.xgboost.xgboost_ray.ModinXGBoostActor` object ``set_train_data`` method is
+   called remotely. This method runs loading row partitions in actor according to the dictionary with partitions
    distribution from previous step. When data is passed to the actor, the row partitions are automatically materialized
    (``ray.ObjectRef`` -> ``pandas.DataFrame``).
 
-2. Methods ``train`` or ``predict`` of :py:class:`~modin.experimental.xgboost.xgboost_ray.ModinXGBoostActor` class object are called remotely.
+6. ``train`` method of :py:class:`~modin.experimental.xgboost.xgboost_ray.ModinXGBoostActor` class object is called remotely. This method
+   runs XGBoost training on local data of actor, connects to ``Rabit Tracker`` for sharing training state between
+   actors and returns dictionary with `booster` and `evaluation results`.
+
+7. At the final stage results from actors are returned. `booster` and `evals_result` are returned using ``ray.get``
+   function from remote actor.
+
+
+Prediction
+**********
 
-   * ``train``: method runs XGBoost training on local data of actor, connects to ``Rabit Tracker`` for sharing
-     training state between actors and returns dictionary with `booster` and `evaluation results`.
-   * ``predict``: method runs XGBoost prediction on local data of actor and returns IP address of actor and partial
-     prediction (``pandas.DataFrame``).
+1. The data is passed to :py:func:`~modin.experimental.xgboost.xgboost_ray._predict`
+   function as a :py:class:`~modin.experimental.xgboost.DMatrix` object.
 
-3. On the final stage results from actors are returned.
+2. :py:func:`~modin.experimental.xgboost.xgboost_ray._map_predict` function is applied remotely for each partition
+   of the data to make a partial prediction.
 
-   * ``train``: `booster` and `evals_result` is returned using ``ray.get`` function from remote actor. Placement
-     group which was created on the step 3 is removed to free resources. :py:class:`~modin.experimental.xgboost.Booster`
-     object is created and returned to user. 
-   * ``predict``: using ``ray.wait`` function we wait until all actors finish computing local predictions. Placement group
-     which was created on the step 3 is removed to free resources. ``modin.pandas.DataFrame`` is created from
-     ``ray.ObjectRef`` objects which is returned from actors. Modin DataFrame is returned to user.
+3. Result ``modin.pandas.DataFrame`` is created from ``ray.ObjectRef`` objects, obtained in the previous step.
 
 
 Internal API
@@ -119,3 +112,4 @@ Internal API
 .. autofunction:: modin.experimental.xgboost.xgboost_ray._predict
 .. autofunction:: modin.experimental.xgboost.xgboost_ray._get_num_actors
 .. autofunction:: modin.experimental.xgboost.xgboost_ray._split_data_across_actors
+.. autofunction:: modin.experimental.xgboost.xgboost_ray._map_predict
diff --git a/modin/config/envvars.py b/modin/config/envvars.py
index 516e6f60..17aad272 100644
--- a/modin/config/envvars.py
+++ b/modin/config/envvars.py
@@ -277,6 +277,13 @@ class AsvImplementation(EnvironmentVariable, type=ExactStr):
     default = "modin"
 
 
+class AsvDataSizeConfig(EnvironmentVariable, type=ExactStr):
+    """Allows to override default size of data (shapes)."""
+
+    varname = "MODIN_ASV_DATASIZE_CONFIG"
+    default = None
+
+
 class ProgressBar(EnvironmentVariable, type=bool):
     """Whether or not to show the progress bar."""
 
diff --git a/modin/engines/base/io/text/csv_dispatcher.py b/modin/engines/base/io/text/csv_dispatcher.py
index ad76e409..e0dfffdf 100644
--- a/modin/engines/base/io/text/csv_dispatcher.py
+++ b/modin/engines/base/io/text/csv_dispatcher.py
@@ -18,11 +18,13 @@ from modin.engines.base.io.text.text_file_dispatcher import (
     ColumnNamesTypes,
 )
 import pandas
+from pandas.core.dtypes.common import is_list_like
 from csv import QUOTE_NONE, Dialect
 import sys
 from typing import Union, Sequence, Callable, Dict, Tuple
 from pandas._typing import FilePathOrBuffer
 import pandas._libs.lib as lib
+import numpy as np
 
 from modin.config import NPartitions
 
@@ -55,6 +57,13 @@ class CSVDispatcher(TextFileDispatcher):
         -------
         new_query_compiler : BaseQueryCompiler
             Query compiler with imported data for further processing.
+
+        Notes
+        -----
+        `skiprows` is handled diferently based on the parameter type because of
+        performance reasons. If `skiprows` is integer - rows will be skipped during
+        data file partitioning and wouldn't be actually read. If `skiprows` is array
+        or callable - full data file will be read and only then rows will be dropped.
         """
         filepath_or_buffer_md = (
             cls.get_path(filepath_or_buffer)
@@ -64,17 +73,29 @@ class CSVDispatcher(TextFileDispatcher):
         compression_infered = cls.infer_compression(
             filepath_or_buffer, kwargs.get("compression")
         )
-        use_modin_impl = cls._read_csv_check_support(
-            filepath_or_buffer, kwargs, compression_infered
-        )
-        if not use_modin_impl:
-            return cls.single_worker_read(filepath_or_buffer, **kwargs)
-
         # Getting frequently used read_csv kwargs
         names = kwargs.get("names", lib.no_default)
         index_col = kwargs.get("index_col", None)
         encoding = kwargs.get("encoding", None)
         skiprows = kwargs.get("skiprows", None)
+        header = kwargs.get("header", "infer")
+        # 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
+        # by workers)
+        header_size = cls._define_header_size(
+            header,
+            names,
+        )
+        skiprows_md, pre_reading = cls._manage_skiprows_parameter(skiprows, header_size)
+        should_handle_skiprows = skiprows_md is not None and not isinstance(
+            skiprows_md, int
+        )
+
+        use_modin_impl = cls._read_csv_check_support(
+            filepath_or_buffer, kwargs, compression_infered
+        )
+        if not use_modin_impl:
+            return cls.single_worker_read(filepath_or_buffer, **kwargs)
 
         is_quoting = kwargs.get("quoting", "") != QUOTE_NONE
         quotechar = kwargs.get("quotechar", '"').encode(
@@ -90,13 +111,9 @@ class CSVDispatcher(TextFileDispatcher):
         # information will be obtained further from empty_df, so no need to handle it
         # by workers)
         header_size = cls._define_header_size(
-            kwargs.get("header", "infer"),
+            header,
             names,
         )
-        # Since skiprows can be only integer here (non-integer skiprows trigger fallback
-        # to pandas implementation for now) we can process header_size and skiprows
-        # simultaneously
-        skiprows = skiprows + header_size if skiprows else header_size
 
         pd_df_metadata = pandas.read_csv(
             filepath_or_buffer,
@@ -112,6 +129,7 @@ class CSVDispatcher(TextFileDispatcher):
             num_splits=num_splits,
             header_size=header_size if not pass_names else 0,
             names=names if not pass_names else column_names,
+            header=header if not pass_names else "infer",
             skipfooter=0,
             skiprows=None,
             nrows=None,
@@ -122,10 +140,12 @@ class CSVDispatcher(TextFileDispatcher):
             splits = cls.partitioned_file(
                 f,
                 num_partitions=NPartitions.get(),
-                nrows=kwargs.get("nrows", None),
-                skiprows=skiprows,
+                nrows=kwargs.get("nrows", None) if not should_handle_skiprows else None,
+                skiprows=skiprows_md if not should_handle_skiprows else None,
                 quotechar=quotechar,
                 is_quoting=is_quoting,
+                header_size=header_size,
+                pre_reading=pre_reading,
             )
 
         partition_ids, index_ids, dtypes_ids = cls._launch_tasks(
@@ -140,9 +160,12 @@ class CSVDispatcher(TextFileDispatcher):
             index_name=pd_df_metadata.index.name,
             column_widths=column_widths,
             column_names=column_names,
+            skiprows_md=skiprows_md if should_handle_skiprows else None,
+            header_size=header_size,
             squeeze=kwargs.get("squeeze", False),
             skipfooter=kwargs.get("skipfooter", None),
             parse_dates=kwargs.get("parse_dates", False),
+            nrows=kwargs.get("nrows", None) if should_handle_skiprows else None,
         )
         return new_query_compiler
 
@@ -190,10 +213,6 @@ class CSVDispatcher(TextFileDispatcher):
         if read_csv_kwargs.get("chunksize") is not None:
             return False
 
-        skiprows = read_csv_kwargs.get("skiprows")
-        if skiprows is not None and not isinstance(skiprows, int):
-            return False
-
         return True
 
     @classmethod
@@ -203,7 +222,7 @@ class CSVDispatcher(TextFileDispatcher):
         index_name: str,
     ) -> Tuple[IndexColType, list]:
         """
-        Compute the resulting DataFrame index and index lengths for each or partitions.
+        Compute the resulting DataFrame index and index lengths for each of partitions.
 
         Parameters
         ----------
@@ -241,6 +260,8 @@ class CSVDispatcher(TextFileDispatcher):
         index_name: str,
         column_widths: list,
         column_names: ColumnNamesTypes,
+        skiprows_md: Union[Sequence, callable, None] = None,
+        header_size: int = None,
         **kwargs,
     ):
         """
@@ -263,6 +284,10 @@ class CSVDispatcher(TextFileDispatcher):
             Number of columns in each partition.
         column_names : ColumnNamesTypes
             Array with columns names.
+        skiprows_md : array-like or callable, optional
+            Specifies rows to skip.
+        header_size : int, default: 0
+            Number of rows, that occupied by header.
         **kwargs : dict
             Parameters of `read_csv` function needed for postprocessing.
 
@@ -300,9 +325,84 @@ class CSVDispatcher(TextFileDispatcher):
             new_query_compiler = new_query_compiler.drop(
                 new_query_compiler.index[-skipfooter:]
             )
+        if skiprows_md is not None:
+            # skip rows that passed as array or callable
+            nrows = kwargs.get("nrows", None)
+            index_range = pandas.RangeIndex(len(new_query_compiler.index))
+            if is_list_like(skiprows_md):
+                new_query_compiler = new_query_compiler.view(
+                    index=index_range.delete(skiprows_md - header_size)
+                )
+            elif callable(skiprows_md):
+                mod_index = skiprows_md(index_range + header_size)
+                mod_index = (
+                    mod_index
+                    if isinstance(mod_index, np.ndarray)
+                    else mod_index.to_numpy("bool")
+                )
+                view_idx = index_range[~mod_index]
+                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)
+
+            if nrows:
+                new_query_compiler = new_query_compiler.view(
+                    pandas.RangeIndex(len(new_query_compiler.index))[:nrows]
+                )
         if kwargs.get("squeeze", False) and len(new_query_compiler.columns) == 1:
             return new_query_compiler[new_query_compiler.columns[0]]
         if index_col is None:
             new_query_compiler._modin_frame.synchronize_labels(axis=0)
 
         return new_query_compiler
+
+    @classmethod
+    def _manage_skiprows_parameter(
+        cls,
+        skiprows: Union[int, Sequence[int], Callable, None] = None,
+        header_size: int = 0,
+    ) -> Tuple[Union[int, Sequence, Callable], bool, int]:
+        """
+        Manage read_csv `skiprows` parameter.
+
+        Change `skiprows` parameter in the way Modin could more optimally
+        process it. If `skiprows` is an array, this array will be sorted and
+        then, if array is uniformly distributed, `skiprows` will be "squashed"
+        into integer value and `pre_reading` parameter will be set if needed
+        (in this case fastpath can be done).
+
+        Parameters
+        ----------
+        skiprows : int, array or callable, optional
+                Original skiprows parameter of read_csv function.
+        header_size : int, default: 0
+                Number of rows that are used by header.
+
+        Returns
+        -------
+        skiprows : int, array or callable
+                Updated skiprows parameter.
+        pre_reading : int
+                The number of rows that should be read before data file
+                splitting for further reading (the number of rows for
+                the first partition).
+        """
+        pre_reading = 0
+        uniform_skiprows = False
+        if is_list_like(skiprows):
+            skiprows = np.sort(skiprows)
+            if np.all(np.diff(skiprows) == 1):
+                uniform_skiprows = True
+
+        if uniform_skiprows:
+            pre_reading = max(0, skiprows[0] - header_size)
+            skiprows = len(skiprows)
+        elif skiprows is None:
+            skiprows = 0
+
+        return skiprows, pre_reading
diff --git a/modin/engines/base/io/text/json_dispatcher.py b/modin/engines/base/io/text/json_dispatcher.py
index 9bf13201..6f4e3a61 100644
--- a/modin/engines/base/io/text/json_dispatcher.py
+++ b/modin/engines/base/io/text/json_dispatcher.py
@@ -17,7 +17,6 @@ from modin.engines.base.io.text.text_file_dispatcher import TextFileDispatcher
 from io import BytesIO
 import pandas
 import numpy as np
-from csv import QUOTE_NONE
 
 from modin.config import NPartitions
 
@@ -73,7 +72,6 @@ class JSONDispatcher(TextFileDispatcher):
             splits = cls.partitioned_file(
                 f,
                 num_partitions=NPartitions.get(),
-                is_quoting=(args.get("quoting", "") != QUOTE_NONE),
             )
             for start, end in splits:
                 args.update({"start": start, "end": end})
diff --git a/modin/engines/base/io/text/text_file_dispatcher.py b/modin/engines/base/io/text/text_file_dispatcher.py
index 9de9428e..6fcc15b1 100644
--- a/modin/engines/base/io/text/text_file_dispatcher.py
+++ b/modin/engines/base/io/text/text_file_dispatcher.py
@@ -198,6 +198,8 @@ class TextFileDispatcher(FileDispatcher):
         skiprows: int = None,
         quotechar: bytes = b'"',
         is_quoting: bool = True,
+        header_size: int = 0,
+        pre_reading: int = 0,
     ):
         """
         Compute chunk sizes in bytes for every partition.
@@ -211,33 +213,59 @@ class TextFileDispatcher(FileDispatcher):
             If not specified grabs the value from `modin.config.NPartitions.get()`.
         nrows : int, optional
             Number of rows of file to read.
-        skiprows : array or callable, optional
+        skiprows : int, optional
             Specifies rows to skip.
         quotechar : bytes, default: b'"'
             Indicate quote in a file.
         is_quoting : bool, default: True
             Whether or not to consider quotes.
+        header_size : int, default: 0
+            Number of rows, that occupied by header.
+        pre_reading : int, default: 0
+            Number of rows between header and skipped rows, that should be read.
 
         Returns
         -------
         list
-            An array, where each element of array is a tuple of two ints:
-            beginning and the end offsets of the current chunk.
+            List with the next elements:
+                int : partition start read byte
+                int : partition end read byte
         """
+        read_rows_counter = 0
+        outside_quotes = True
+
         if num_partitions is None:
-            num_partitions = NPartitions.get()
+            num_partitions = NPartitions.get() - 1 if pre_reading else NPartitions.get()
 
         rows_skipper = cls.rows_skipper_builder(f, quotechar, is_quoting=is_quoting)
         result = []
 
         file_size = cls.file_size(f)
 
+        rows_skipper(header_size)
+
+        if pre_reading:
+            pre_reading_start = f.tell()
+            outside_quotes, read_rows = cls._read_rows(
+                f,
+                nrows=pre_reading,
+                quotechar=quotechar,
+                is_quoting=is_quoting,
+                outside_quotes=outside_quotes,
+            )
+            read_rows_counter += read_rows
+
+            result.append((pre_reading_start, f.tell()))
+
+            # add outside_quotes
+            if is_quoting and not outside_quotes:
+                warnings.warn("File has mismatched quotes")
+
         rows_skipper(skiprows)
 
         start = f.tell()
 
         if nrows:
-            read_rows_counter = 0
             partition_size = max(1, num_partitions, nrows // num_partitions)
             while f.tell() < file_size and read_rows_counter < nrows:
                 if read_rows_counter + partition_size > nrows:
diff --git a/modin/experimental/xgboost/xgboost.py b/modin/experimental/xgboost/xgboost.py
index 2505bad3..4d569ce5 100644
--- a/modin/experimental/xgboost/xgboost.py
+++ b/modin/experimental/xgboost/xgboost.py
@@ -54,6 +54,12 @@ class DMatrix(xgb.DMatrix):
         self.data = unwrap_partitions(data, axis=0, get_ip=True)
         self.label = unwrap_partitions(label, axis=0)
 
+        self.metadata = (
+            data.index,
+            data.columns,
+            data._query_compiler._modin_frame._row_lengths,
+        )
+
     def __iter__(self):
         """
         Return unwrapped `self.data` and `self.label`.
@@ -94,25 +100,20 @@ class Booster(xgb.Booster):
     def predict(
         self,
         data: DMatrix,
-        num_actors: Optional[int] = None,
         **kwargs,
     ):
         """
         Run distributed prediction with a trained booster.
 
-        During work it evenly distributes `data` between workers,
-        runs xgb.predict on each worker for subset of `data` and creates
-        Modin DataFrame with prediction results.
+        During execution it runs ``xgb.predict`` on each worker for subset of `data`
+        and creates Modin DataFrame with prediction results.
 
         Parameters
         ----------
         data : modin.experimental.xgboost.DMatrix
             Input data used for prediction.
-        num_actors : int, optional
-            Number of actors for prediction. If unspecified, this value will be
-            computed automatically.
         **kwargs : dict
-            Other parameters are the same as `xgboost.Booster.predict`.
+            Other parameters are the same as for ``xgboost.Booster.predict``.
 
         Returns
         -------
@@ -130,7 +131,7 @@ class Booster(xgb.Booster):
             data, DMatrix
         ), f"Type of `data` is {type(data)}, but expected {DMatrix}."
 
-        result = _predict(self.copy(), data, num_actors, **kwargs)
+        result = _predict(self.copy(), data, **kwargs)
         LOGGER.info("Prediction finished")
 
         return result
@@ -188,7 +189,7 @@ def train(
     assert isinstance(
         dtrain, DMatrix
     ), f"Type of `dtrain` is {type(dtrain)}, but expected {DMatrix}."
-    result = _train(dtrain, num_actors, params, *args, evals=evals, **kwargs)
+    result = _train(dtrain, params, *args, num_actors=num_actors, evals=evals, **kwargs)
     if isinstance(evals_result, dict):
         evals_result.update(result["history"])
 
diff --git a/modin/experimental/xgboost/xgboost_ray.py b/modin/experimental/xgboost/xgboost_ray.py
index 59edb293..6000f18f 100644
--- a/modin/experimental/xgboost/xgboost_ray.py
+++ b/modin/experimental/xgboost/xgboost_ray.py
@@ -24,6 +24,7 @@ from typing import Dict, List
 import math
 from collections import defaultdict
 
+import numpy as np
 import xgboost as xgb
 import ray
 from ray.services import get_node_ip_address
@@ -38,7 +39,7 @@ LOGGER = logging.getLogger("[modin.xgboost]")
 @ray.remote(num_cpus=0)
 class ModinXGBoostActor:
     """
-    Ray actor-class runs training/prediction on remote workers.
+    Ray actor-class runs training on the remote worker.
 
     Parameters
     ----------
@@ -87,17 +88,6 @@ class ModinXGBoostActor:
 
         return xgb.DMatrix(X, y, nthread=self._nthreads)
 
-    def get_ip(self):
-        """
-        Get IP address of actor.
-
-        Returns
-        -------
-        str
-            An IP address of node used by actor.
-        """
-        return get_node_ip_address()
-
     def set_train_data(self, *X_y, add_as_eval_method=None):
         """
         Set train data for actor.
@@ -116,25 +106,6 @@ class ModinXGBoostActor:
         if add_as_eval_method is not None:
             self._evals.append((self._dtrain, add_as_eval_method))
 
-    def set_predict_data(
-        self,
-        *X,
-    ):
-        """
-        Set prediction data for actor.
-
-        Parameters
-        ----------
-        *X : iterable
-            Sequence of ray.ObjectRef objects. When it is passed in actor, auto-materialization
-            of ray.ObjectRef -> pandas.DataFrame happens.
-        """
-        X = pandas.concat(X, axis=0)
-        self._dpredict = {
-            "dmatrix": xgb.DMatrix(X, nthread=self._nthreads),
-            "index": X.index,
-        }
-
     def add_eval_data(self, *X_y, eval_method):
         """
         Add evaluation data for actor.
@@ -196,36 +167,6 @@ class ModinXGBoostActor:
             LOGGER.info(f"Local training time: {time.time() - s} s")
             return {"booster": bst, "history": evals_result}
 
-    def predict(self, booster: xgb.Booster, **kwargs):
-        """
-        Run local XGBoost prediction.
-
-        Parameters
-        ----------
-        booster : xgboost.Booster
-            A trained booster.
-        **kwargs : dict
-            Other parameters for `xgboost.Booster.predict`.
-
-        Returns
-        -------
-        tuple
-            Pair of IP address of caller and pandas.DataFrame
-            with partial prediction result.
-        """
-        local_dpredict = self._dpredict
-        booster.set_param({"nthread": self._nthreads})
-
-        s = time.time()
-
-        predictions = pandas.DataFrame(
-            booster.predict(local_dpredict["dmatrix"], **kwargs),
-            index=local_dpredict["index"],
-        )
-        LOGGER.info(f"Local prediction time: {time.time() - s} s")
-
-        return get_node_ip_address(), predictions
-
 
 def _get_cluster_cpus():
     """
@@ -276,18 +217,15 @@ def _get_cpus_per_actor(num_actors):
     return cpus_per_actor
 
 
-def _get_num_actors(num_actors):
+def _get_num_actors(num_actors=None):
     """
-    Get number on actors to create.
-
-    In case `num_actors` is 'default_train' string,
-    integer number of actors will be computed  by condition
-    2 CPUs per 1 actor, in case 'default_predict' 8 CPUs per 1 actor.
+    Get number of actors to create.
 
     Parameters
     ----------
-    num_actors : int or {'default_train', 'default_predict'}
-        Number of actors to create or str to set defaults.
+    num_actors : int, optional
+        Desired number of actors. If is None, integer number of actors
+        will be computed by condition 2 CPUs per 1 actor.
 
     Returns
     -------
@@ -295,12 +233,9 @@ def _get_num_actors(num_actors):
         Number of actors to create.
     """
     min_cpus_per_node = _get_min_cpus_per_node()
-    if num_actors == "default_train":
+    if num_actors is None:
         num_actors_per_node = max(1, int(min_cpus_per_node // 2))
         return num_actors_per_node * len(ray.nodes())
-    elif num_actors == "default_predict":
-        num_actors_per_node = max(1, int(min_cpus_per_node // 8))
-        return num_actors_per_node * len(ray.nodes())
     elif isinstance(num_actors, int):
         assert (
             num_actors % len(ray.nodes()) == 0
@@ -344,7 +279,10 @@ def create_actors(num_actors):
 
 
 def _split_data_across_actors(
-    actors: List, set_func, X_parts, y_parts=None, is_predict=False
+    actors: List,
+    set_func,
+    X_parts,
+    y_parts,
 ):
     """
     Split row partitions of data between actors.
@@ -357,49 +295,35 @@ def _split_data_across_actors(
         The function for setting data in actor.
     X_parts : list
         Row partitions of X data.
-    y_parts : list, optional
+    y_parts : list
         Row partitions of y data.
-    is_predict : bool, default: False
-        Is split data for predict or not.
     """
     X_parts_by_actors = _assign_row_partitions_to_actors(
         actors,
         X_parts,
-        is_predict=is_predict,
     )
 
-    if y_parts is not None:
-        y_parts_by_actors = _assign_row_partitions_to_actors(
-            actors,
-            y_parts,
-            data_for_aligning=X_parts_by_actors,
-        )
+    y_parts_by_actors = _assign_row_partitions_to_actors(
+        actors,
+        y_parts,
+        data_for_aligning=X_parts_by_actors,
+    )
 
     for rank, (_, actor) in enumerate(actors):
-        X_parts = X_parts_by_actors[rank] if is_predict else X_parts_by_actors[rank][0]
-        if y_parts is None:
-            set_func(actor, *X_parts)
-        else:
-            y_parts = y_parts_by_actors[rank][0]
-            set_func(actor, *(X_parts + y_parts))
+        set_func(actor, *(X_parts_by_actors[rank][0] + y_parts_by_actors[rank][0]))
 
 
 def _assign_row_partitions_to_actors(
     actors: List,
     row_partitions,
     data_for_aligning=None,
-    is_predict=False,
 ):
     """
     Assign row_partitions to actors.
 
-    In case of `is_predict` == False, `row_partitions` will be
-    assigned to actors according to their IPs. If distribution isn't
-    even, partitions will be moved from actor with excess partitions
-    to actor with lack of them.
-
-    In case of `is_predict` == True, `row_partitions` will be
-    assigned evenly to actors in order.
+    `row_partitions` will be assigned to actors according to their IPs.
+    If distribution isn't even, partitions will be moved from actor
+    with excess partitions to actor with lack of them.
 
     Parameters
     ----------
@@ -410,8 +334,6 @@ def _assign_row_partitions_to_actors(
     data_for_aligning : dict, optional
         Data according to the order of which should be
         distributed `row_partitions`. Used to align y with X.
-    is_predict : bool, default: False
-        Is split data for predict or not.
 
     Returns
     -------
@@ -420,138 +342,115 @@ def _assign_row_partitions_to_actors(
         as {actor_rank: (partitions, order)}.
     """
     num_actors = len(actors)
-    if not is_predict:
-        if data_for_aligning is None:
-            parts_ips_ref, parts_ref = zip(*row_partitions)
-
-            # Group actors which are one the same ip
-            actor_ips = defaultdict(list)
-            for rank, (ip, _) in enumerate(actors):
-                actor_ips[ip].append(rank)
-
-            # Get distribution of parts between nodes ({ip:[(part, position),..],..})
-            init_parts_distribution = defaultdict(list)
-            for idx, (ip, part_ref) in enumerate(
-                zip(ray.get(list(parts_ips_ref)), parts_ref)
-            ):
-                init_parts_distribution[ip].append((part_ref, idx))
-
-            num_parts = len(parts_ref)
-            min_parts_per_actor = math.floor(num_parts / num_actors)
-            max_parts_per_actor = math.ceil(num_parts / num_actors)
-            num_actors_with_max_parts = num_parts % num_actors
-
-            row_partitions_by_actors = defaultdict(list)
-            # Fill actors without movement parts between ips
-            for actor_ip, ranks in actor_ips.items():
-                # Loop across actors which are placed on actor_ip
-                for rank in ranks:
-                    num_parts_on_ip = len(init_parts_distribution[actor_ip])
-
-                    # Check that have something to distribute on this ip
-                    if num_parts_on_ip == 0:
-                        break
-                    # Check that node with `actor_ip` has enough parts for minimal
-                    # filling actor with `rank`
-                    if num_parts_on_ip >= min_parts_per_actor:
-                        # Check that node has enough parts for max filling
-                        # actor with `rank`
-                        if (
-                            num_parts_on_ip >= max_parts_per_actor
-                            and num_actors_with_max_parts > 0
-                        ):
-                            pop_slice = slice(0, max_parts_per_actor)
-                            num_actors_with_max_parts -= 1
-                        else:
-                            pop_slice = slice(0, min_parts_per_actor)
-
-                        row_partitions_by_actors[rank].extend(
-                            init_parts_distribution[actor_ip][pop_slice]
-                        )
-                        # Delete parts which we already assign
-                        del init_parts_distribution[actor_ip][pop_slice]
-                    else:
-                        row_partitions_by_actors[rank].extend(
-                            init_parts_distribution[actor_ip]
-                        )
-                        init_parts_distribution[actor_ip] = []
-
-            # Remove empty IPs
-            for ip in list(init_parts_distribution):
-                if len(init_parts_distribution[ip]) == 0:
-                    init_parts_distribution.pop(ip)
-
-            # IP's aren't necessary now
-            init_parts_distribution = [
-                pair for pairs in init_parts_distribution.values() for pair in pairs
-            ]
-
-            # Fill the actors with extra parts (movements data between nodes)
-            for rank in range(len(actors)):
-                num_parts_on_rank = len(row_partitions_by_actors[rank])
-
-                if num_parts_on_rank == max_parts_per_actor or (
-                    num_parts_on_rank == min_parts_per_actor
-                    and num_actors_with_max_parts == 0
-                ):
-                    continue
-
-                if num_actors_with_max_parts > 0:
-                    pop_slice = slice(0, max_parts_per_actor - num_parts_on_rank)
-                    num_actors_with_max_parts -= 1
-                else:
-                    pop_slice = slice(0, min_parts_per_actor - num_parts_on_rank)
-
-                row_partitions_by_actors[rank].extend(
-                    init_parts_distribution[pop_slice]
-                )
-                del init_parts_distribution[pop_slice]
-
-            if len(init_parts_distribution) != 0:
-                raise RuntimeError(
-                    f"Not all partitions were ditributed between actors: {len(init_parts_distribution)} left."
-                )
-
-            row_parts_by_ranks = dict()
-            for rank, pairs_part_pos in dict(row_partitions_by_actors).items():
-                parts, order = zip(*pairs_part_pos)
-                row_parts_by_ranks[rank] = (list(parts), list(order))
-        else:
-            row_parts_by_ranks = {rank: ([], []) for rank in range(len(actors))}
-
-            for rank, (_, order_of_indexes) in data_for_aligning.items():
-                row_parts_by_ranks[rank][1].extend(order_of_indexes)
-                for row_idx in order_of_indexes:
-                    row_parts_by_ranks[rank][0].append(row_partitions[row_idx])
-    else:
-        row_parts_by_ranks = defaultdict(list)
-        _, parts_ref = zip(*row_partitions)
+    if data_for_aligning is None:
+        parts_ips_ref, parts_ref = zip(*row_partitions)
+
+        # Group actors which are one the same ip
+        actor_ips = defaultdict(list)
+        for rank, (ip, _) in enumerate(actors):
+            actor_ips[ip].append(rank)
+
+        # Get distribution of parts between nodes ({ip:[(part, position),..],..})
+        init_parts_distribution = defaultdict(list)
+        for idx, (ip, part_ref) in enumerate(
+            zip(ray.get(list(parts_ips_ref)), parts_ref)
+        ):
+            init_parts_distribution[ip].append((part_ref, idx))
 
         num_parts = len(parts_ref)
         min_parts_per_actor = math.floor(num_parts / num_actors)
         max_parts_per_actor = math.ceil(num_parts / num_actors)
         num_actors_with_max_parts = num_parts % num_actors
 
-        start_idx = 0
-        for rank, actor in enumerate(actors):
+        row_partitions_by_actors = defaultdict(list)
+        # Fill actors without movement parts between ips
+        for actor_ip, ranks in actor_ips.items():
+            # Loop across actors which are placed on actor_ip
+            for rank in ranks:
+                num_parts_on_ip = len(init_parts_distribution[actor_ip])
+
+                # Check that have something to distribute on this ip
+                if num_parts_on_ip == 0:
+                    break
+                # Check that node with `actor_ip` has enough parts for minimal
+                # filling actor with `rank`
+                if num_parts_on_ip >= min_parts_per_actor:
+                    # Check that node has enough parts for max filling
+                    # actor with `rank`
+                    if (
+                        num_parts_on_ip >= max_parts_per_actor
+                        and num_actors_with_max_parts > 0
+                    ):
+                        pop_slice = slice(0, max_parts_per_actor)
+                        num_actors_with_max_parts -= 1
+                    else:
+                        pop_slice = slice(0, min_parts_per_actor)
+
+                    row_partitions_by_actors[rank].extend(
+                        init_parts_distribution[actor_ip][pop_slice]
+                    )
+                    # Delete parts which we already assign
+                    del init_parts_distribution[actor_ip][pop_slice]
+                else:
+                    row_partitions_by_actors[rank].extend(
+                        init_parts_distribution[actor_ip]
+                    )
+                    init_parts_distribution[actor_ip] = []
+
+        # Remove empty IPs
+        for ip in list(init_parts_distribution):
+            if len(init_parts_distribution[ip]) == 0:
+                init_parts_distribution.pop(ip)
+
+        # IP's aren't necessary now
+        init_parts_distribution = [
+            pair for pairs in init_parts_distribution.values() for pair in pairs
+        ]
+
+        # Fill the actors with extra parts (movements data between nodes)
+        for rank in range(len(actors)):
+            num_parts_on_rank = len(row_partitions_by_actors[rank])
+
+            if num_parts_on_rank == max_parts_per_actor or (
+                num_parts_on_rank == min_parts_per_actor
+                and num_actors_with_max_parts == 0
+            ):
+                continue
+
             if num_actors_with_max_parts > 0:
-                num_actor_parts = max_parts_per_actor
+                pop_slice = slice(0, max_parts_per_actor - num_parts_on_rank)
                 num_actors_with_max_parts -= 1
             else:
-                num_actor_parts = min_parts_per_actor
+                pop_slice = slice(0, min_parts_per_actor - num_parts_on_rank)
+
+            row_partitions_by_actors[rank].extend(init_parts_distribution[pop_slice])
+            del init_parts_distribution[pop_slice]
+
+        if len(init_parts_distribution) != 0:
+            raise RuntimeError(
+                f"Not all partitions were ditributed between actors: {len(init_parts_distribution)} left."
+            )
+
+        row_parts_by_ranks = dict()
+        for rank, pairs_part_pos in dict(row_partitions_by_actors).items():
+            parts, order = zip(*pairs_part_pos)
+            row_parts_by_ranks[rank] = (list(parts), list(order))
+    else:
+        row_parts_by_ranks = {rank: ([], []) for rank in range(len(actors))}
 
-            idx_slice = slice(start_idx, start_idx + num_actor_parts)
-            row_parts_by_ranks[rank].extend(parts_ref[idx_slice])
-            start_idx += num_actor_parts
+        for rank, (_, order_of_indexes) in data_for_aligning.items():
+            row_parts_by_ranks[rank][1].extend(order_of_indexes)
+            for row_idx in order_of_indexes:
+                row_parts_by_ranks[rank][0].append(row_partitions[row_idx])
 
     return row_parts_by_ranks
 
 
 def _train(
     dtrain,
-    num_actors,
     params: Dict,
     *args,
+    num_actors=None,
     evals=(),
     **kwargs,
 ):
@@ -568,13 +467,13 @@ def _train(
     ----------
     dtrain : modin.experimental.DMatrix
         Data to be trained against.
-    num_actors : int, optional
-        Number of actors for training. If unspecified, this value will be
-        computed automatically.
     params : dict
         Booster params.
     *args : iterable
         Other parameters for `xgboost.train`.
+    num_actors : int, optional
+        Number of actors for training. If unspecified, this value will be
+        computed automatically.
     evals : list of pairs (modin.experimental.xgboost.DMatrix, str), default: empty
         List of validation sets for which metrics will be evaluated during training.
         Validation metrics will help us track the performance of the model.
@@ -594,9 +493,7 @@ def _train(
 
     assert len(X_row_parts) == len(y_row_parts), "Unaligned train data"
 
-    num_actors = _get_num_actors(
-        num_actors if isinstance(num_actors, int) else "default_train"
-    )
+    num_actors = _get_num_actors(num_actors)
 
     if num_actors > len(X_row_parts):
         num_actors = len(X_row_parts)
@@ -618,7 +515,7 @@ def _train(
                     *X_y, eval_method=eval_method
                 ),
                 eval_X,
-                y_parts=eval_y,
+                eval_y,
             )
 
     # Split data across workers
@@ -628,7 +525,7 @@ def _train(
             *X_y, add_as_eval_method=add_as_eval_method
         ),
         X_row_parts,
-        y_parts=y_row_parts,
+        y_row_parts,
     )
     LOGGER.info(f"Data preparation time: {time.time() - s} s")
 
@@ -648,18 +545,46 @@ def _train(
         return result
 
 
+@ray.remote
+def _map_predict(booster, part, columns, **kwargs):
+    """
+    Run prediction on a remote worker.
+
+    Parameters
+    ----------
+    booster : xgboost.Booster or ray.ObjectRef
+        A trained booster.
+    part : pandas.DataFrame or ray.ObjectRef
+        Partition of full data used for local prediction.
+    columns : list or ray.ObjectRef
+        Columns for the result.
+    **kwargs : dict
+        Other parameters are the same as for ``xgboost.Booster.predict``.
+
+    Returns
+    -------
+    ray.ObjectRef
+        ``ray.ObjectRef`` with partial prediction.
+    """
+    dmatrix = xgb.DMatrix(part)
+    prediction = pandas.DataFrame(
+        booster.predict(dmatrix, **kwargs),
+        index=part.index,
+        columns=columns,
+    )
+    return prediction
+
+
 def _predict(
     booster,
     data,
-    num_actors,
     **kwargs,
 ):
     """
     Run distributed prediction with a trained booster on Ray backend.
 
-    During work it evenly distributes `data` between workers,
-    runs xgb.predict on each worker for subset of `data` and creates
-    Modin DataFrame with prediction results.
+    During execution it runs ``xgb.predict`` on each worker for subset of `data`
+    and creates Modin DataFrame with prediction results.
 
     Parameters
     ----------
@@ -667,11 +592,8 @@ def _predict(
         A trained booster.
     data : modin.experimental.xgboost.DMatrix
         Input data used for prediction.
-    num_actors : int, optional
-        Number of actors for prediction. If unspecified, this value will be
-        computed automatically.
-    **kwargs : dist
-        Other parameters are the same as `xgboost.Booster.predict`.
+    **kwargs : dict
+        Other parameters are the same as for ``xgboost.Booster.predict``.
 
     Returns
     -------
@@ -680,37 +602,39 @@ def _predict(
     """
     s = time.time()
 
-    X_row_parts, _ = data
-
-    num_actors = _get_num_actors(
-        num_actors if isinstance(num_actors, int) else "default_predict"
-    )
-
-    if num_actors > len(X_row_parts):
-        num_actors = len(X_row_parts)
-
-    # Create remote actors
-    actors = create_actors(num_actors)
+    # Get metadata from DMatrix
+    input_index, input_columns, row_lengths = data.metadata
 
-    # Split data across workers
-    _split_data_across_actors(
-        actors,
-        lambda actor, *X: actor.set_predict_data.remote(*X),
-        X_row_parts,
-        is_predict=True,
-    )
+    # Infer columns of result
+    def _get_num_columns(booster, n_features, **kwargs):
+        rng = np.random.RandomState(777)
+        test_data = rng.randn(1, n_features)
+        test_predictions = booster.predict(
+            xgb.DMatrix(test_data), validate_features=False, **kwargs
+        )
+        num_columns = (
+            test_predictions.shape[1] if len(test_predictions.shape) > 1 else 1
+        )
+        return num_columns
 
-    LOGGER.info(f"Data preparation time: {time.time() - s} s")
-    s = time.time()
+    result_num_columns = _get_num_columns(booster, len(input_columns), **kwargs)
+    new_columns = list(range(result_num_columns))
 
+    # Put common data in object store
     booster = ray.put(booster)
+    new_columns_ref = ray.put(new_columns)
 
-    predictions = [
-        tuple(actor.predict.options(num_returns=2).remote(booster, **kwargs))
-        for _, actor in actors
+    prediction_refs = [
+        _map_predict.remote(booster, part, new_columns_ref, **kwargs)
+        for _, part in data.data
     ]
-
-    result = from_partitions(predictions, 0)
+    predictions = from_partitions(
+        prediction_refs,
+        0,
+        index=input_index,
+        columns=new_columns,
+        row_lengths=row_lengths,
+        column_widths=[len(new_columns)],
+    )
     LOGGER.info(f"Prediction time: {time.time() - s} s")
-
-    return result
+    return predictions


Test patch:
diff --git a/asv_bench/test/__init__.py b/asv_bench/test/__init__.py
new file mode 100644
index 00000000..cae6413e
--- /dev/null
+++ b/asv_bench/test/__init__.py
@@ -0,0 +1,12 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
diff --git a/asv_bench/test/test_utils.py b/asv_bench/test/test_utils.py
new file mode 100644
index 00000000..95103ec4
--- /dev/null
+++ b/asv_bench/test/test_utils.py
@@ -0,0 +1,65 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import pytest
+from unittest.mock import patch, mock_open
+
+from benchmarks.utils import data_shapes
+from benchmarks.utils import get_benchmark_shapes
+
+from modin.config import AsvDataSizeConfig
+
+
+@pytest.mark.parametrize(
+    "asv_config_content, result",
+    [
+        (
+            '{"TimeJoin": [[[10, 10], [15, 15]], [[11, 11], [13, 13]]], \
+                "TimeGroupBy": [[11, 11], [13, 13]]}',
+            [
+                [
+                    # binary shapes
+                    [[10, 10], [15, 15]],
+                    [[11, 11], [13, 13]]
+                ],
+                [
+                    # unary shapes
+                    [11, 11],
+                    [13, 13]
+                ],
+            ],
+        ),
+    ],
+)
+@patch.object(data_shapes, "CONFIG_FROM_FILE", new=None)
+def test_get_benchmark_shapes(asv_config_content, result):
+    AsvDataSizeConfig.put("mock_filename")
+    with patch("builtins.open", mock_open(read_data=asv_config_content)):
+        assert result[0] == get_benchmark_shapes("TimeJoin")
+        assert result[1] == get_benchmark_shapes("TimeGroupBy")
+
+
+@pytest.mark.parametrize(
+    "asv_config_content, result",
+    [
+        (
+            '{"TimeJoin": [[[10, 10], [15, 15]]]',
+            [[100, 100]],
+        ),
+    ],
+)
+@patch.object(data_shapes, "CONFIG_FROM_FILE", new=None)
+def test_get_benchmark_shapes_default(asv_config_content, result):
+    AsvDataSizeConfig.put(None)
+    with patch.object(data_shapes, "DEFAULT_CONFIG", new={"TimeJoin": result}):
+        assert result == get_benchmark_shapes("TimeJoin")
diff --git a/modin/experimental/xgboost/test/test_xgboost.py b/modin/experimental/xgboost/test/test_xgboost.py
index 430232e6..a74a9399 100644
--- a/modin/experimental/xgboost/test/test_xgboost.py
+++ b/modin/experimental/xgboost/test/test_xgboost.py
@@ -104,7 +104,7 @@ def test_xgb_with_binary_classification_datasets(data, num_actors, modin_type_y)
             )
 
     predictions = bst.predict(xgb_dmatrix)
-    modin_predictions = modin_bst.predict(mxgb_dmatrix, num_actors=num_actors)
+    modin_predictions = modin_bst.predict(mxgb_dmatrix)
 
     preds = pd.DataFrame(predictions).apply(lambda x: round(x))
     modin_preds = modin_predictions.apply(lambda x: round(x))
@@ -197,7 +197,7 @@ def test_xgb_with_multiclass_classification_datasets(data, num_actors, modin_typ
         )
 
     predictions = bst.predict(xgb_dmatrix)
-    modin_predictions = modin_bst.predict(mxgb_dmatrix, num_actors=num_actors)
+    modin_predictions = modin_bst.predict(mxgb_dmatrix)
 
     array_preds = np.asarray([np.argmax(line) for line in predictions])
     modin_array_preds = np.asarray(
@@ -280,7 +280,7 @@ def test_xgb_with_regression_datasets(data, num_actors, modin_type_y):
             )
 
     predictions = bst.predict(train_xgb_dmatrix)
-    modin_predictions = modin_bst.predict(train_mxgb_dmatrix, num_actors=num_actors)
+    modin_predictions = modin_bst.predict(train_mxgb_dmatrix)
 
     val = mean_squared_error(y_train, predictions)
     modin_val = mean_squared_error(y_train, modin_predictions)
diff --git a/modin/pandas/test/test_io.py b/modin/pandas/test/test_io.py
index ee39d766..e9609270 100644
--- a/modin/pandas/test/test_io.py
+++ b/modin/pandas/test/test_io.py
@@ -16,6 +16,7 @@ import numpy as np
 import pandas
 from pandas.errors import ParserWarning
 import pandas._libs.lib as lib
+from pandas.core.dtypes.common import is_list_like
 from collections import OrderedDict
 from modin.config import TestDatasetSize, Engine, Backend, IsExperimental
 from modin.utils import to_pandas
@@ -339,55 +340,115 @@ class TestCsv:
             skipfooter=skipfooter,
         )
 
-    @pytest.mark.parametrize("true_values", [["Yes"], ["Yes", "true"], None])
-    @pytest.mark.parametrize("false_values", [["No"], ["No", "false"], None])
-    @pytest.mark.parametrize("skiprows", [2, lambda x: x % 2])
-    @pytest.mark.parametrize("skipfooter", [0, 10])
+    @pytest.mark.parametrize("header", ["infer", None, 0])
+    @pytest.mark.parametrize(
+        "skiprows",
+        [
+            2,
+            lambda x: x % 2,
+            lambda x: x > 25,
+            lambda x: x > 128,
+            np.arange(10, 50),
+            np.arange(10, 50, 2),
+        ],
+    )
     @pytest.mark.parametrize("nrows", [35, None])
-    @pytest.mark.parametrize("names", [["c1", "c2", "c3", "c4"], None])
+    @pytest.mark.parametrize(
+        "names",
+        [
+            [f"c{col_number}" for col_number in range(4)],
+            [f"c{col_number}" for col_number in range(6)],
+            None,
+        ],
+    )
+    @pytest.mark.parametrize("encoding", ["latin1", "windows-1251", None])
     def test_read_csv_parsing_2(
         self,
-        true_values,
-        false_values,
+        make_csv_file,
+        request,
+        header,
         skiprows,
-        skipfooter,
         nrows,
         names,
+        encoding,
     ):
-        xfail_case_1 = (
-            (false_values or true_values)
-            and Engine.get() != "Python"
-            and Backend.get() != "Omnisci"
-        )
-        xfail_case_2 = (
+        xfail_case = (
             Backend.get() == "Omnisci"
+            and header is not None
             and isinstance(skiprows, int)
             and names is None
-            and skipfooter == 0
             and nrows is None
-            and not true_values
-            and not false_values
         )
-        if xfail_case_1:
-            pytest.xfail("modin and pandas dataframes differs - issue #2446")
-        if xfail_case_2:
+        if xfail_case:
             pytest.xfail(
                 "read_csv fails because of duplicated columns names - issue #3080"
             )
+        if request.config.getoption(
+            "--simulate-cloud"
+        ).lower() != "off" and is_list_like(skiprows):
+            pytest.xfail(
+                reason="The reason of tests fail in `cloud` mode is unknown for now - issue #2340"
+            )
+        if encoding:
+            unique_filename = get_unique_filename()
+            make_csv_file(
+                filename=unique_filename,
+                encoding=encoding,
+            )
+        kwargs = {
+            "filepath_or_buffer": unique_filename
+            if encoding
+            else pytest.csvs_names["test_read_csv_regular"],
+            "header": header,
+            "skiprows": skiprows,
+            "nrows": nrows,
+            "names": names,
+            "encoding": encoding,
+        }
 
+        if Engine.get() != "Python":
+            df = pandas.read_csv(**dict(kwargs, nrows=1))
+            # in that case first partition will contain str
+            if df[df.columns[0]][df.index[0]] in ["c1", "col1", "c3", "col3"]:
+                pytest.xfail("read_csv incorrect output with float data - issue #2634")
         eval_io(
             fn_name="read_csv",
             check_exception_type=None,  # issue #2320
             raising_exceptions=None,
             check_kwargs_callable=not callable(skiprows),
             # read_csv kwargs
+            **kwargs,
+        )
+
+    @pytest.mark.parametrize("true_values", [["Yes"], ["Yes", "true"], None])
+    @pytest.mark.parametrize("false_values", [["No"], ["No", "false"], None])
+    @pytest.mark.parametrize("skipfooter", [0, 10])
+    @pytest.mark.parametrize("nrows", [35, None])
+    def test_read_csv_parsing_3(
+        self,
+        true_values,
+        false_values,
+        skipfooter,
+        nrows,
+    ):
+        xfail_case = (
+            (false_values or true_values)
+            and Engine.get() != "Python"
+            and Backend.get() != "Omnisci"
+        )
+        if xfail_case:
+            pytest.xfail("modin and pandas dataframes differs - issue #2446")
+
+        eval_io(
+            fn_name="read_csv",
+            check_exception_type=None,  # issue #2320
+            raising_exceptions=None,
+            # read_csv kwargs
             filepath_or_buffer=pytest.csvs_names["test_read_csv_yes_no"],
             true_values=true_values,
             false_values=false_values,
-            skiprows=skiprows,
             skipfooter=skipfooter,
             nrows=nrows,
-            names=names,
         )
 
     def test_read_csv_skipinitialspace(self):
@@ -891,12 +952,6 @@ class TestCsv:
                 StringIO(open(pytest.csvs_names["test_read_csv_regular"], "r").read())
             )
 
-        with pytest.warns(UserWarning):
-            pd.read_csv(
-                pytest.csvs_names["test_read_csv_regular"],
-                skiprows=lambda x: x in [0, 2],
-            )
-
     @pytest.mark.xfail(
         condition="config.getoption('--simulate-cloud').lower() != 'off'",
         reason="The reason of tests fail in `cloud` mode is unknown for now - issue #2340",