Repository: modin-project/modin
Pull request: REFACTOR-#2918: Refactor internal functionality of partitions
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 #2918  <!-- issue must be created for each patch -->
- [x] tests passing


Problem statement:
Refactor internal functionality of partitions
We can do the following things:
1) Get rid of passing `ray.ObjectRef(s)` within a list to a partition's `apply` (and to a remote call then) to avoid explicit call `ray.get`.
2) Add preprocessing a function for dask engine as it is done for ray engine.

Hints from issue discussion:

Commit message: REFACTOR-#2918: Add comments

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

Patch to review:
diff --git a/modin/engines/base/frame/partition.py b/modin/engines/base/frame/partition.py
index 4dff475c19a..403374b4fcd 100644
--- a/modin/engines/base/frame/partition.py
+++ b/modin/engines/base/frame/partition.py
@@ -44,7 +44,7 @@ def get(self):
         """
         pass
 
-    def apply(self, func, **kwargs):
+    def apply(self, func, *args, **kwargs):
         """
         Apply a function to the object wrapped by this partition.
 
@@ -52,6 +52,8 @@ def apply(self, func, **kwargs):
         ----------
         func : callable
             Function to apply.
+        *args : iterable
+            Additional positional arguments to be passed in `func`.
         **kwargs : dict
             Additional keyword arguments to be passed in `func`.
 
@@ -68,7 +70,7 @@ def apply(self, func, **kwargs):
         """
         pass
 
-    def add_to_apply_calls(self, func, **kwargs):
+    def add_to_apply_calls(self, func, *args, **kwargs):
         """
         Add a function to the call queue.
 
@@ -76,6 +78,8 @@ def add_to_apply_calls(self, func, **kwargs):
         ----------
         func : callable
             Function to be added to the call queue.
+        *args : iterable
+            Additional positional arguments to be passed in `func`.
         **kwargs : dict
             Additional keyword arguments to be passed in `func`.
 
diff --git a/modin/engines/dask/pandas_on_dask/frame/partition.py b/modin/engines/dask/pandas_on_dask/frame/partition.py
index fd732a3a6d8..1174a8e6839 100644
--- a/modin/engines/dask/pandas_on_dask/frame/partition.py
+++ b/modin/engines/dask/pandas_on_dask/frame/partition.py
@@ -21,37 +21,11 @@
 from distributed.client import default_client
 from distributed import Future
 from distributed.utils import get_ip
-import cloudpickle as pkl
 from dask.distributed import wait
 
 from modin.pandas.indexing import compute_sliced_len
 
 
-def apply_list_of_funcs(funcs, df):
-    """
-    Execute all operations stored in the call queue on the partition in a worker process.
-
-    Parameters
-    ----------
-    funcs : list
-        A call queue that needs to be executed on the partition.
-    df : pandas.DataFrame
-        A pandas DataFrame the call queue needs to be executed on.
-
-    Returns
-    -------
-    pandas.DataFrame
-        The resulting pandas DataFrame.
-    str
-        The node IP address of the worker process.
-    """
-    for func, kwargs in funcs:
-        if isinstance(func, bytes):
-            func = pkl.loads(func)
-        df = func(df, **kwargs)
-    return df, get_ip()
-
-
 class PandasOnDaskFramePartition(PandasFramePartition):
     """
     The class implements the interface in ``PandasFramePartition``.
@@ -94,7 +68,7 @@ def get(self):
             return self.future
         return self.future.result()
 
-    def apply(self, func, **kwargs):
+    def apply(self, func, *args, **kwargs):
         """
         Apply a function to the object wrapped by this partition.
 
@@ -102,8 +76,10 @@ def apply(self, func, **kwargs):
         ----------
         func : callable
             A function to apply.
-        **kwargs
-            Additional keyword arguments to be passed in ``func``.
+        *args : iterable
+            Additional positional arguments to be passed in `func`.
+        **kwargs : dict
+            Additional keyword arguments to be passed in `func`.
 
         Returns
         -------
