Repository: modin-project/modin
Issue category: Performance Issue

Binary operations in Modin is slower than in pandas in case operands are Modin structures 
<details><summary>Benchmark code</summary>

```python
import time

import numpy as np
import modin.pandas as pd
# import pandas as pd
import modin.config as cfg

cfg.BenchmarkMode.put(True)
# cfg.NPartitions.put(1)

def generate_data(path, nrows=5000, n_random_cols=100):
    labels = ["red", "blue", "black"]
    data = {
        "Int": np.random.randint(0, 100, nrows),
        "Float": np.random.rand(nrows),
        "Color": [labels[np.random.randint(0, len(labels), 1)[0]] for _ in range(nrows)],
    }
    data_random = {f"col{i}":  np.random.rand(nrows) for i in range(n_random_cols)}
    data.update(data_random)
    df = pd.DataFrame(data)

    df.to_csv(path, index=False)

if __name__ == "__main__":
    path = "data.csv"

    # generate_data(path, nrows=5000000, n_random_cols=100)

    print("Reading data...")
    df = pd.read_csv(path)

    # df = df._to_pandas()
    print(f"Original shape: {df.shape}")

    print("Filters...")

    t1 = time.time()
    filter_series1 = (df.Int < 50)
    print(f'Time binary op #1 (scalar): {time.time() - t1} s')

    t2 = time.time()
    filter_series2 = (df.Color == "red")
    print(f'Time binary op #2 (scalar): {time.time() - t2} s')

    t3 = time.time()
    filter_series3 = (df.Color != "blue")
    print(f'Time binary op #3 (scalar): {time.time() - t3} s')

    t4 = time.time()
    filter_series4 = (df.Float < 0.5)
    print(f'Time binary op #4 (scalar): {time.time() - t4} s')

    t5 = time.time()
    filter_series = filter_series1 & filter_series2 & filter_series3 & filter_series4
    print(f'Time binary op #5 (query compilers): {time.time() - t5} s')

    t6 = time.time()
    df = df[filter_series] # Default to pandas in case indexer is query_compiler
    print(f'Time getting by mask: {time.time() - t6} s')

    print(f"Result shape: {df.shape}")
    print(f'Time filters: {time.time() - t1} s')

```

</details>

Table below shows number from binary operation №5 (from the code above) (`&` operation between Modin Series).
| Shape | (10m, 103) | (10m, 4) | (5m, 103) |
| ----------- | ----------- | ----------- | ----------- |
| modin      | 4.996 | 5.33 | 4.98 |
| modin (NPartitions==1)| 0.184| 0.17| 0.11 |
| pandas | 0.025 | 0.024 | 0.008 |

Decreasing of number of partitions improves Modin's time. Possible, there is some gaps in parallelization.


Functions to edit:
modin/core/dataframe/pandas/dataframe/dataframe.py:PandasDataframe.binary_op
modin/core/dataframe/pandas/partitioning/partition_manager.py:PandasDataframePartitionManager.binary_operation
modin/core/execution/ray/implementations/pandas_on_ray/partitioning/partition_manager.py:PandasOnRayDataframePartitionManager.binary_operation

Patch:
diff --git a/docs/release_notes/release_notes-0.16.0.rst b/docs/release_notes/release_notes-0.16.0.rst
index 1d1adaf0105..372c9c89cad 100644
--- a/docs/release_notes/release_notes-0.16.0.rst
+++ b/docs/release_notes/release_notes-0.16.0.rst
@@ -7,8 +7,9 @@ Key Features and Updates
 
 * Stability and Bugfixes
   * FIX-#4543: Fix `read_csv` in case skiprows=<0, []> (#4544)
+  * FIX-#4059: Add cell-wise execution for binary ops, fix bin ops for empty dataframes (#4391)
 * Performance enhancements
