Repository: dask/dask

Issue:
Deprecate convert_dtype in apply
deprecated in pandas, so lets rip it out now before we move to dask-expr

Discussion:
github-actions[bot]: ## Unit Test Results
_See [test report](https://dask.github.io/dask/test_report.html) for an extended history of previous test failures. This is useful for diagnosing flaky tests._

     15 files  ±0       15 suites  ±0   3h 24m 10s :stopwatch: - 16m 50s
 12 911 tests ±0   11 959 :white_check_mark: ±0     929 :zzz: ±0  23 :x: ±0 
159 969 runs  ±0  143 379 :white_check_mark: +2  16 567 :zzz:  - 2  23 :x: ±0 

For more details on these failures, see [this check](https://github.com/dask/dask/runs/20618956311).

Results for commit b5aa9e68. ± Comparison against base commit e72f93e4.

[test-results]:data:application/gzip;base64,H4sIACA1qWUC/03Nyw6DIBCF4VdpWHch4IDTl2lGLgmplwZh1fTdi1rF5f9NcubDfBjcwh43DvcbW3JItWyOlMI8rS0ENIXKNW13gZwf/VyyMStyBKz4Cu+CKCp5CkMhIU9xMc6x0Lod87S/RlT473O6lVJX3Ke5AqUPu25vcJ028ziGVIL1QIROdQYkSWldh2Rtw4Wm1hKB7rzvey8b9v0BzXTtMxsBAAA=


Patch:
diff --git a/dask/dataframe/__init__.py b/dask/dataframe/__init__.py
index b247c4b76..0f64a152d 100644
--- a/dask/dataframe/__init__.py
+++ b/dask/dataframe/__init__.py
@@ -14,67 +14,82 @@ def _dask_expr_enabled() -> bool:
 
 
 if _dask_expr_enabled():
-    from dask_expr import (
-        DataFrame,
-        Index,
-        Series,
-        concat,
-        from_array,
-        from_dask_array,
-        from_dask_dataframe,
-        from_delayed,
-        from_dict,
-        from_graph,
-        from_map,
-        from_pandas,
-        get_dummies,
-        isna,
-        map_overlap,
-        map_partitions,
-        merge,
-        merge_asof,
-        pivot_table,
-        read_csv,
-        read_hdf,
-        read_json,
-        read_orc,
-        read_parquet,
-        read_sql,
-        read_sql_query,
-        read_sql_table,
-        read_table,
-        repartition,
-        to_bag,
-        to_csv,
-        to_datetime,
-        to_hdf,
-        to_json,
-        to_numeric,
-        to_orc,
-        to_parquet,
-        to_records,
-        to_sql,
-        to_timedelta,
-    )
+    try:
+        from dask_expr import (
+            DataFrame,
+            Index,
+            Series,
+            concat,
+            from_array,
+            from_dask_array,
+            from_dask_dataframe,
+            from_delayed,
+            from_dict,
+            from_graph,
+            from_map,
+            from_pandas,
+            get_dummies,
+            isna,
+            map_overlap,
+            map_partitions,
+            merge,
+            merge_asof,
+            pivot_table,
+            read_csv,
+            read_hdf,
+            read_json,
+            read_orc,
+            read_parquet,
+            read_sql,
+            read_sql_query,
+            read_sql_table,
+            read_table,
+            repartition,
+            to_bag,
+            to_csv,
+            to_datetime,
+            to_hdf,
+            to_json,
+            to_numeric,
+            to_orc,
+            to_parquet,
+            to_records,
+            to_sql,
+            to_timedelta,
+        )
+
+        import dask.dataframe._pyarrow_compat
+        from dask.base import compute
+        from dask.dataframe import backends, dispatch
+        from dask.dataframe.io import demo
+        from dask.dataframe.utils import assert_eq
+
+        def raise_not_implemented_error(attr_name):
+            def inner_func(*args, **kwargs):
+                raise NotImplementedError(
+                    f"Function {attr_name} is not implemented for dask-expr."
+                )
+
+            return inner_func
 
-    import dask.dataframe._pyarrow_compat
-    from dask.base import compute
-    from dask.dataframe import backends, dispatch
-    from dask.dataframe.io import demo
-    from dask.dataframe.utils import assert_eq
+        _Frame = raise_not_implemented_error("_Frame")
+        Aggregation = raise_not_implemented_error("Aggregation")
+        read_fwf = raise_not_implemented_error("read_fwf")
+        melt = raise_not_implemented_error("melt")
 
-    def raise_not_implemented_error(attr_name):
-        def inner_func(*args, **kwargs):
-            raise NotImplementedError(
-                f"Function {attr_name} is not implemented for dask-expr."
-            )
+    # Due to the natural circular imports caused from dask-expr
+    # wanting to import things from dask.dataframe, this module's init
+    # can be run multiple times as it walks code trying to import
+    # dask-expr while dask-expr is also trying to import from dask.dataframe
+    # Each time this happens and hits a circular import, we can reload
+    # dask.dataframe to update itself until dask-expr is fully initialized.
+    # TODO: This can go away when dask-expr is merged into dask
+    except ImportError:
+        import importlib
 
-        return inner_func
+        import dask.dataframe as dd
 
-    _Frame = raise_not_implemented_error("_Frame")
-    Aggregation = raise_not_implemented_error("Aggregation")
-    read_fwf = raise_not_implemented_error("read_fwf")
-    melt = raise_not_implemented_error("melt")
+        dd = importlib.reload(dd)
 
 else:
     try:
diff --git a/dask/dataframe/methods.py b/dask/dataframe/methods.py
index 546510ee5..6b82ad912 100644
--- a/dask/dataframe/methods.py
+++ b/dask/dataframe/methods.py
@@ -32,6 +32,7 @@ from dask.dataframe.dispatch import (  # noqa: F401
     union_categoricals,
 )
 from dask.dataframe.utils import is_dataframe_like, is_index_like, is_series_like
+from dask.utils import _deprecated_kwarg
 
 # cuDF may try to import old dispatch functions
 hash_df = hash_object_dispatch
@@ -56,6 +57,7 @@ def iloc(df, cindexer=None):
     return df.iloc[:, cindexer]
 
 
+@_deprecated_kwarg("convert_dtype", None)
 def apply(df, *args, **kwargs):
     with check_convert_dtype_deprecation():
         with check_apply_dataframe_deprecation():


Test patch:
diff --git a/dask/dataframe/io/tests/test_parquet.py b/dask/dataframe/io/tests/test_parquet.py
index 6d423b1ba..cfd3b91ae 100644
--- a/dask/dataframe/io/tests/test_parquet.py
+++ b/dask/dataframe/io/tests/test_parquet.py
@@ -2267,7 +2267,6 @@ def test_writing_parquet_with_unknown_kwargs(tmpdir, engine):
         ddf.to_parquet(fn, engine=engine, unknown_key="unknown_value")
 
 
-@pytest.mark.skipif(DASK_EXPR_ENABLED, reason="circular import")
 def test_to_parquet_with_get(tmpdir, engine):
     from dask.multiprocessing import get as mp_get
 
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index 8ac825885..470ea0088 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -3411,7 +3411,7 @@ def test_apply_convert_dtype(convert_dtype):
     df = pd.DataFrame({"x": [2, 3, 4, 5], "y": [10, 20, 30, 40]})
     ddf = dd.from_pandas(df, npartitions=2)
     kwargs = {} if convert_dtype is None else {"convert_dtype": convert_dtype}
-    should_warn = PANDAS_GE_210 and convert_dtype is not None
+    pd_should_warn = PANDAS_GE_210 and convert_dtype is not None
 
     meta_val = ddf.x._meta_nonempty.iloc[0]
 
@@ -3420,11 +3420,17 @@ def test_apply_convert_dtype(convert_dtype):
         assert x != meta_val
         return x + 1
 
-    with _check_warning(should_warn, FutureWarning, "the convert_dtype parameter"):
+    with _check_warning(pd_should_warn, FutureWarning, "the convert_dtype parameter"):
         expected = df.x.apply(func, **kwargs)
-    with _check_warning(should_warn, FutureWarning, "the convert_dtype parameter"):
+    with _check_warning(pd_should_warn, FutureWarning, "the convert_dtype parameter"):
         result = ddf.x.apply(func, **kwargs, meta=expected)
-    assert_eq(result, expected)
+
+    dask_should_warn = pytest.warns(FutureWarning, match="the 'convert_dtype' keyword")
+    if convert_dtype is None:
+        dask_should_warn = contextlib.nullcontext()
+
+    with dask_should_warn:
+        assert_eq(result, expected)
 
 
 def test_apply_warns():