adding _extract_partitions

modin/core/dataframe/pandas/dataframe/dataframe.py

@@ -3203,6 +3203,25 @@ def _prepare_frame_to_broadcast(self, axis, indices, broadcast_all):
             passed_len += len(internal)
         return result_dict
 
+    def _extract_partitions(self):
+        """
+        Extract partitions if partitions are present.
+
+        If partitions are empty return a dummy partition with empty data but
+        index and columns of current dataframe.
+
+        Returns
+        -------
+        np.ndarray
+            NumPy array with extracted partitions.
+        """
+        if self._partitions.size > 0:
+            return self._partitions
+        else:
+            return self._partition_mgr_cls.create_partition_from_metadata(
+                index=self.index, columns=self.columns
+            )
+
     @lazy_metadata_decorator(apply_axis="both")
     def broadcast_apply_select_indices(
         self,
@@ -3348,26 +3367,10 @@ def broadcast_apply_full_axis(
         PandasDataframe
             New Modin DataFrame.
         """
-
-        def get_partitions(df):
-            """Deal with the corner case if the "other" dataframe has no partitions."""
-            if df._partitions.size > 0:
-                return df._partitions
-            else:
-                return np.array(
-                    [
-                        [
-                            self._partition_mgr_cls._partition_class.put(
-                                pandas.DataFrame(index=df.index, columns=df.columns)
-                            )
-                        ]
-                    ]
-                )
-
         if other is not None:
             if not isinstance(other, list):
                 other = [other]
-            other = [get_partitions(o) for o in other] if len(other) else None
+            other = [o._extract_partitions() for o in other] if len(other) else None
 
         if apply_indices is not None:
             numeric_indices = self.get_axis(axis ^ 1).get_indexer_for(apply_indices)

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

@@ -176,21 +176,22 @@ def preprocess_func(cls, map_func):
     # END Abstract Methods
 
     @classmethod
-    def create_partition_from_data(cls, data):
+    def create_partition_from_metadata(cls, **metadata):
         """
-        Create NumPy array of partitions that wrapps the given data.
+        Create NumPy array of partitions that holds an empty dataframe with given metadata.
 
         Parameters
         ----------
-        data : pandas.DataFrame or pandas.Series
-            Data that has to be wrapped in partition.
+        metadata : dict
+            Metadata that has to be wrapped in a partition.
 
         Returns
         -------
         np.ndarray
             A NumPy 2D array of a single partition which contains the data.
         """
-        return np.array([[cls._partition_class.put(data)]])
+        metadata_dataframe = pandas.DataFrame(**metadata)
+        return np.array([[cls._partition_class.put(metadata_dataframe)]])
 
     @classmethod
     def column_partitions(cls, partitions, full_axis=True):
@@ -1131,15 +1132,15 @@ def combine(cls, partitions):
             A NumPy 2D array of a single partition.
         """
 
+        if partitions.size <= 1:
+            return partitions
+
         def to_pandas_remote(df, partition_shape, *dfs):
             """Copy of ``cls.to_pandas()`` method adapted for a remote function."""
             return create_pandas_df_from_partitions(
                 (df,) + dfs, partition_shape, called_from_remote=True
             )
 
-        if partitions.size <= 1:
-            return partitions
-
         preprocessed_func = cls.preprocess_func(to_pandas_remote)
         partition_shape = partitions.shape
         partitions_flattened = partitions.flatten()