-  *
+  * PERF-#4182: Add cell-wise execution for binary ops, fix bin ops for empty dataframes (#4391)
 * Benchmarking enhancements
   *
 * Refactor Codebase
diff --git a/modin/core/dataframe/pandas/dataframe/dataframe.py b/modin/core/dataframe/pandas/dataframe/dataframe.py
index bcfc194a6b7..607a4da4420 100644
--- a/modin/core/dataframe/pandas/dataframe/dataframe.py
+++ b/modin/core/dataframe/pandas/dataframe/dataframe.py
@@ -2507,18 +2507,38 @@ def binary_op(self, op, right_frame, join_type="outer"):
         left_parts, right_parts, joined_index, row_lengths = self._copartition(
             0, right_frame, join_type, sort=True
         )
-        # unwrap list returned by `copartition`.
-        right_parts = right_parts[0]
-        new_frame = self._partition_mgr_cls.binary_operation(
-            1, left_parts, lambda l, r: op(l, r), right_parts
+        new_left_frame = self.__constructor__(
+            left_parts, joined_index, self.columns, row_lengths, self._column_widths
         )
-        new_columns = self.columns.join(right_frame.columns, how=join_type)
+        new_right_frame = self.__constructor__(
+            right_parts[0],
+            joined_index,
+            right_frame.columns,
+            row_lengths,
+            right_frame._column_widths,
+        )
+
+        (
+            left_parts,
+            right_parts,
+            joined_columns,
+            column_widths,
+        ) = new_left_frame._copartition(1, new_right_frame, join_type, sort=True)
+
+        new_frame = (
+            np.array([])
+            if len(left_parts) == 0 or len(right_parts[0]) == 0
+            else self._partition_mgr_cls.binary_operation(
+                left_parts, op, right_parts[0]
+            )
+        )
+
         return self.__constructor__(
             new_frame,
             joined_index,
-            new_columns,
+            joined_columns,
             row_lengths,
-            column_widths=self._column_widths_cache,
+            column_widths,
         )
 
     @lazy_metadata_decorator(apply_axis="both")
diff --git a/modin/core/dataframe/pandas/partitioning/partition_manager.py b/modin/core/dataframe/pandas/partitioning/partition_manager.py
index 232ea093fc9..69379a67d1d 100644
--- a/modin/core/dataframe/pandas/partitioning/partition_manager.py
+++ b/modin/core/dataframe/pandas/partitioning/partition_manager.py
@@ -1242,44 +1242,39 @@ def compute_part_size(indexer, remote_part, part_idx, axis):
 
     @classmethod
     @wait_computations_if_benchmark_mode
-    def binary_operation(cls, axis, left, func, right):
+    def binary_operation(cls, left, func, right):
         """
-        Apply a function that requires two PandasDataframe objects.
+        Apply a function that requires two ``PandasDataframe`` objects.
 
         Parameters
         ----------
-        axis : {0, 1}
-            The axis to apply the function over (0 - rows, 1 - columns).
         left : np.ndarray
-            The partitions of left PandasDataframe.
+            The partitions of left ``PandasDataframe``.
         func : callable
             The function to apply.
         right : np.ndarray
-            The partitions of right PandasDataframe.
+            The partitions of right ``PandasDataframe``.
 
         Returns
         -------
         np.ndarray
             A NumPy array with new partitions.
         """
-        if axis:
-            left_partitions = cls.row_partitions(left)
-            right_partitions = cls.row_partitions(right)
-        else:
-            left_partitions = cls.column_partitions(left)
-            right_partitions = cls.column_partitions(right)
+        [part.drain_call_queue() for part in right.flatten()]
+
         func = cls.preprocess_func(func)
-        result = np.array(
+        return np.array(
             [
-                left_partitions[i].apply(
-                    func,
-                    num_splits=NPartitions.get(),
-                    other_axis_partition=right_partitions[i],
-                )
-                for i in range(len(left_partitions))
+                [
+                    part.apply(
+                        func,
+                        right[row_idx][col_idx]._data,
+                    )
+                    for col_idx, part in enumerate(left[row_idx])
+                ]
+                for row_idx in range(len(left))
             ]
         )
-        return result if axis else result.T
 
     @classmethod
     @wait_computations_if_benchmark_mode
diff --git a/modin/core/execution/ray/implementations/pandas_on_ray/partitioning/partition_manager.py b/modin/core/execution/ray/implementations/pandas_on_ray/partitioning/partition_manager.py
index b09b257dcea..f525d2ada7b 100644
--- a/modin/core/execution/ray/implementations/pandas_on_ray/partitioning/partition_manager.py
+++ b/modin/core/execution/ray/implementations/pandas_on_ray/partitioning/partition_manager.py
@@ -525,14 +525,12 @@ def apply_func_to_indices_both_axis(
 
     @classmethod
     @progress_bar_wrapper
-    def binary_operation(cls, axis, left, func, right):
+    def binary_operation(cls, left, func, right):
         """
         Apply a function that requires partitions of two ``PandasOnRayDataframe`` objects.
 
         Parameters
         ----------
-        axis : {0, 1}
-            The axis to apply the function over (0 - rows, 1 - columns).
         left : np.ndarray
             The partitions of left ``PandasOnRayDataframe``.
         func : callable
@@ -546,5 +544,5 @@ def binary_operation(cls, axis, left, func, right):
             A NumPy array with new partitions.
         """
         return super(PandasOnRayDataframePartitionManager, cls).binary_operation(
-            axis, left, func, right
+            left, func, right
         )

Test patch: diff --git a/modin/pandas/test/dataframe/test_binary.py b/modin/pandas/test/dataframe/test_binary.py
index 6018bf6da7b..c338bdbd030 100644
--- a/modin/pandas/test/dataframe/test_binary.py
+++ b/modin/pandas/test/dataframe/test_binary.py
@@ -249,3 +249,40 @@ def test_duplicate_indexes():
     modin_df2, pandas_df2 = create_test_dfs({"a": data, "b": data})
     df_equals(modin_df1 / modin_df2, pandas_df1 / pandas_df2)
     df_equals(modin_df1 / modin_df1, pandas_df1 / pandas_df1)
+
+
+@pytest.mark.parametrize("subset_operand", ["left", "right"])
+def test_mismatched_col_partitions(subset_operand):
+    data = [0, 1, 2, 3]
+    modin_df1, pandas_df1 = create_test_dfs({"a": data, "b": data})
+    modin_df_tmp, pandas_df_tmp = create_test_dfs({"c": data})
+
+    modin_df2 = pd.concat([modin_df1, modin_df_tmp], axis=1)
+    pandas_df2 = pandas.concat([pandas_df1, pandas_df_tmp], axis=1)
+
+    if subset_operand == "right":
+        modin_res = modin_df2 + modin_df1
+        pandas_res = pandas_df2 + pandas_df1
+    else:
+        modin_res = modin_df1 + modin_df2
+        pandas_res = pandas_df1 + pandas_df2
+
+    df_equals(modin_res, pandas_res)
+
+
+@pytest.mark.parametrize("empty_operand", ["right", "left", "both"])
+def test_empty_df(empty_operand):
+    modin_df, pandas_df = create_test_dfs([0, 1, 2, 0, 1, 2])
+    modin_df_empty, pandas_df_empty = create_test_dfs()
+
+    if empty_operand == "right":
+        modin_res = modin_df + modin_df_empty
+        pandas_res = pandas_df + pandas_df_empty
+    elif empty_operand == "left":
+        modin_res = modin_df_empty + modin_df
+        pandas_res = pandas_df_empty + pandas_df
+    else:
+        modin_res = modin_df_empty + modin_df_empty
+        pandas_res = pandas_df_empty + pandas_df_empty
+
+    df_equals(modin_res, pandas_res)