Add support for DataFrame.melt

README.md

@@ -61,5 +61,4 @@ API Coverage
 
 Dask-Expr covers almost everything of the Dask DataFrame API. The only missing features are:
 
-- ``melt``
 - named GroupBy Aggregations

dask_expr/_collection.py

@@ -3988,6 +3988,24 @@ def to_html(self, max_rows=5):
             layers=maybe_pluralize(n_expr, "expression"),
         )
 
+    @derived_from(pd.DataFrame)
+    def melt(
+        self,
+        id_vars=None,
+        value_vars=None,
+        var_name=None,
+        value_name="value",
+        col_level=None,
+    ):
+        return melt(
+            self,
+            id_vars=id_vars,
+            value_vars=value_vars,
+            var_name=var_name,
+            value_name=value_name,
+            col_level=col_level,
+        )
+
     def _repr_data(self):
         meta = self._meta
         index = self._repr_divisions
@@ -5302,6 +5320,26 @@ def concat(
     )
 
 
+def melt(
+    frame,
+    id_vars=None,
+    value_vars=None,
+    var_name=None,
+    value_name="value",
+    col_level=None,
+):
+    return map_partitions(
+        M.melt,
+        frame,
+        clear_divisions=True,
+        id_vars=id_vars,
+        value_vars=value_vars,
+        var_name=var_name,
+        value_name=value_name,
+        col_level=col_level,
+    )
+
+
 def merge(
     left,
     right,

dask_expr/tests/test_collection.py

@@ -2606,3 +2606,33 @@ def test_shape_integer(df):
     result = df.shape[0].compute()
     assert isinstance(result, int)
     assert result == 100
+
+
+@pytest.mark.parametrize(
+    "kwargs",
+    [
+        {},
+        dict(id_vars="int"),
+        dict(value_vars="int"),
+        dict(value_vars=["obj", "int"], var_name="myvar"),
+        dict(id_vars="s1", value_vars=["obj", "int"], value_name="myval"),
+        dict(value_vars=["obj", "s1"]),
+        dict(value_vars=["s1", "s2"]),
+    ],
+)
+def test_melt(kwargs):
+    # Duplicates `dask.dataframe.tests.test_multi.py::test_melt`,
+    # but fails without `clear_divisions=True` in `melt` logic
+    # (upstream test is less thorough).
+    pdf = pd.DataFrame(
+        {
+            "obj": list("abcd") * 5,
+            "s1": list("XY") * 10,
+            "s2": list("abcde") * 4,
+            "int": np.random.randn(20),
+        }
+    )
+    pdf = pdf.astype({"s1": "string[pyarrow]", "s2": "string[pyarrow]"})
+    ddf = from_pandas(pdf, 4)
+
+    assert_eq(ddf.melt(**kwargs), pdf.melt(**kwargs), check_index=False)