@@ -114,15 +90,23 @@ def apply(self, func, **kwargs):
         -----
         The keyword arguments are sent as a dictionary.
         """
-        func = pkl.dumps(func)
-        call_queue = self.call_queue + [[func, kwargs]]
-        future = default_client().submit(
-            apply_list_of_funcs, call_queue, self.future, pure=False
-        )
-        futures = [default_client().submit(lambda l: l[i], future) for i in range(2)]
+        client = default_client()
+        call_queue = self.call_queue + [[func, args, kwargs]]
+        if len(call_queue) > 1:
+            future = client.submit(
+                apply_list_of_funcs, self.future, call_queue, pure=False
+            )
+        else:
+            # We handle `len(call_queue) == 1` in a different way because
+            # this improves performance a bit.
+            func, args, kwargs = call_queue[0]
+            future = client.submit(apply_func, self.future, func, *args, **kwargs)
+        futures = [
+            client.submit(lambda l, i: l[i], future, i, pure=False) for i in range(2)
+        ]
         return PandasOnDaskFramePartition(futures[0], ip=futures[1])
 
-    def add_to_apply_calls(self, func, **kwargs):
+    def add_to_apply_calls(self, func, *args, **kwargs):
         """
         Add a function to the call queue.
 
@@ -130,8 +114,10 @@ def add_to_apply_calls(self, func, **kwargs):
         ----------
         func : callable
             Function to be added to the call queue.
+        *args : iterable
+            Additional positional arguments to be passed in `func`.
         **kwargs : dict
-            Additional keyword arguments to be passed in ``func``.
+            Additional keyword arguments to be passed in `func`.
 
         Returns
         -------
@@ -143,16 +129,29 @@ def add_to_apply_calls(self, func, **kwargs):
         The keyword arguments are sent as a dictionary.
         """
         return PandasOnDaskFramePartition(
-            self.future, call_queue=self.call_queue + [[pkl.dumps(func), kwargs]]
+            self.future, call_queue=self.call_queue + [[func, args, kwargs]]
         )
 
     def drain_call_queue(self):
         """Execute all operations stored in the call queue on the object wrapped by this partition."""
         if len(self.call_queue) == 0:
             return
-        new_partition = self.apply(lambda x: x)
-        self.future = new_partition.future
-        self._ip_cache = new_partition._ip_cache
+        call_queue = self.call_queue
+        client = default_client()
+        if len(call_queue) > 1:
+            future = client.submit(
+                apply_list_of_funcs, self.future, call_queue, pure=False
+            )
+        else:
+            # We handle `len(call_queue) == 1` in a different way because
+            # this improves performance a bit.
+            func, args, kwargs = call_queue[0]
+            future = client.submit(apply_func, self.future, func, *args, **kwargs)
+        futures = [
+            client.submit(lambda l, i: l[i], future, i, pure=False) for i in range(2)
+        ]
+        self.future = futures[0]
+        self._ip_cache = futures[1]
         self.call_queue = []
 
     def wait(self):
@@ -266,7 +265,7 @@ def preprocess_func(cls, func):
         callable
             An object that can be accepted by ``apply``.
         """
-        return func
+        return default_client().scatter(func, hash=False, broadcast=True)
 
     @classmethod
     def _length_extraction_fn(cls):
