Repository: modin-project/modin
Pull request: FIX-#6879: Convert the right DF to single partition before broadcasting in query_compiler.merge
<!--
Thank you for your contribution!
Please review the contributing docs: https://modin.readthedocs.io/en/latest/development/contributing.html
if you have questions about contributing.
-->

## What do these changes do?

<!-- Please give a short brief about these changes. -->

- [x] first commit message and PR title follow format outlined [here](https://modin.readthedocs.io/en/latest/development/contributing.html#commit-message-formatting)
  > **_NOTE:_**  If you edit the PR title to match this format, you need to add another commit (even if it's empty) or amend your last commit for the CI job that checks the PR title to pick up the new PR title.
- [ ] passes `flake8 modin/ asv_bench/benchmarks scripts/doc_checker.py`
- [ ] passes `black --check modin/ asv_bench/benchmarks scripts/doc_checker.py`
- [ ] signed commit with `git commit -s` <!-- you can amend your commit with a signature via `git commit -amend -s` -->
- [x] Resolves #6879  <!-- issue must be created for each patch -->
- [ ] tests added and passing
- [ ] module layout described at `docs/development/architecture.rst` is up-to-date <!-- if you have added, renamed or removed files or directories please update the documentation accordingly -->


Problem statement:
The query_compiler.merge reconstructs the Right dataframe for every partition of Left Dataframe
The query_compiler.merge reconstructs the Right dataframe from its partitions for every partition of Left Dataframe, The concat operation results in higher memory consumption when the size of right dataframe is large.

A possible option is to combine the right Dataframe partitions to a single partition dataframe by calling a remote function. This single partiotion dataframe is then passed to each partition of left dataframe thus avoiding the reconstruction in every worker while doing merge.

Hints from issue discussion:

Commit message: Update modin/core/dataframe/pandas/partitioning/partition_manager.py

Co-authored-by: Anatoly Myachev <anatoliimyachev@mail.com>

Patch to review:
diff --git a/modin/core/dataframe/pandas/dataframe/dataframe.py b/modin/core/dataframe/pandas/dataframe/dataframe.py
index 920b9b18583..e3ad15e1154 100644
--- a/modin/core/dataframe/pandas/dataframe/dataframe.py
+++ b/modin/core/dataframe/pandas/dataframe/dataframe.py
@@ -2762,6 +2762,35 @@ def explode(self, axis: Union[int, Axis], func: Callable) -> "PandasDataframe":
             partitions, new_index, new_columns, row_lengths, column_widths
         )
 
+    def combine(self) -> "PandasDataframe":
+        """
+        Create a single partition PandasDataframe from the partitions of the current dataframe.
+
+        Returns
+        -------
+        PandasDataframe
+            A single partition PandasDataframe.
+        """
+        partitions = self._partition_mgr_cls.combine(self._partitions)
+        result = self.__constructor__(
+            partitions,
+            index=self.copy_index_cache(),
+            columns=self.copy_columns_cache(),
+            row_lengths=(
+                [sum(self._row_lengths_cache)]
+                if self._row_lengths_cache is not None
+                else None
+            ),
+            column_widths=(
+                [sum(self._column_widths_cache)]
+                if self._column_widths_cache is not None
+                else None
+            ),
+            dtypes=self.copy_dtypes_cache(),
+        )
+        result.synchronize_labels()
+        return result
+
     @lazy_metadata_decorator(apply_axis="both")
     def apply_full_axis(
         self,
diff --git a/modin/core/dataframe/pandas/partitioning/partition_manager.py b/modin/core/dataframe/pandas/partitioning/partition_manager.py
index 2ccbd1de47f..d928c3a57e5 100644
--- a/modin/core/dataframe/pandas/partitioning/partition_manager.py
+++ b/modin/core/dataframe/pandas/partitioning/partition_manager.py
@@ -1141,6 +1141,85 @@ def _apply_func_to_list_of_partitions(cls, func, partitions, **kwargs):
         preprocessed_func = cls.preprocess_func(func)
         return [obj.apply(preprocessed_func, **kwargs) for obj in partitions]
 
+    @classmethod
+    def combine(cls, partitions):
+        """
+        Convert a NumPy 2D array of partitions to a NumPy 2D array of a single partition.
+
+        Parameters
+        ----------
+        partitions : np.ndarray
+            The partitions which have to be converted to a single partition.
+
+        Returns
+        -------
+        np.ndarray
+            A NumPy 2D array of a single partition.
+        """
+
+        def to_pandas_remote(data, partition_shape, *partition_data):
+            """Copy of ``cls.to_pandas()`` method adapted for a remote function."""
+            if all(
+                isinstance(obj, (pandas.DataFrame, pandas.Series))
+                for obj in partition_data
+            ):
+                height, width, *_ = tuple(partition_shape) + (0,)
+                # restore 2d array
+                objs = iter(partition_data)
+                partition_data = [
+                    [next(objs) for _ in range(width)] for __ in range(height)
+                ]
+            else:
+                # Partitions do not always contain pandas objects, for example, hdk uses pyarrow tables.
+                # This implementation comes from the fact that calling `partition.get`
+                # function is not always equivalent to `partition.to_pandas`.
+                partition_data = [
+                    [obj.to_pandas() for obj in part] for part in partition_data
+                ]
+            if all(
+                isinstance(part, pandas.Series)
+                for row in partition_data
+                for part in row
+            ):
+                axis = 0
+            elif all(
+                isinstance(part, pandas.DataFrame)
+                for row in partition_data
+                for part in row
+            ):
+                axis = 1
+            else:
+                ErrorMessage.catch_bugs_and_request_email(True)
+
+            def is_part_empty(part):
+                return part.empty and (
+                    not isinstance(part, pandas.DataFrame) or (len(part.columns) == 0)
+                )
+
+            df_rows = [
+                pandas.concat([part for part in row], axis=axis, copy=False)
+                for row in partition_data
+                if not all(is_part_empty(part) for part in row)
+            ]
+            if len(df_rows) == 0:
+                return pandas.DataFrame()
+            else:
+                return concatenate(df_rows)
+
+        preprocessed_func = cls.preprocess_func(to_pandas_remote)
+        partition_shape = partitions.shape
+        partitions_flattened = partitions.flatten()
+        for idx, part in enumerate(partitions_flattened):
+            if hasattr(part, "force_materialization"):
+                partitions_flattened[idx] = part.force_materialization()
+        partition_refs = [
+            partition.list_of_blocks[0] for partition in partitions_flattened
+        ]
+        combined_partition = partitions.flat[0].apply(
+            preprocessed_func, partition_shape, *partition_refs
+        )
+        return np.array([combined_partition]).reshape(1, -1)
+
     @classmethod
     @wait_computations_if_benchmark_mode
     def apply_func_to_select_indices(
diff --git a/modin/core/storage_formats/pandas/query_compiler.py b/modin/core/storage_formats/pandas/query_compiler.py
index 90a3557f067..715efb3c4e8 100644
--- a/modin/core/storage_formats/pandas/query_compiler.py
+++ b/modin/core/storage_formats/pandas/query_compiler.py
@@ -520,6 +520,7 @@ def merge(self, right, **kwargs):
         left_index = kwargs.get("left_index", False)
         right_index = kwargs.get("right_index", False)
         sort = kwargs.get("sort", False)
+        right_to_broadcast = right._modin_frame.combine()
 
         if how in ["left", "inner"] and left_index is False and right_index is False:
             kwargs["sort"] = False
@@ -620,7 +621,7 @@ def map_func(
                     axis=1,
                     func=map_func,
                     enumerate_partitions=how == "left",
-                    other=right._modin_frame,
+                    other=right_to_broadcast,
                     # We're going to explicitly change the shape across the 1-axis,
                     # so we want for partitioning to adapt as well
                     keep_partitioning=False,
@@ -681,6 +682,7 @@ def join(self, right, **kwargs):
         on = kwargs.get("on", None)
         how = kwargs.get("how", "left")
         sort = kwargs.get("sort", False)
+        right_to_broadcast = right._modin_frame.combine()
 
         if how in ["left", "inner"]:
 
@@ -697,7 +699,7 @@ def map_func(left, right, kwargs=kwargs):  # pragma: no cover
                     num_splits=merge_partitioning(
                         self._modin_frame, right._modin_frame, axis=1
                     ),
-                    other=right._modin_frame,
+                    other=right_to_broadcast,
                 )
             )
             return new_self.sort_rows_by_column_values(on) if sort else new_self


Code review comments:
diff_hunk: @@ -65,3 +67,54 @@ def concatenate(dfs):
         # so do it explicitly
         return dfs[0].copy()
     return pandas.concat(dfs, copy=True)
+
+
+def create_dataframe_from_partition_data(partition_data, partition_shape):
+    """
+    Convert partition data of multiple dataframes to a single dataframe.
+
+    Parameters
+    ----------
+    partition_data : list
+        List of pandas DataFrames/Object references holding pandas DataFrames.
path: modin/core/dataframe/pandas/utils.py
text: @user1:
?
```suggestion
        List of pandas DataFrames/Object references holding pandas DataFrames.
    partition_shape : int or tuple
```