FIX-#7170: Don't use 'MinPartitionSize' configuration variable in rem…

…ote context

Signed-off-by: Anatoly Myachev <[email protected]>

modin/core/dataframe/pandas/partitioning/axis_partition.py

@@ -18,6 +18,7 @@
 import numpy as np
 import pandas
 
+from modin.config import MinPartitionSize
 from modin.core.dataframe.base.partitioning.axis_partition import (
     BaseDataframeAxisPartition,
 )
@@ -392,6 +393,7 @@ def deploy_axis_func(
         maintain_partitioning,
         *partitions,
         lengths=None,
+        min_block_size=None,
         manual_partition=False,
         return_generator=False,
     ):
@@ -473,10 +475,16 @@ def deploy_axis_func(
                     lengths = None
         if return_generator:
             return generate_result_of_axis_func_pandas(
-                axis, num_splits, result, lengths
+                axis,
+                num_splits,
+                result,
+                lengths,
+                min_block_size=min_block_size,
             )
         else:
-            return split_result_of_axis_func_pandas(axis, num_splits, result, lengths)
+            return split_result_of_axis_func_pandas(
+                axis, num_splits, result, lengths, min_block_size=min_block_size
+            )
 
     @classmethod
     def deploy_func_between_two_axis_partitions(

modin/core/execution/dask/implementations/pandas_on_dask/partitioning/virtual_partition.py

@@ -17,6 +17,7 @@
 from distributed import Future
 from distributed.utils import get_ip
 
+from modin.config import MinPartitionSize
 from modin.core.dataframe.pandas.partitioning.axis_partition import (
     PandasDataframeAxisPartition,
 )
@@ -161,6 +162,7 @@ def deploy_axis_func(
             f_kwargs={
                 "lengths": lengths,
                 "manual_partition": manual_partition,
+                "min_block_size": MinPartitionSize.get(),
             },
             num_returns=result_num_splits * (1 + cls._PARTITIONS_METADATA_LEN),
             pure=False,

modin/core/execution/ray/implementations/pandas_on_ray/partitioning/virtual_partition.py

@@ -17,6 +17,7 @@
 import ray
 from ray.util import get_node_ip_address
 
+from modin.config import MinPartitionSize
 from modin.core.dataframe.pandas.partitioning.axis_partition import (
     PandasDataframeAxisPartition,
 )
@@ -189,6 +190,7 @@ def deploy_axis_func(
             f_kwargs=f_kwargs,
             manual_partition=manual_partition,
             lengths=lengths,
+            min_block_size=MinPartitionSize.get(),
             return_generator=True,
         )
 

modin/core/execution/unidist/implementations/pandas_on_unidist/partitioning/virtual_partition.py

@@ -18,6 +18,7 @@
 import pandas
 import unidist
 
+from modin.config import MinPartitionSize
 from modin.core.dataframe.pandas.partitioning.axis_partition import (
     PandasDataframeAxisPartition,
 )
@@ -189,6 +190,7 @@ def deploy_axis_func(
             *partitions,
             manual_partition=manual_partition,
             lengths=lengths,
+            min_block_size=MinPartitionSize.get(),
         )
 
     @classmethod

modin/core/storage_formats/pandas/utils.py

@@ -125,6 +125,7 @@ def generate_result_of_axis_func_pandas(
             length_list = get_length_list(
                 result.shape[axis], num_splits, min_block_size
             )
+        # print(f"{length_list=}")
         # Inserting the first "zero" to properly compute cumsum indexing slices
         length_list = np.insert(length_list, obj=0, values=[0])
         sums = np.cumsum(length_list)