@@ -351,3 +350,52 @@ def empty(cls):
             A new ``PandasOnDaskFramePartition`` object.
         """
         return cls(pandas.DataFrame(), 0, 0)
+
+
+def apply_func(partition, func, *args, **kwargs):
+    """
+    Execute a function on the partition in a worker process.
+
+    Parameters
+    ----------
+    partition : pandas.DataFrame
+        A pandas DataFrame the function needs to be executed on.
+    func : callable
+        Function that needs to be executed on `partition`.
+    *args
+        Additional positional arguments to be passed in `func`.
+    **kwargs
+        Additional keyword arguments to be passed in `func`.
+
+    Returns
+    -------
+    pandas.DataFrame
+        The resulting pandas DataFrame.
+    str
+        The node IP address of the worker process.
+    """
+    result = func(partition, *args, **kwargs)
+    return result, get_ip()
+
+
+def apply_list_of_funcs(partition, funcs):
+    """
+    Execute all operations stored in the call queue on the partition in a worker process.
+
+    Parameters
+    ----------
+    partition : pandas.DataFrame
+        A pandas DataFrame the call queue needs to be executed on.
+    funcs : list
+        A call queue that needs to be executed on the partition.
+
+    Returns
+    -------
+    pandas.DataFrame
+        The resulting pandas DataFrame.
+    str
+        The node IP address of the worker process.
+    """
+    for func, args, kwargs in funcs:
+        partition = func(partition, *args, **kwargs)
+    return partition, get_ip()
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 23879cc945c..9d493add939 100644
--- a/modin/engines/dask/pandas_on_dask/frame/partition_manager.py
+++ b/modin/engines/dask/pandas_on_dask/frame/partition_manager.py
@@ -25,53 +25,6 @@
 import pandas
 
 from distributed.client import default_client
-import cloudpickle as pkl
-
-
-def deploy_func(df, apply_func, call_queue_df=None, call_queues_other=None, *others):
-    """
-    Perform `apply_func` for `df` remotely.
-
-    Parameters
-    ----------
-    df : distributed.Future
-        Dataframe to which `apply_func` will be applied.
-        After running function, automaterialization
-        ``distributed.Future``->``pandas.DataFrame`` happens.
-    apply_func : callable
-        The function to apply.
-    call_queue_df : list, default: None
-        The call_queue to be executed on `df`.
-    call_queues_other : list, default: None
-        The call_queue to be executed on `others`.
-    *others : iterable
-        List of other parameters.
-
-    Returns
-    -------
-    The same as returns of `apply_func`.
-    """
-    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 PandasOnDaskFramePartitionManager(PandasFramePartitionManager):
@@ -151,30 +104,20 @@ def broadcast_apply(cls, axis, apply_func, left, right, other_name="r"):
             NumPy array of result partition objects.
         """
 
-        def mapper(df, others):
+        def map_func(df, *others):
             other = pandas.concat(others, axis=axis ^ 1)
             return apply_func(df, **{other_name: other})
 
-        client = default_client()
+        map_func = cls.preprocess_func(map_func)
+        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]
-                            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,
-                        )
+                    part.apply(
+                        map_func,
+                        *rt_axis_parts[col_idx].list_of_blocks
+                        if axis
+                        else rt_axis_parts[row_idx].list_of_blocks,
                     )
                     for col_idx, part in enumerate(left[row_idx])
                 ]
diff --git a/modin/engines/python/pandas_on_python/frame/partition.py b/modin/engines/python/pandas_on_python/frame/partition.py
index 00b095678be..0aee2e43294 100644
--- a/modin/engines/python/pandas_on_python/frame/partition.py
+++ b/modin/engines/python/pandas_on_python/frame/partition.py
@@ -68,7 +68,7 @@ def get(self):
         self.drain_call_queue()
         return self.data.copy()
 
-    def apply(self, func, **kwargs):
+    def apply(self, func, *args, **kwargs):
         """
         Apply a function to the object wrapped by this partition.
 
@@ -76,6 +76,8 @@ def apply(self, func, **kwargs):
         ----------
         func : callable
             Function to apply.
+        *args : iterable
+            Additional positional arguments to be passed in `func`.
         **kwargs : dict
             Additional keyword arguments to be passed in `func`.
 
@@ -85,15 +87,15 @@ def apply(self, func, **kwargs):
             New ``PandasOnPythonFramePartition`` object.
         """
 
-        def call_queue_closure(data, call_queues):
+        def call_queue_closure(data, call_queue):
             """
-            Apply callables from `call_queues` on copy of the `data` and return the result.
+            Apply callables from `call_queue` on copy of the `data` and return the result.
 
             Parameters
             ----------
             data : pandas.DataFrame or pandas.Series
                 Data to use for computations.
-            call_queues : array-like
+            call_queue : array-like
                 Array with callables and it's kwargs to be applied to the `data`.
 
             Returns
@@ -101,9 +103,9 @@ def call_queue_closure(data, call_queues):
             pandas.DataFrame or pandas.Series
             """
             result = data.copy()
-            for func, kwargs in call_queues:
+            for func, args, kwargs in call_queue:
                 try:
-                    result = func(result, **kwargs)
+                    result = func(result, *args, **kwargs)
                 except Exception as e:
                     self.call_queue = []
                     raise e
@@ -111,9 +113,9 @@ def call_queue_closure(data, call_queues):
 
         self.data = call_queue_closure(self.data, self.call_queue)
         self.call_queue = []
-        return PandasOnPythonFramePartition(func(self.data.copy(), **kwargs))
+        return PandasOnPythonFramePartition(func(self.data.copy(), *args, **kwargs))
 
