Repository: modin-project/modin

Issue:
Remove `DataFrame.to_pickle_distributed` in favour of `DataFrame.modin.to_pickle_distributed`

Discussion:

Patch:
diff --git a/modin/core/dataframe/pandas/dataframe/dataframe.py b/modin/core/dataframe/pandas/dataframe/dataframe.py
index 11709df0..b69bf25a 100644
--- a/modin/core/dataframe/pandas/dataframe/dataframe.py
+++ b/modin/core/dataframe/pandas/dataframe/dataframe.py
@@ -3203,6 +3203,25 @@ class PandasDataframe(ClassLogger):
             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,
@@ -3351,7 +3370,7 @@ class PandasDataframe(ClassLogger):
         if other is not None:
             if not isinstance(other, list):
                 other = [other]
-            other = [o._partitions 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)
diff --git a/modin/core/dataframe/pandas/partitioning/partition_manager.py b/modin/core/dataframe/pandas/partitioning/partition_manager.py
index 27def624..f9a02f7d 100644
--- a/modin/core/dataframe/pandas/partitioning/partition_manager.py
+++ b/modin/core/dataframe/pandas/partitioning/partition_manager.py
@@ -175,6 +175,24 @@ class PandasDataframePartitionManager(ClassLogger, ABC):
 
     # END Abstract Methods
 
+    @classmethod
+    def create_partition_from_metadata(cls, **metadata):
+        """
+        Create NumPy array of partitions that holds an empty dataframe with given metadata.
+
+        Parameters
+        ----------
+        **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.
+        """
+        metadata_dataframe = pandas.DataFrame(**metadata)
+        return np.array([[cls._partition_class.put(metadata_dataframe)]])
+
     @classmethod
     def column_partitions(cls, partitions, full_axis=True):
         """
@@ -1113,6 +1131,8 @@ class PandasDataframePartitionManager(ClassLogger, ABC):
         np.ndarray
             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."""
diff --git a/modin/experimental/pandas/__init__.py b/modin/experimental/pandas/__init__.py
index 34598a8c..19573399 100644
--- a/modin/experimental/pandas/__init__.py
+++ b/modin/experimental/pandas/__init__.py
@@ -32,9 +32,6 @@ Examples
 >>> df = pd.read_csv_glob("data*.csv")
 """
 
-import functools
-import warnings
-
 from modin.pandas import *  # noqa F401, F403
 
 from .io import (  # noqa F401
@@ -45,20 +42,4 @@ from .io import (  # noqa F401
     read_pickle_distributed,
     read_sql,
     read_xml_glob,
-    to_pickle_distributed,
 )
-
-old_to_pickle_distributed = to_pickle_distributed
-
-
-@functools.wraps(to_pickle_distributed)
-def to_pickle_distributed(*args, **kwargs):
-    warnings.warn(
-        "`DataFrame.to_pickle_distributed` is deprecated and will be removed in a future version. "
-        + "Please use `DataFrame.modin.to_pickle_distributed` instead.",
-        category=FutureWarning,
-    )
-    return old_to_pickle_distributed(*args, **kwargs)
-
-
-setattr(DataFrame, "to_pickle_distributed", to_pickle_distributed)  # noqa: F405


Test patch:
diff --git a/modin/pandas/test/dataframe/test_join_sort.py b/modin/pandas/test/dataframe/test_join_sort.py
index 2925f999..653e39be 100644
--- a/modin/pandas/test/dataframe/test_join_sort.py
+++ b/modin/pandas/test/dataframe/test_join_sort.py
@@ -383,6 +383,13 @@ def test_merge(test_data, test_data2):
         modin_df.merge("Non-valid type")
 
 
+def test_merge_empty():
+    data = np.random.uniform(0, 100, size=(2**6, 2**6))
+    pandas_df = pandas.DataFrame(data)
+    modin_df = pd.DataFrame(data)
+    eval_general(modin_df, pandas_df, lambda df: df.merge(df.iloc[:0]))
+
+
 def test_merge_with_mi_columns():
     modin_df1, pandas_df1 = create_test_dfs(
         {