Remove legacy conversion functions

Author: phofl

dask_expr/_backends.py

@@ -4,8 +4,9 @@
 import pandas as pd
 from dask.backends import CreationDispatch
 from dask.dataframe.backends import DataFrameBackendEntrypoint
-from dask.dataframe.dispatch import to_pandas_dispatch
+from dask.dataframe.dispatch import get_parallel_type, to_pandas_dispatch
 
+from dask_expr import FrameBase
 from dask_expr._dispatch import get_collection_type
 from dask_expr._expr import ToBackend
 
@@ -130,3 +131,8 @@ def get_collection_type_object(_):
 @get_collection_type.register_lazy("cudf")
 def _register_cudf():
     import dask_cudf  # noqa: F401
+
+
+@get_parallel_type.register(FrameBase)
+def get_parallel_type_frame(o):
+    return get_parallel_type(o._meta)

dask_expr/_collection.py

@@ -23,7 +23,6 @@
 from dask.dataframe.core import (
     _concat,
     _convert_to_numeric,
-    _Frame,
     _repr_data_series,
     _sqrt_and_convert_to_timedelta,
     check_divisions,
@@ -32,7 +31,6 @@
     is_dataframe_like,
     is_series_like,
     meta_warning,
-    new_dd_object,
 )
 from dask.dataframe.dispatch import is_categorical_dtype, make_meta, meta_nonempty
 from dask.dataframe.multi import warn_dtype_mismatch
@@ -1370,25 +1368,6 @@ def repartition(
                 Repartition(self, npartitions, divisions, force, partition_size, freq)
             )
 
-    def to_legacy_dataframe(self, optimize: bool = True, **optimize_kwargs) -> _Frame:
-        """Convert to a legacy dask-dataframe collection
-
-        Parameters
-        ----------
-        optimize
-            Whether to optimize the underlying `Expr` object before conversion.
-        **optimize_kwargs
-            Key-word arguments to pass through to `optimize`.
-        """
-        warnings.warn(
-            "to_legacy_dataframe is deprecated and will be removed in a future release. "
-            "The legacy implementation as a whole is deprecated and will be removed, making "
-            "this method unnecessary.",
-            FutureWarning,
-        )
-        df = self.optimize(**optimize_kwargs) if optimize else self
-        return new_dd_object(df.dask, df._name, df._meta, df.divisions)
-
     def to_dask_array(
         self, lengths=None, meta=None, optimize: bool = True, **optimize_kwargs
     ) -> Array:
@@ -5052,28 +5031,6 @@ def from_dict(
     )
 
 
-def from_legacy_dataframe(ddf: _Frame, optimize: bool = True) -> FrameBase:
-    """Create a dask-expr collection from a legacy dask-dataframe collection
-
-    Parameters
-    ----------
-    optimize
-        Whether to optimize the graph before conversion.
-    """
-    warnings.warn(
-        "from_legacy_dataframe is deprecated and will be removed in a future release. "
-        "The legacy implementation as a whole is deprecated and will be removed, making "
-        "this method unnecessary.",
-        FutureWarning,
-    )
-    graph = ddf.dask
-    if optimize:
-        graph = ddf.__dask_optimize__(graph, ddf.__dask_keys__())
-    return from_graph(
-        graph, ddf._meta, ddf.divisions, ddf.__dask_keys__(), key_split(ddf._name)
-    )
-
-
 def from_dask_array(x, columns=None, index=None, meta=None):
     """Create a Dask DataFrame from a Dask Array.
 
@@ -5809,7 +5766,7 @@ def merge_asof(
     del kwargs["on"]
 
     for o in [left_on, right_on]:
-        if isinstance(o, _Frame):
+        if isinstance(o, FrameBase):
             raise NotImplementedError(
                 "Dask collections not currently allowed in merge columns"
             )

dask_expr/io/tests/test_io.py

@@ -15,14 +15,13 @@
     from_array,
     from_dask_array,
     from_dict,
-    from_legacy_dataframe,
     from_map,
     from_pandas,
     optimize,
     read_csv,
     read_parquet,
 )
-from dask_expr._expr import Expr, Replace
+from dask_expr._expr import Replace
 from dask_expr.io import FromArray, FromMap, ReadCSV, ReadParquet, parquet
 from dask_expr.tests._util import _backend_library
 
@@ -227,25 +226,6 @@ def test_parquet_complex_filters(tmpdir):
     assert_eq(got.optimize(), expect)
 
 
-@pytest.mark.parametrize("optimize", [True, False])
-def test_from_legacy_dataframe(optimize):
-    ddf = dd.from_dict({"a": range(100)}, npartitions=10)
-    with pytest.warns(FutureWarning, match="is deprecated"):
-        df = from_legacy_dataframe(ddf, optimize=optimize)
-    assert isinstance(df.expr, Expr)
-    assert_eq(df, ddf)
-
-
-@pytest.mark.parametrize("optimize", [True, False])
-def test_to_legacy_dataframe(optimize):
-    pdf = pd.DataFrame({"x": [1, 4, 3, 2, 0, 5]})
-    df = from_pandas(pdf, npartitions=2)
-    with pytest.warns(FutureWarning, match="is deprecated"):
-        ddf = df.to_legacy_dataframe(optimize=optimize)
-    assert isinstance(ddf, dd.core.DataFrame)
-    assert_eq(df, ddf)
-
-
 @pytest.mark.parametrize("optimize", [True, False])
 def test_to_dask_array(optimize):
     pdf = pd.DataFrame({"x": [1, 4, 3, 2, 0, 5]})