-    def add_to_apply_calls(self, func, **kwargs):
+    def add_to_apply_calls(self, func, *args, **kwargs):
         """
         Add a function to the call queue.
 
@@ -121,6 +123,8 @@ def add_to_apply_calls(self, func, **kwargs):
         ----------
         func : callable
             Function to be added to the call queue.
+        *args : iterable
+            Additional positional arguments to be passed in `func`.
         **kwargs : dict
             Additional keyword arguments to be passed in `func`.
 
@@ -130,7 +134,7 @@ def add_to_apply_calls(self, func, **kwargs):
             New ``PandasOnPythonFramePartition`` object with extended call queue.
         """
         return PandasOnPythonFramePartition(
-            self.data.copy(), call_queue=self.call_queue + [(func, kwargs)]
+            self.data.copy(), call_queue=self.call_queue + [(func, args, kwargs)]
         )
 
     def drain_call_queue(self):
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 88bdcf34b6f..926fe49b27f 100644
--- a/modin/engines/ray/pandas_on_ray/frame/axis_partition.py
+++ b/modin/engines/ray/pandas_on_ray/frame/axis_partition.py
@@ -75,7 +75,7 @@ def deploy_axis_func(
         """
         lengths = kwargs.get("_lengths", None)
         return deploy_ray_func.options(
-            num_returns=num_splits * 4 if lengths is None else len(lengths) * 4
+            num_returns=(num_splits if lengths is None else len(lengths)) * 4
         ).remote(
             PandasFrameAxisPartition.deploy_axis_func,
             axis,
diff --git a/modin/engines/ray/pandas_on_ray/frame/partition.py b/modin/engines/ray/pandas_on_ray/frame/partition.py
index 58aa2e991b5..94aca747b31 100644
--- a/modin/engines/ray/pandas_on_ray/frame/partition.py
+++ b/modin/engines/ray/pandas_on_ray/frame/partition.py
@@ -74,7 +74,7 @@ def get(self):
             self.drain_call_queue()
         return ray.get(self.oid)
 
-    def apply(self, func, **kwargs):
+    def apply(self, func, *args, **kwargs):
         """
         Apply a function to the object wrapped by this partition.
 
@@ -82,7 +82,9 @@ def apply(self, func, **kwargs):
         ----------
         func : callable or ray.ObjectRef
             A function to apply.
-        **kwargs
+        *args : iterable
+            Additional positional arguments to be passed in `func`.
+        **kwargs : dict
             Additional keyword arguments to be passed in `func`.
 
         Returns
@@ -96,11 +98,17 @@ def apply(self, func, **kwargs):
         handle it correctly either way. The keyword arguments are sent as a dictionary.
         """
         oid = self.oid
-        call_queue = self.call_queue + [(func, kwargs)]
-        result, length, width, ip = deploy_ray_func.remote(call_queue, oid)
+        call_queue = self.call_queue + [(func, args, kwargs)]
+        if len(call_queue) > 1:
+            result, length, width, ip = apply_list_of_funcs.remote(oid, call_queue)
+        else:
+            # We handle `len(call_queue) == 1` in a different way because
+            # this dramatically improves performance.
+            func, args, kwargs = call_queue[0]
+            result, length, width, ip = apply_func.remote(oid, func, *args, **kwargs)
         return PandasOnRayFramePartition(result, length, width, ip)
 
-    def add_to_apply_calls(self, func, **kwargs):
+    def add_to_apply_calls(self, func, *args, **kwargs):
         """
         Add a function to the call queue.
 
@@ -108,6 +116,8 @@ def add_to_apply_calls(self, func, **kwargs):
         ----------
         func : callable or ray.ObjectRef
             Function to be added to the call queue.
+        *args : iterable
+            Additional positional arguments to be passed in `func`.
         **kwargs : dict
             Additional keyword arguments to be passed in `func`.
 
@@ -122,7 +132,7 @@ def add_to_apply_calls(self, func, **kwargs):
         handle it correctly either way. The keyword arguments are sent as a dictionary.
         """
         return PandasOnRayFramePartition(
-            self.oid, call_queue=self.call_queue + [(func, kwargs)]
+            self.oid, call_queue=self.call_queue + [(func, args, kwargs)]
         )
 
     def drain_call_queue(self):
@@ -131,12 +141,23 @@ def drain_call_queue(self):
             return
         oid = self.oid
         call_queue = self.call_queue
-        (
-            self.oid,
-            self._length_cache,
-            self._width_cache,
-            self._ip_cache,
-        ) = deploy_ray_func.remote(call_queue, oid)
+        if len(call_queue) > 1:
+            (
+                self.oid,
+                self._length_cache,
+                self._width_cache,
+                self._ip_cache,
+            ) = apply_list_of_funcs.remote(oid, call_queue)
+        else:
+            # We handle `len(call_queue) == 1` in a different way because
+            # this dramatically improves performance.
+            func, args, kwargs = call_queue[0]
+            (
+                self.oid,
+                self._length_cache,
+                self._width_cache,
+                self._ip_cache,
+            ) = apply_func.remote(oid, func, *args, **kwargs)
         self.call_queue = []
 
     def wait(self):
@@ -369,16 +390,58 @@ def get_index_and_columns(df):
 
 
 @ray.remote(num_returns=4)
-def deploy_ray_func(call_queue, partition):  # pragma: no cover
+def apply_func(partition, func, *args, **kwargs):  # pragma: no cover
+    """
+    Execute a function on the partition in a worker process.
+
+    Parameters
+    ----------
+    partition : pandas.DataFrame
+        A pandas DataFrame the function needs to be executed on.
+    func : callable
+        Function that needs to be executed on the partition.
+    *args : iterable
+        Additional positional arguments to be passed in `func`.
+    **kwargs : dict
+        Additional keyword arguments to be passed in `func`.
+
+    Returns
+    -------
+    pandas.DataFrame
+        The resulting pandas DataFrame.
+    int
+        The number of rows of the resulting pandas DataFrame.
+    int
+        The number of columns of the resulting pandas DataFrame.
+    str
+        The node IP address of the worker process.
+    """
+    try:
+        result = func(partition, *args, **kwargs)
+    # Sometimes Arrow forces us to make a copy of an object before we operate on it. We
+    # don't want the error to propagate to the user, and we want to avoid copying unless
+    # we absolutely have to.
+    except ValueError:
+        result = func(partition.copy(), *args, **kwargs)
+    return (
+        result,
+        len(result) if hasattr(result, "__len__") else 0,
+        len(result.columns) if hasattr(result, "columns") else 0,
+        get_node_ip_address(),
+    )
+
+
+@ray.remote(num_returns=4)
+def apply_list_of_funcs(partition, funcs):  # pragma: no cover
     """
     Execute all operations stored in the call queue on the partition in a worker process.
 
     Parameters
     ----------
-    call_queue : list
-        A call queue that needs to be executed on the partition.
     partition : pandas.DataFrame
         A pandas DataFrame the call queue needs to be executed on.
+    funcs : list
+        A call queue that needs to be executed on the partition.
 
     Returns
     -------
@@ -395,26 +458,38 @@ def deploy_ray_func(call_queue, partition):  # pragma: no cover
     def deserialize(obj):
         if isinstance(obj, ObjectIDType):
             return ray.get(obj)
-        return obj
+        elif isinstance(obj, (tuple, list)) and any(
+            isinstance(o, ObjectIDType) for o in obj
+        ):
+            return ray.get(list(obj))
+        elif isinstance(obj, dict) and any(
+            isinstance(val, ObjectIDType) for val in obj.values()
+        ):
+            return dict(zip(obj.keys(), ray.get(list(obj.values()))))
+        else:
+            return obj
 
-    if len(call_queue) > 1:
-        for func, kwargs in call_queue[:-1]:
+    if len(funcs) > 1:
+        for func, args, kwargs in funcs[:-1]:
             func = deserialize(func)
+            args = deserialize(args)
             kwargs = deserialize(kwargs)
             try:
-                partition = func(partition, **kwargs)
+                partition = func(partition, *args, **kwargs)
             except ValueError:
-                partition = func(partition.copy(), **kwargs)
-    func, kwargs = call_queue[-1]
+                partition = func(partition.copy(), *args, **kwargs)
+    func, args, kwargs = funcs[-1]
     func = deserialize(func)
+    args = deserialize(args)
     kwargs = deserialize(kwargs)
+
     try:
-        result = func(partition, **kwargs)
+        result = func(partition, *args, **kwargs)
     # Sometimes Arrow forces us to make a copy of an object before we operate on it. We
     # don't want the error to propagate to the user, and we want to avoid copying unless
     # we absolutely have to.
     except ValueError:
-        result = func(partition.copy(), **kwargs)
+        result = func(partition.copy(), *args, **kwargs)
     return (
         result,
         len(result) if hasattr(result, "__len__") else 0,
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 34d6cceed5c..dad79b5bf52 100644
--- a/modin/engines/ray/pandas_on_ray/frame/partition_manager.py
+++ b/modin/engines/ray/pandas_on_ray/frame/partition_manager.py
@@ -86,52 +86,6 @@ def magic(*args, **kwargs):
     return magic
 
 
-@ray.remote
-def func(df, apply_func, call_queue_df=None, call_queues_other=None, *others):
-    """
-    Perform `apply_func` for `df` remotely.
-
-    Parameters
-    ----------
-    df : ray.ObjectRef
-        Dataframe to which `apply_func` will be applied.
-        After running function, automaterialization
-        ``ray.ObjectRef``->``pandas.DataFrame`` happens.
-    apply_func : {callable, ray.ObjectRef}
-        The function to apply.
-    call_queue_df : list, optional
-        The call queue to be executed on `df`.
-    call_queues_other : list, optional
-        The call queue to be executed on `others`.
-    *others : iterable
-        List of other parameters.
-
-    Returns
-    -------
-    object
-        The same as returns of `apply_func`.
-    """
-    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 PandasOnRayFramePartitionManager(GenericRayFramePartitionManager):
     """The class implements the interface in `PandasFramePartitionManager`."""
 
@@ -208,35 +162,26 @@ def broadcast_apply(cls, axis, apply_func, left, right, other_name="r"):
             An array of partition objects.
         """
 
-        def mapper(df, others):
+        def map_func(df, *others):
             other = pandas.concat(others, axis=axis ^ 1)
             return apply_func(df, **{other_name: other})
 
-        mapper = ray.put(mapper)
-        new_partitions = np.array(
+        map_func = cls.preprocess_func(map_func)
+        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]
-                            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]]
-                            ),
-                        )
+                    part.apply(
+                        map_func,
+                        *rt_axis_parts[col_idx].list_of_blocks
+                        if axis
+                        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
 
     @classmethod
     @progress_bar_wrapper


Code review comments:
diff_hunk: @@ -151,30 +104,20 @@ def broadcast_apply(cls, axis, apply_func, left, right, other_name="r"):
             NumPy array of result partition objects.
         """
 
-        def mapper(df, others):
+        def map_func(df, *others):
             other = pandas.concat(others, axis=axis ^ 1)
             return apply_func(df, **{other_name: other})
 
-        client = default_client()
+        map_func = cls.preprocess_func(map_func)
+        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]
-                            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,
-                        )
+                    part.apply(
+                        map_func,
+                        *rt_axis_parts[col_idx].list_of_blocks
+                        if axis
+                        else rt_axis_parts[row_idx].list_of_blocks,
path: modin/engines/dask/pandas_on_dask/frame/partition_manager.py
text: @user1:
if the star ("unpacking a tuple") is applied to the whole `a if foo else b` statement I suggest wrapping it in braces for readability... if it's applied to only "true" branch then why such asymmetry?

@author:
Yes, the star should be applied to the whole statement. Will be fixed.

@author:
Fixed

diff_hunk: @@ -395,26 +458,38 @@ def deploy_ray_func(call_queue, partition):  # pragma: no cover
     def deserialize(obj):
         if isinstance(obj, ObjectIDType):
             return ray.get(obj)
-        return obj
+        elif isinstance(obj, (tuple, list)) and any(
+            isinstance(o, ObjectIDType) for o in obj
+        ):
+            return ray.get(list(obj))
+        elif isinstance(obj, dict) and any(
+            isinstance(val, ObjectIDType) for val in obj.values()
+        ):
+            return dict(zip(obj.keys(), ray.get(list(obj.values()))))
+        else:
+            return obj
 
-    if len(call_queue) > 1:
-        for func, kwargs in call_queue[:-1]:
+    if len(funcs) > 1:
path: modin/engines/ray/pandas_on_ray/frame/partition.py
text: @user1:
why do we need this `if` and why do we special-case the last entry in the list?..

@author:
I left this code as it was but yes, we can rid of special-casing the last entry in the list.

@author:
Fixed