Repository: modin-project/modin
Pull request: REFACTOR-#2642: Refactor partition API
Signed-off-by: Igoshev, Yaroslav <yaroslav.igoshev@intel.com>

<!--
Thank you for your contribution!
Please review the contributing docs: https://modin.readthedocs.io/en/latest/CONTRIBUTING.html
if you have questions about contributing.
-->

## What do these changes do?

<!-- Please give a short brief about these changes. -->

- [x] commit message follows format outlined [here](https://modin.readthedocs.io/en/latest/CONTRIBUTING.html)
- [x] passes `flake8 modin`
- [x] passes `black --check modin`
- [x] signed commit with `git commit -s` <!-- you can amend your commit with a signature via `git commit -amend -s` -->
- [x] Resolves #2642  <!-- issue must be created for each patch -->
- [x] tests passing


Problem statement:
Refactor partition API
### Describe the problem
<!-- Describe the problem clearly here. -->

For Modin on Ray engine with pandas backend IPs of the remote partitions may not match actual locations if the partitions are lower than 100 kB. Ray saves such objects (<= 100 kB, by default) in in-process store of the calling process. We can't get IPs for such objects with saving good performance. So, we need to add a doc page for this and perform some refactoring.

Hints from issue discussion:

Commit message: REFACTOR-#2642: Apply comment; Use apply instead of new func

Signed-off-by: Igoshev, Yaroslav <yaroslav.igoshev@intel.com>

Patch to review:
diff --git a/modin/config/envvars.py b/modin/config/envvars.py
index 245619273ef..45dce382711 100644
--- a/modin/config/envvars.py
+++ b/modin/config/envvars.py
@@ -155,6 +155,14 @@ def _get_default(cls):
         return CpuCount.get()
 
 
+class RayPartitionThreshold(EnvironmentVariable, type=int):
+    """
+    What objects can be saved in in-process store (in bytes)
+    """
+
+    varname = "MODIN_ON_RAY_PARTITION_THRESHOLD"
+
+
 class RayPlasmaDir(EnvironmentVariable, type=ExactStr):
     """
     Path to Plasma storage for Ray
diff --git a/modin/distributed/dataframe/pandas/partitions.py b/modin/distributed/dataframe/pandas/partitions.py
index e71522ed22f..07ea4762bd7 100644
--- a/modin/distributed/dataframe/pandas/partitions.py
+++ b/modin/distributed/dataframe/pandas/partitions.py
@@ -13,13 +13,13 @@
 
 import numpy as np
 
-from modin.pandas.dataframe import DataFrame
 from modin.backends.pandas.query_compiler import PandasQueryCompiler
+from modin.pandas.dataframe import DataFrame
 
 
-def unwrap_partitions(api_layer_object, axis=None, bind_ip=False):
+def unwrap_partitions(api_layer_object, axis=None, get_ip=False):
     """
-    Unwrap partitions of the `api_layer_object`.
+    Unwrap partitions of the ``api_layer_object``.
 
     Parameters
     ----------
@@ -27,20 +27,21 @@ def unwrap_partitions(api_layer_object, axis=None, bind_ip=False):
         The API layer object.
     axis : None, 0 or 1. Default is None
         The axis to unwrap partitions for (0 - row partitions, 1 - column partitions).
-        If axis is None, all the partitions of the API layer object are unwrapped.
-    bind_ip : boolean. Default is False
-        Whether to bind node ip address to each partition or not.
+        If ``axis=None``, all the partitions of the API layer object are unwrapped.
+    get_ip : boolean. Default is False
+        Whether to get node ip address to each partition or not.
 
     Returns
     -------
     list
-        A list of Ray.ObjectRef/Dask.Future to partitions of the `api_layer_object`
+        A list of Ray.ObjectRef/Dask.Future to partitions of the ``api_layer_object``
         if Ray/Dask is used as an engine.
 
     Notes
     -----
-    In case bind_ip=True, a list containing tuples of Ray.ObjectRef/Dask.Future to node ip addresses
-    and partitions of the `api_layer_object`, respectively, is returned if Ray/Dask is used as an engine.
+    If ``get_ip=True``, a list of tuples of node ip addresses and Ray.ObjectRef/Dask.Future to
+    partitions of the ``api_layer_object``, respectively, is returned if Ray/Dask is used as an engine
+    (i.e. [(str, Ray.ObjectRef/Dask.Future), ...]).
     """
     if not hasattr(api_layer_object, "_query_compiler"):
         raise ValueError(
@@ -50,9 +51,9 @@ def unwrap_partitions(api_layer_object, axis=None, bind_ip=False):
     if axis is None:
 
         def _unwrap_partitions(oid):
-            if bind_ip:
+            if get_ip:
                 return [
-                    [(partition.ip, getattr(partition, oid)) for partition in row]
+                    [(partition.ip(), getattr(partition, oid)) for partition in row]
                     for row in api_layer_object._query_compiler._modin_frame._partitions
                 ]
             else:
@@ -78,7 +79,7 @@ def _unwrap_partitions(oid):
             )
         )
         return [
-            part.coalesce(bind_ip=bind_ip).unwrap(squeeze=True, bind_ip=bind_ip)
+            part.coalesce(get_ip=get_ip).unwrap(squeeze=True, get_ip=get_ip)
             for part in partitions
         ]
 
@@ -90,15 +91,16 @@ def from_partitions(partitions, axis):
     Parameters
     ----------
     partitions : list
-        List of Ray.ObjectRef/Dask.Future referencing to partitions in depend of the engine used.
-        Or list containing tuples of Ray.ObjectRef/Dask.Future referencing to ip addresses of partitions
-        and partitions itself in depend of the engine used.
+        List of Ray.ObjectRef/Dask.Future to partitions depending on the engine used.
+        Or list of tuples of node ip addresses and Ray.ObjectRef/Dask.Future to partitions
+        depending on the engine used (i.e. [(str, Ray.ObjectRef/Dask.Future), ...]).
     axis : None, 0 or 1
-        The `axis` parameter is used to identify what are the partitions passed.
+        The ``axis`` parameter is used to identify what are the partitions passed.
         You have to set:
-        - `axis` to 0 if you want to create DataFrame from row partitions.
-        - `axis` to 1 if you want to create DataFrame from column partitions.
-        - `axis` to None if you want to create DataFrame from 2D list of partitions.
+
+        * ``axis=0`` if you want to create DataFrame from row partitions
+        * ``axis=1`` if you want to create DataFrame from column partitions
+        * ``axis=None`` if you want to create DataFrame from 2D list of partitions
 
     Returns
     -------
diff --git a/modin/engines/base/frame/axis_partition.py b/modin/engines/base/frame/axis_partition.py
index 2c42ee9cc5a..7e055b54953 100644
--- a/modin/engines/base/frame/axis_partition.py
+++ b/modin/engines/base/frame/axis_partition.py
@@ -95,14 +95,14 @@ def shuffle(self, func, lengths, **kwargs):
     def _wrap_partitions(self, partitions):
         return [self.partition_type(obj) for obj in partitions]
 
-    def coalesce(self, bind_ip=False):
+    def coalesce(self, get_ip=False):
         """
         Coalesce the axis partitions into a single partition.
 
         Parameters
         ----------
-        bind_ip : boolean, default False
-            Whether to bind node ip address to a single partition or not.
+        get_ip : boolean, default False
+            Whether to get node ip address to a single partition or not.
 
         Returns
         -------
@@ -110,9 +110,9 @@ def coalesce(self, bind_ip=False):
             An axis partition containing only a single coalesced partition.
         """
         coalesced = self.apply(lambda x: x, num_splits=1, maintain_partitioning=False)
-        return type(self)(coalesced, bind_ip=bind_ip)
+        return type(self)(coalesced, get_ip=get_ip)
 
-    def unwrap(self, squeeze=False, bind_ip=False):
+    def unwrap(self, squeeze=False, get_ip=False):
         """
         Unwrap partitions from axis partition.
 
@@ -120,8 +120,8 @@ def unwrap(self, squeeze=False, bind_ip=False):
         ----------
         squeeze : boolean, default False
             The flag used to unwrap only one partition.
-        bind_ip : boolean, default False
-            Whether to bind node ip address to each partition or not.
+        get_ip : boolean, default False
+            Whether to get node ip address to each partition or not.
 
         Returns
         -------
@@ -130,17 +130,17 @@ def unwrap(self, squeeze=False, bind_ip=False):
 
         Notes
         -----
-        In case bind_ip=True, list containing tuples of Ray.ObjectRef/Dask.Future
-        to node ip addresses and unwrapped partitions, respectively, is returned
-        if Ray/Dask is used as an engine.
+        If `get_ip=True`, list of tuples of node ip addresses and
+        Ray.ObjectRef/Dask.Future to unwrapped partitions, respectively, is returned
+        if Ray/Dask is used as an engine (i.e. [(str, Ray.ObjectRef/Dask.Future), ...]).
         """
         if squeeze and len(self.list_of_blocks) == 1:
-            if bind_ip:
+            if get_ip:
                 return self.list_of_ips[0], self.list_of_blocks[0]
             else:
                 return self.list_of_blocks[0]
         else:
-            if bind_ip:
+            if get_ip:
                 return list(zip(self.list_of_ips, self.list_of_blocks))
             else:
                 return self.list_of_blocks
diff --git a/modin/engines/dask/pandas_on_dask/frame/axis_partition.py b/modin/engines/dask/pandas_on_dask/frame/axis_partition.py
index 8ee82996cc8..8ad9a56ab77 100644
--- a/modin/engines/dask/pandas_on_dask/frame/axis_partition.py
+++ b/modin/engines/dask/pandas_on_dask/frame/axis_partition.py
@@ -21,13 +21,13 @@
 
 
 class PandasOnDaskFrameAxisPartition(PandasFrameAxisPartition):
-    def __init__(self, list_of_blocks, bind_ip=False):
+    def __init__(self, list_of_blocks, get_ip=False):
         # Unwrap from BaseFramePartition object for ease of use
         for obj in list_of_blocks:
             obj.drain_call_queue()
         self.list_of_blocks = [obj.future for obj in list_of_blocks]
-        if bind_ip:
-            self.list_of_ips = [obj.ip for obj in list_of_blocks]
+        if get_ip:
+            self.list_of_ips = [obj.ip() for obj in list_of_blocks]
 
     partition_type = PandasOnDaskFramePartition
     instance_type = Future
diff --git a/modin/engines/dask/pandas_on_dask/frame/partition.py b/modin/engines/dask/pandas_on_dask/frame/partition.py
index bbc16a27e19..929b6884514 100644
--- a/modin/engines/dask/pandas_on_dask/frame/partition.py
+++ b/modin/engines/dask/pandas_on_dask/frame/partition.py
@@ -13,10 +13,11 @@
 
 import pandas
 
-from modin.engines.base.frame.partition import BaseFramePartition
 from modin.data_management.utils import length_fn_pandas, width_fn_pandas
+from modin.engines.base.frame.partition import BaseFramePartition
 
 from distributed.client import get_client
+from distributed import Future
 from distributed.utils import get_ip
 import cloudpickle as pkl
 
@@ -48,7 +49,7 @@ def __init__(self, future, length=None, width=None, ip=None, call_queue=None):
         self.call_queue = call_queue
         self._length_cache = length
         self._width_cache = width
-        self.ip = ip
+        self._ip_cache = ip
 
     def get(self):
         """Flushes the call_queue and returns the data.
@@ -96,7 +97,7 @@ def drain_call_queue(self):
             return
         new_partition = self.apply(lambda x: x)
         self.future = new_partition.future
-        self.ip = new_partition.ip
+        self._ip_cache = new_partition._ip_cache
         self.call_queue = []
 
     def mask(self, row_indices, col_indices):
@@ -111,7 +112,11 @@ def mask(self, row_indices, col_indices):
 
     def __copy__(self):
         return PandasOnDaskFramePartition(
-            self.future, self._length_cache, self._width_cache
+            self.future,
+            length=self._length_cache,
+            width=self._width_cache,
+            ip=self._ip_cache,
+            call_queue=self.call_queue,
         )
 
     def to_pandas(self):
@@ -192,17 +197,24 @@ def width_extraction_fn(cls):
     def length(self):
         if self._length_cache is None:
             self._length_cache = self.apply(lambda df: len(df)).future
-        if isinstance(self._length_cache, type(self.future)):
+        if isinstance(self._length_cache, Future):
             self._length_cache = self._length_cache.result()
         return self._length_cache
 
     def width(self):
         if self._width_cache is None:
             self._width_cache = self.apply(lambda df: len(df.columns)).future
-        if isinstance(self._width_cache, type(self.future)):
+        if isinstance(self._width_cache, Future):
             self._width_cache = self._width_cache.result()
         return self._width_cache
 
+    def ip(self):
+        if self._ip_cache is None:
+            self._ip_cache = self.apply(lambda df: df)._ip_cache
+        if isinstance(self._ip_cache, Future):
+            self._ip_cache = self._ip_cache.result()
+        return self._ip_cache
+
     @classmethod
     def empty(cls):
         return cls(pandas.DataFrame(), 0, 0)
diff --git a/modin/engines/dask/pandas_on_dask/frame/partition_manager.py b/modin/engines/dask/pandas_on_dask/frame/partition_manager.py
index afd58db5d41..bf9f14f0aad 100644
--- a/modin/engines/dask/pandas_on_dask/frame/partition_manager.py
+++ b/modin/engines/dask/pandas_on_dask/frame/partition_manager.py
@@ -23,31 +23,6 @@
 import pandas
 
 from distributed.client import _get_global_client
-import cloudpickle as pkl
-
-
-def deploy_func(df, apply_func, call_queue_df=None, call_queues_other=None, *others):
-    if call_queue_df is not None and len(call_queue_df) > 0:
-        for call, kwargs in call_queue_df:
-            if isinstance(call, bytes):
-                call = pkl.loads(call)
-            if isinstance(kwargs, bytes):
-                kwargs = pkl.loads(kwargs)
-            df = call(df, **kwargs)
-    new_others = np.empty(shape=len(others), dtype=object)
-    for i, call_queue_other in enumerate(call_queues_other):
-        other = others[i]
-        if call_queue_other is not None and len(call_queue_other) > 0:
-            for call, kwargs in call_queue_other:
-                if isinstance(call, bytes):
-                    call = pkl.loads(call)
-                if isinstance(kwargs, bytes):
-                    kwargs = pkl.loads(kwargs)
-                other = call(other, **kwargs)
-        new_others[i] = other
-    if isinstance(apply_func, bytes):
-        apply_func = pkl.loads(apply_func)
-    return apply_func(df, new_others)
 
 
 class DaskFrameManager(BaseFrameManager):
@@ -104,30 +79,21 @@ def get_indices(cls, axis, partitions, index_func):
 
     @classmethod
     def broadcast_apply(cls, axis, apply_func, left, right, other_name="r"):
-        def mapper(df, others):
+        def map_func(df, others):
             other = pandas.concat(others, axis=axis ^ 1)
             return apply_func(df, **{other_name: other})
 
-        client = _get_global_client()
+        rt_axis_parts = cls.axis_partition(right, axis ^ 1)
         return np.array(
             [
                 [
-                    PandasOnDaskFramePartition(
-                        client.submit(
-                            deploy_func,
-                            part.future,
-                            mapper,
-                            part.call_queue,
-                            [obj[col_idx].call_queue for obj in right]
+                    part.apply(
+                        map_func,
+                        **{
+                            "others": rt_axis_parts[col_idx].list_of_blocks
                             if axis
-                            else [obj.call_queue for obj in right[row_idx]],
-                            *(
-                                [obj[col_idx].future for obj in right]
-                                if axis
-                                else [obj.future for obj in right[row_idx]]
-                            ),
-                            pure=False,
-                        )
+                            else rt_axis_parts[row_idx].list_of_blocks
+                        },
                     )
                     for col_idx, part in enumerate(left[row_idx])
                 ]
diff --git a/modin/engines/ray/pandas_on_ray/frame/axis_partition.py b/modin/engines/ray/pandas_on_ray/frame/axis_partition.py
index 21989fe1536..72b9e2ac340 100644
--- a/modin/engines/ray/pandas_on_ray/frame/axis_partition.py
+++ b/modin/engines/ray/pandas_on_ray/frame/axis_partition.py
@@ -21,13 +21,13 @@
 
 
 class PandasOnRayFrameAxisPartition(PandasFrameAxisPartition):
-    def __init__(self, list_of_blocks, bind_ip=False):
+    def __init__(self, list_of_blocks, get_ip=False):
         # Unwrap from BaseFramePartition object for ease of use
         for obj in list_of_blocks:
             obj.drain_call_queue()
         self.list_of_blocks = [obj.oid for obj in list_of_blocks]
-        if bind_ip:
-            self.list_of_ips = [obj.ip for obj in list_of_blocks]
+        if get_ip:
+            self.list_of_ips = [obj.ip() for obj in list_of_blocks]
 
     partition_type = PandasOnRayFramePartition
     instance_type = ray.ObjectRef
diff --git a/modin/engines/ray/pandas_on_ray/frame/partition.py b/modin/engines/ray/pandas_on_ray/frame/partition.py
index 242265ee070..c9a97c19749 100644
--- a/modin/engines/ray/pandas_on_ray/frame/partition.py
+++ b/modin/engines/ray/pandas_on_ray/frame/partition.py
@@ -13,13 +13,13 @@
 
 import pandas
 
-from modin.engines.base.frame.partition import BaseFramePartition
 from modin.data_management.utils import length_fn_pandas, width_fn_pandas
+from modin.engines.base.frame.partition import BaseFramePartition
 from modin.engines.ray.utils import handle_ray_task_error
 
 import ray
-from ray.services import get_node_ip_address
 from ray.worker import RayTaskError
+from ray.services import get_node_ip_address
 
 
 class PandasOnRayFramePartition(BaseFramePartition):
@@ -32,7 +32,7 @@ def __init__(self, object_id, length=None, width=None, ip=None, call_queue=None)
         self.call_queue = call_queue
         self._length_cache = length
         self._width_cache = width
-        self.ip = ip
+        self._ip_cache = ip
 
     def get(self):
         """Gets the object out of the plasma store.
@@ -79,13 +79,17 @@ def drain_call_queue(self):
             self.oid,
             self._length_cache,
             self._width_cache,
-            self.ip,
+            self._ip_cache,
         ) = deploy_ray_func.remote(call_queue, oid)
         self.call_queue = []
 
     def __copy__(self):
         return PandasOnRayFramePartition(
-            self.oid, self._length_cache, self._width_cache, call_queue=self.call_queue
+            self.oid,
+            length=self._length_cache,
+            width=self._width_cache,
+            ip=self._ip_cache,
+            call_queue=self.call_queue,
         )
 
     def to_pandas(self):
@@ -189,6 +193,19 @@ def width(self):
                 handle_ray_task_error(e)
         return self._width_cache
 
+    def ip(self):
+        if self._ip_cache is None:
+            if len(self.call_queue):
+                self.drain_call_queue()
+            else:
+                self._ip_cache = self.apply(lambda df: df)._ip_cache
+        if isinstance(self._ip_cache, ray.ObjectID):
+            try:
+                self._ip_cache = ray.get(self._ip_cache)
+            except RayTaskError as e:
+                handle_ray_task_error(e)
+        return self._ip_cache
+
     @classmethod
     def length_extraction_fn(cls):
         return length_fn_pandas
diff --git a/modin/engines/ray/pandas_on_ray/frame/partition_manager.py b/modin/engines/ray/pandas_on_ray/frame/partition_manager.py
index 530db055971..fa9634d280c 100644
--- a/modin/engines/ray/pandas_on_ray/frame/partition_manager.py
+++ b/modin/engines/ray/pandas_on_ray/frame/partition_manager.py
@@ -25,29 +25,6 @@
 import ray
 
 
-@ray.remote
-def func(df, apply_func, call_queue_df=None, call_queues_other=None, *others):
-    if call_queue_df is not None and len(call_queue_df) > 0:
-        for call, kwargs in call_queue_df:
-            if isinstance(call, ray.ObjectRef):
-                call = ray.get(call)
-            if isinstance(kwargs, ray.ObjectRef):
-                kwargs = ray.get(kwargs)
-            df = call(df, **kwargs)
-    new_others = np.empty(shape=len(others), dtype=object)
-    for i, call_queue_other in enumerate(call_queues_other):
-        other = others[i]
-        if call_queue_other is not None and len(call_queue_other) > 0:
-            for call, kwargs in call_queue_other:
-                if isinstance(call, ray.ObjectRef):
-                    call = ray.get(call)
-                if isinstance(kwargs, ray.ObjectRef):
-                    kwargs = ray.get(kwargs)
-                other = call(other, **kwargs)
-        new_others[i] = other
-    return apply_func(df, new_others)
-
-
 class PandasOnRayFrameManager(RayFrameManager):
     """This method implements the interface in `BaseFrameManager`."""
 
@@ -101,33 +78,24 @@ def get_indices(cls, axis, partitions, index_func=None):
 
     @classmethod
     def broadcast_apply(cls, axis, apply_func, left, right, other_name="r"):
-        def mapper(df, others):
-            other = pandas.concat(others, axis=axis ^ 1)
+        def map_func(df, others):
+            other = pandas.concat(ray.get(others), axis=axis ^ 1)
             return apply_func(df, **{other_name: other})
 
-        mapper = ray.put(mapper)
-        new_partitions = np.array(
+        rt_axis_parts = cls.axis_partition(right, axis ^ 1)
+        return np.array(
             [
                 [
-                    PandasOnRayFramePartition(
-                        func.remote(
-                            part.oid,
-                            mapper,
-                            part.call_queue,
-                            [obj[col_idx].call_queue for obj in right]
+                    part.apply(
+                        map_func,
+                        **{
+                            "others": rt_axis_parts[col_idx].list_of_blocks
                             if axis
-                            else [obj.call_queue for obj in right[row_idx]],
-                            *(
-                                [obj[col_idx].oid for obj in right]
-                                if axis
-                                else [obj.oid for obj in right[row_idx]]
-                            ),
-                        )
+                            else rt_axis_parts[row_idx].list_of_blocks
+                        },
                     )
                     for col_idx, part in enumerate(left[row_idx])
                 ]
                 for row_idx in range(len(left))
             ]
         )
-
-        return new_partitions
diff --git a/modin/engines/ray/utils.py b/modin/engines/ray/utils.py
index 4641fa5d31b..3128202d142 100644
--- a/modin/engines/ray/utils.py
+++ b/modin/engines/ray/utils.py
@@ -22,6 +22,7 @@
     Memory,
     RayPlasmaDir,
     IsOutOfCore,
+    RayPartitionThreshold,
 )
 
 
@@ -146,6 +147,11 @@ def initialize_ray(
                     object_store_memory = None
             else:
                 object_store_memory = int(object_store_memory)
+            partition_threshold = RayPartitionThreshold.get()
+            if partition_threshold is not None:
+                partition_threshold = {
+                    "max_direct_call_object_size": partition_threshold,
+                }
             ray.init(
                 num_cpus=CpuCount.get(),
                 include_dashboard=False,
@@ -157,6 +163,7 @@ def initialize_ray(
                 logging_level=100,
                 _memory=object_store_memory,
                 _lru_evict=True,
+                _system_config=partition_threshold,
             )
         _move_stdlib_ahead_of_site_packages()
         ray.worker.global_worker.run_function_on_all_workers(
diff --git a/modin/experimental/xgboost/xgboost_ray.py b/modin/experimental/xgboost/xgboost_ray.py
index fed5d28d365..da172774f0e 100644
--- a/modin/experimental/xgboost/xgboost_ray.py
+++ b/modin/experimental/xgboost/xgboost_ray.py
@@ -304,8 +304,8 @@ def _train(
     X, y = dtrain
     assert len(X) == len(y)
 
-    X_row_parts = unwrap_partitions(X, axis=0, bind_ip=not evenly_data_distribution)
-    y_row_parts = unwrap_partitions(y, axis=0, bind_ip=not evenly_data_distribution)
+    X_row_parts = unwrap_partitions(X, axis=0, get_ip=not evenly_data_distribution)
+    y_row_parts = unwrap_partitions(y, axis=0, get_ip=not evenly_data_distribution)
     assert len(X_row_parts) == len(y_row_parts), "Unaligned train data"
 
     # Create remote actors
@@ -325,8 +325,8 @@ def _train(
                 lambda actor, *X_y: actor.add_eval_data.remote(
                     *X_y, eval_method=eval_method
                 ),
-                unwrap_partitions(eval_X, axis=0, bind_ip=not evenly_data_distribution),
-                unwrap_partitions(eval_y, axis=0, bind_ip=not evenly_data_distribution),
+                unwrap_partitions(eval_X, axis=0, get_ip=not evenly_data_distribution),
+                unwrap_partitions(eval_y, axis=0, get_ip=not evenly_data_distribution),
                 evenly_data_distribution=evenly_data_distribution,
             )
 
@@ -369,7 +369,7 @@ def _predict(
     s = time.time()
 
     X, _ = data
-    X_row_parts = unwrap_partitions(X, axis=0, bind_ip=not evenly_data_distribution)
+    X_row_parts = unwrap_partitions(X, axis=0, get_ip=not evenly_data_distribution)
 
     # Create remote actors
     actors = create_actors(nthread=nthread)


Code review comments:
diff_hunk: @@ -95,33 +95,33 @@ def shuffle(self, func, lengths, **kwargs):
     def _wrap_partitions(self, partitions):
         return [self.partition_type(obj) for obj in partitions]
 
-    def coalesce(self, bind_ip=False):
+    def coalesce(self, get_ip=False):
path: modin/engines/base/frame/axis_partition.py
text: @user1:
Do you think `force_materialize` is a better name for this? `coalesce` generally happens anyway.

@author:
I was thinking on this. I would prefer something of these: `use_ip`, `materialize_ip`, or `get_ip` at last :) because ObjectRefs to data will not be materilized yet.

@user1:
In the case of `axis_partition`, we return a single object which will have to be materialized. `force_materialize` provides a future, so I think it still makes sense instead of `coalesce`.

@author:
Do you want to rename `coalesce` to `force_materialization`? If so, I don't follow your thought. Why do we want to materialize `axis_partition` after its coaliscing? We need to return ObjectRef to the `axis_partition` with saving asynchrony when `get_ip=False`.

@user1:
My thinking was: the assumption from the developers is that they can treat the axis partition as if it were already coalesced, so a `coalesce` api might cause confusion. `force_materialization` is a scarier name that will not encourage misuse. The fact that it is asynchronous does not change that the whole axis partition is materialized.

@author:
Okay, I see. We probably need to rename it to `force_materilization`.

diff_hunk: @@ -155,6 +155,14 @@ def _get_default(cls):
         return CpuCount.get()
 
 
+class RayPartitionThreshold(EnvironmentVariable, type=int):
path: modin/config/envvars.py
text: @user1:
This is quite specific, do users actually need such low level control over the storage semantics of this specific engine?

@author:
I think, this may be useful for those who do not want to add additional lines with `import ray` and `ray init(...)` to their workflow/scripts. The whole contol is left to Modin, but not to the developer.

@user1:
Let's postpone this specific update to a new issue and PR. I'm not sure it is directly related to the partition API. It has to do with initialization of Ray.

@author:
Okay, it makes sense.