Repository: narwhals-dev/narwhals

Issue:
[Enh]: Support for `DataFrame.pivot`
### We would like to learn about your use case. For example, if this feature is needed  to adopt Narwhals in an open source project, could you please enter the link to it below?


_No response_

### Please describe the purpose of the new feature or describe the problem to solve.

_pivot_-al feature to add!

### Suggest a solution if possible.

_No response_

### If you have tried alternatives, please describe them below.

_No response_

### Additional information that may help us understand your needs.

Challenges:
- Regarding StableAPI, polars pivot would bring the first challenge in supporting two different APIs [v1.0](https://docs.pola.rs/api/python/stable/reference/dataframe/api/polars.DataFrame.pivot.html) vs [v0.20](https://docs.pola.rs/api/python/version/0.20/reference/dataframe/api/polars.DataFrame.pivot.html) 
- I am not sure that pandas-like `pivot_table` does support all the `aggregate_function`s out of the box 

Discussion:
FBruzzesi: @MarcoGorelli I am considering closing this one for now.
 
There a couple of things that strongly bug me:

1. Multiple columns in polars have a very custom layout, hard to reproduce in practice from pandas
    ```py
    import polars as pl

    data = {
        "ix": [1, 2, 1, 1, 2, 2],
        "col": ["b", "b", "a", "a", "a", "a"],
        "col_b": ["x", "y", "x", "y", "x", "y"],
        "foo": [7, 1, 0, 1, 2, 2],
        "bar": [9, 4, 0, 2, 0, 0],
    }

    df= pl.DataFrame(data)
    df_pl.pivot(on=["col", "col_b"], index="ix", values=["foo", "bar"], aggregate_function="len").columns
    ```
    > ['ix', 'foo_{"b","x"}', 'foo_{"b","y"}', 'foo_{"a","x"}', 'foo_{"a","y"}', 'bar_{"b","x"}', 'bar_{"b","y"}', 'bar_{"a","x"}', 'bar_{"a","y"}']
2. The other issue is that polars behavior changed a lot both from 0.20 and 1.x, as well as from minor 0.20.x versions.

While for point 1 it would be ok to just say we have partial support and so on, for point 2 something like #461 feels like a stronger requirement. For now I implemented the logic in narhwals DataFrame just to make an example of how it would work.

WDYT?
FBruzzesi: Having some troubles with regression on polars pre 1.0 on output column names 🙈
FBruzzesi: I am closing this for now.

Supporting polars<1.0 is quite hard as the API changed so significantly that mapping from one to another for all the edge cases is quite too complex.

A doable proposal could be to only allow for `on` and `index` to not take multiple values.

Additional, modin is now failing while it wasn't before 🥲

Patch:
diff --git a/docs/api-reference/dataframe.md b/docs/api-reference/dataframe.md
index 447acbc1..00ff2122 100644
--- a/docs/api-reference/dataframe.md
+++ b/docs/api-reference/dataframe.md
@@ -26,6 +26,7 @@
         - lazy
         - null_count
         - pipe
+        - pivot
         - rename
         - row
         - rows
diff --git a/narwhals/_pandas_like/dataframe.py b/narwhals/_pandas_like/dataframe.py
index 3bf60f84..25a5c236 100644
--- a/narwhals/_pandas_like/dataframe.py
+++ b/narwhals/_pandas_like/dataframe.py
@@ -827,6 +827,90 @@ class PandasLikeDataFrame:
     def gather_every(self: Self, n: int, offset: int = 0) -> Self:
         return self._from_native_frame(self._native_frame.iloc[offset::n])
 
+    def pivot(
+        self: Self,
+        on: str | list[str],
+        *,
+        index: str | list[str] | None,
+        values: str | list[str] | None,
+        aggregate_function: Any | None,
+        maintain_order: bool,
+        sort_columns: bool,
+        separator: str = "_",
+    ) -> Self:
+        if self._implementation is Implementation.PANDAS and (
+            self._backend_version < (1, 1)
+        ):  # pragma: no cover
+            msg = "pivot is only supported for pandas>=1.1"
+            raise NotImplementedError(msg)
+        if self._implementation is Implementation.MODIN:
+            msg = "pivot is not supported for Modin backend due to https://github.com/modin-project/modin/issues/7409."
+            raise NotImplementedError(msg)
+        from itertools import product
+
+        frame = self._native_frame
+
+        if isinstance(on, str):
+            on = [on]
+        if isinstance(index, str):
+            index = [index]
+
+        if values is None:
+            values_ = [c for c in self.columns if c not in {*on, *index}]  # type: ignore[misc]
+        elif isinstance(values, str):  # pragma: no cover
+            values_ = [values]
+        else:
+            values_ = values
+
+        if aggregate_function is None:
+            result = frame.pivot(columns=on, index=index, values=values_)
+
+        elif aggregate_function == "len":
+            result = (
+                frame.groupby([*on, *index])  # type: ignore[misc]
+                .agg({v: "size" for v in values_})
+                .reset_index()
+                .pivot(columns=on, index=index, values=values_)
+            )
+        else:
+            result = frame.pivot_table(
+                values=values_,
+                index=index,
+                columns=on,
+                aggfunc=aggregate_function,
+                margins=False,
+                observed=True,
+            )
+
+        # Put columns in the right order
+        if sort_columns:
+            uniques = {
+                col: sorted(self._native_frame[col].unique().tolist()) for col in on
+            }
+        else:
+            uniques = {col: self._native_frame[col].unique().tolist() for col in on}
+        all_lists = [values_, *list(uniques.values())]
+        ordered_cols = list(product(*all_lists))
+        result = result.loc[:, ordered_cols]
+        columns = result.columns.tolist()
+
+        n_on = len(on)
+        if n_on == 1:
+            new_columns = [
+                separator.join(col).strip() if len(values_) > 1 else col[-1]
+                for col in columns
+            ]
+        else:
+            new_columns = [
+                separator.join([col[0], '{"' + '","'.join(col[-n_on:]) + '"}'])
+                if len(values_) > 1
+                else '{"' + '","'.join(col[-n_on:]) + '"}'
+                for col in columns
+            ]
+        result.columns = new_columns
+        result.columns.names = [""]  # type: ignore[attr-defined]
+        return self._from_native_frame(result.reset_index())
+
     def to_arrow(self: Self) -> Any:
         if self._implementation is Implementation.CUDF:
             return self._native_frame.to_arrow(preserve_index=False)
diff --git a/narwhals/_polars/dataframe.py b/narwhals/_polars/dataframe.py
index b5640493..fffe5092 100644
--- a/narwhals/_polars/dataframe.py
+++ b/narwhals/_polars/dataframe.py
@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 from typing import Any
+from typing import Literal
 from typing import Sequence
 
 from narwhals._polars.namespace import PolarsNamespace
@@ -16,14 +17,16 @@ if TYPE_CHECKING:
     from types import ModuleType
 
     import numpy as np
+    import polars as pl
     from typing_extensions import Self
 
+    from narwhals.dtypes import DType
     from narwhals.typing import DTypes
 
 
 class PolarsDataFrame:
     def __init__(
-        self, df: Any, *, backend_version: tuple[int, ...], dtypes: DTypes
+        self, df: pl.DataFrame, *, backend_version: tuple[int, ...], dtypes: DTypes
     ) -> None:
         self._native_frame = df
         self._backend_version = backend_version
@@ -91,19 +94,21 @@ class PolarsDataFrame:
             return self._native_frame.__array__(dtype)
         return self._native_frame.__array__(dtype)
 
-    def collect_schema(self) -> dict[str, Any]:
+    def collect_schema(self) -> dict[str, DType]:
         if self._backend_version < (1,):
-            schema = self._native_frame.schema
+            return {
+                name: native_to_narwhals_dtype(dtype, self._dtypes)
+                for name, dtype in self._native_frame.schema.items()
+            }
         else:
-            schema = dict(self._native_frame.collect_schema())
-        return {
-            name: native_to_narwhals_dtype(dtype, self._dtypes)
-            for name, dtype in schema.items()
-        }
+            return {
+                name: native_to_narwhals_dtype(dtype, self._dtypes)
+                for name, dtype in self._native_frame.collect_schema().items()
+            }
 
     @property
     def shape(self) -> tuple[int, int]:
-        return self._native_frame.shape  # type: ignore[no-any-return]
+        return self._native_frame.shape
 
     def __getitem__(self, item: Any) -> Any:
         if self._backend_version > (0, 20, 30):
@@ -179,7 +184,7 @@ class PolarsDataFrame:
 
     @property
     def columns(self) -> list[str]:
-        return self._native_frame.columns  # type: ignore[no-any-return]
+        return self._native_frame.columns
 
     def lazy(self) -> PolarsLazyFrame:
         return PolarsLazyFrame(
@@ -243,10 +248,38 @@ class PolarsDataFrame:
             )
         )
 
+    def pivot(
+        self: Self,
+        on: str | list[str],
+        *,
+        index: str | list[str] | None = None,
+        values: str | list[str] | None = None,
+        aggregate_function: Literal[
+            "min", "max", "first", "last", "sum", "mean", "median", "len"
+        ]
+        | None = None,
+        maintain_order: bool = True,
+        sort_columns: bool = False,
+        separator: str = "_",
+    ) -> Self:
+        if self._backend_version < (1, 0, 0):  # pragma: no cover
+            msg = "`pivot` is only supported for Polars>=1.0.0"
+            raise NotImplementedError(msg)
+        result = self._native_frame.pivot(
+            on,
+            index=index,
+            values=values,
+            aggregate_function=aggregate_function,
+            maintain_order=maintain_order,
+            sort_columns=sort_columns,
+            separator=separator,
+        )
+        return self._from_native_object(result)  # type: ignore[no-any-return]
+
 
 class PolarsLazyFrame:
     def __init__(
-        self, df: Any, *, backend_version: tuple[int, ...], dtypes: DTypes
+        self, df: pl.LazyFrame, *, backend_version: tuple[int, ...], dtypes: DTypes
     ) -> None:
         self._native_frame = df
         self._backend_version = backend_version
@@ -285,7 +318,7 @@ class PolarsLazyFrame:
 
     @property
     def columns(self) -> list[str]:
-        return self._native_frame.columns  # type: ignore[no-any-return]
+        return self._native_frame.columns
 
     @property
     def schema(self) -> dict[str, Any]:
@@ -295,15 +328,17 @@ class PolarsLazyFrame:
             for name, dtype in schema.items()
         }
 
-    def collect_schema(self) -> dict[str, Any]:
+    def collect_schema(self) -> dict[str, DType]:
         if self._backend_version < (1,):
-            schema = self._native_frame.schema
+            return {
+                name: native_to_narwhals_dtype(dtype, self._dtypes)
+                for name, dtype in self._native_frame.schema.items()
+            }
         else:
-            schema = dict(self._native_frame.collect_schema())
-        return {
-            name: native_to_narwhals_dtype(dtype, self._dtypes)
-            for name, dtype in schema.items()
-        }
+            return {
+                name: native_to_narwhals_dtype(dtype, self._dtypes)
+                for name, dtype in self._native_frame.collect_schema().items()
+            }
 
     def collect(self) -> PolarsDataFrame:
         return PolarsDataFrame(
diff --git a/narwhals/_polars/expr.py b/narwhals/_polars/expr.py
index 7aa0fb29..cb2e41f2 100644
--- a/narwhals/_polars/expr.py
+++ b/narwhals/_polars/expr.py
@@ -45,8 +45,8 @@ class PolarsExpr:
 
     def cast(self, dtype: DType) -> Self:
         expr = self._native_expr
-        dtype = narwhals_to_native_dtype(dtype, self._dtypes)
-        return self._from_native_expr(expr.cast(dtype))
+        dtype_pl = narwhals_to_native_dtype(dtype, self._dtypes)
+        return self._from_native_expr(expr.cast(dtype_pl))
 
     def map_batches(
         self,
@@ -54,9 +54,9 @@ class PolarsExpr:
         return_dtype: DType | None = None,
     ) -> Self:
         if return_dtype is not None:
-            return_dtype = narwhals_to_native_dtype(return_dtype, self._dtypes)
+            return_dtype_pl = narwhals_to_native_dtype(return_dtype, self._dtypes)
             return self._from_native_expr(
-                self._native_expr.map_batches(function, return_dtype)
+                self._native_expr.map_batches(function, return_dtype_pl)
             )
         else:
             return self._from_native_expr(self._native_expr.map_batches(function))
@@ -65,14 +65,14 @@ class PolarsExpr:
         self, old: Sequence[Any], new: Sequence[Any], *, return_dtype: DType | None
     ) -> Self:
         expr = self._native_expr
-        return_dtype = (
+        return_dtype_pl = (
             narwhals_to_native_dtype(return_dtype, self._dtypes) if return_dtype else None
         )
         if self._backend_version < (1,):
             msg = f"`replace_strict` is only available in Polars>=1.0, found version {self._backend_version}"
             raise NotImplementedError(msg)
         return self._from_native_expr(
-            expr.replace_strict(old, new, return_dtype=return_dtype)
+            expr.replace_strict(old, new, return_dtype=return_dtype_pl)
         )
 
     def __eq__(self, other: object) -> Self:  # type: ignore[override]
diff --git a/narwhals/_polars/series.py b/narwhals/_polars/series.py
index 02b75520..4feff11c 100644
--- a/narwhals/_polars/series.py
+++ b/narwhals/_polars/series.py
@@ -103,8 +103,8 @@ class PolarsSeries:
 
     def cast(self, dtype: DType) -> Self:
         ser = self._native_series
-        dtype = narwhals_to_native_dtype(dtype, self._dtypes)
-        return self._from_native_series(ser.cast(dtype))  # type: ignore[arg-type]
+        dtype_pl = narwhals_to_native_dtype(dtype, self._dtypes)
+        return self._from_native_series(ser.cast(dtype_pl))
 
     def replace_strict(
         self, old: Sequence[Any], new: Sequence[Any], *, return_dtype: DType | None
diff --git a/narwhals/_polars/utils.py b/narwhals/_polars/utils.py
index 295c03bc..79528af1 100644
--- a/narwhals/_polars/utils.py
+++ b/narwhals/_polars/utils.py
@@ -5,10 +5,11 @@ from typing import Any
 from typing import Literal
 
 if TYPE_CHECKING:
+    import polars as pl
+
     from narwhals.dtypes import DType
     from narwhals.typing import DTypes
 
-from narwhals.dependencies import get_polars
 from narwhals.utils import parse_version
 
 
@@ -33,7 +34,7 @@ def extract_args_kwargs(args: Any, kwargs: Any) -> tuple[list[Any], dict[str, An
     return args, kwargs
 
 
-def native_to_narwhals_dtype(dtype: Any, dtypes: DTypes) -> DType:
+def native_to_narwhals_dtype(dtype: pl.DataType, dtypes: DTypes) -> DType:
     import polars as pl  # ignore-banned-import()
 
     if dtype == pl.Float64:
@@ -79,25 +80,26 @@ def native_to_narwhals_dtype(dtype: Any, dtypes: DTypes) -> DType:
         return dtypes.Struct(
             [
                 dtypes.Field(field_name, native_to_narwhals_dtype(field_type, dtypes))
-                for field_name, field_type in dtype
+                for field_name, field_type in dtype  # type: ignore[attr-defined]
             ]
         )
     if dtype == pl.List:
-        return dtypes.List(native_to_narwhals_dtype(dtype.inner, dtypes))
+        return dtypes.List(native_to_narwhals_dtype(dtype.inner, dtypes))  # type: ignore[attr-defined]
     if dtype == pl.Array:
         if parse_version(pl.__version__) < (0, 20, 30):  # pragma: no cover
             return dtypes.Array(
-                native_to_narwhals_dtype(dtype.inner, dtypes), dtype.width
+                native_to_narwhals_dtype(dtype.inner, dtypes),  # type: ignore[attr-defined]
+                dtype.width,  # type: ignore[attr-defined]
             )
         else:
-            return dtypes.Array(native_to_narwhals_dtype(dtype.inner, dtypes), dtype.size)
+            return dtypes.Array(native_to_narwhals_dtype(dtype.inner, dtypes), dtype.size)  # type: ignore[attr-defined]
     return dtypes.Unknown()
 
 
-def narwhals_to_native_dtype(dtype: DType | type[DType], dtypes: DTypes) -> Any:
-    if (pl := get_polars()) is not None and isinstance(
-        dtype, (pl.DataType, pl.DataType.__class__)
-    ):
+def narwhals_to_native_dtype(dtype: DType | type[DType], dtypes: DTypes) -> pl.DataType:
+    import polars as pl  # ignore-banned-import()
+
+    if isinstance(dtype, (pl.DataType, pl.DataType.__class__)):  # type: ignore[arg-type]
         msg = (
             f"Expected Narwhals object, got: {type(dtype)}.\n\n"
             "Perhaps you:\n"
@@ -106,8 +108,6 @@ def narwhals_to_native_dtype(dtype: DType | type[DType], dtypes: DTypes) -> Any:
         )
         raise TypeError(msg)
 
-    import polars as pl  # ignore-banned-import()
-
     if dtype == dtypes.Float64:
         return pl.Float64()
     if dtype == dtypes.Float32:
@@ -142,7 +142,7 @@ def narwhals_to_native_dtype(dtype: DType | type[DType], dtypes: DTypes) -> Any:
     if dtype == dtypes.Date:
         return pl.Date()
     if dtype == dtypes.Datetime or isinstance(dtype, dtypes.Datetime):
-        dt_time_unit = getattr(dtype, "time_unit", "us")
+        dt_time_unit: Literal["ms", "us", "ns"] = getattr(dtype, "time_unit", "us")
         dt_time_zone = getattr(dtype, "time_zone", None)
         return pl.Datetime(dt_time_unit, dt_time_zone)
     if dtype == dtypes.Duration or isinstance(dtype, dtypes.Duration):
@@ -151,13 +151,13 @@ def narwhals_to_native_dtype(dtype: DType | type[DType], dtypes: DTypes) -> Any:
 
     if dtype == dtypes.List:  # pragma: no cover
         msg = "Converting to List dtype is not supported yet"
-        return NotImplementedError(msg)
+        raise NotImplementedError(msg)
     if dtype == dtypes.Struct:  # pragma: no cover
         msg = "Converting to Struct dtype is not supported yet"
-        return NotImplementedError(msg)
+        raise NotImplementedError(msg)
     if dtype == dtypes.Array:  # pragma: no cover
         msg = "Converting to Array dtype is not supported yet"
-        return NotImplementedError(msg)
+        raise NotImplementedError(msg)
     return pl.Unknown()  # pragma: no cover
 
 
diff --git a/narwhals/dataframe.py b/narwhals/dataframe.py
index bb163b28..34482072 100644
--- a/narwhals/dataframe.py
+++ b/narwhals/dataframe.py
@@ -2567,6 +2567,91 @@ class DataFrame(BaseFrame[DataFrameT]):
         """
         return super().gather_every(n=n, offset=offset)
 
+    def pivot(
+        self: Self,
+        on: str | list[str],
+        *,
+        index: str | list[str] | None = None,
+        values: str | list[str] | None = None,
+        aggregate_function: Literal[
+            "min", "max", "first", "last", "sum", "mean", "median", "len"
+        ]
+        | None = None,
+        maintain_order: bool = True,
+        sort_columns: bool = False,
+        separator: str = "_",
+    ) -> Self:
+        r"""
+        Create a spreadsheet-style pivot table as a DataFrame.
+
+        Arguments:
+            on: Name of the column(s) whose values will be used as the header of the
+                output DataFrame.
+            index: One or multiple keys to group by. If None, all remaining columns not
+                specified on `on` and `values` will be used. At least one of `index` and
+                `values` must be specified.
+            values: One or multiple keys to group by. If None, all remaining columns not
+                specified on `on` and `index` will be used. At least one of `index` and
+                `values` must be specified.
+            aggregate_function: Choose from:
+                - None: no aggregation takes place, will raise error if multiple values
+                    are in group.
+                - A predefined aggregate function string, one of
+                    {'min', 'max', 'first', 'last', 'sum', 'mean', 'median', 'len'}
+            maintain_order: Sort the grouped keys so that the output order is predictable.
+            sort_columns: Sort the transposed columns by name. Default is by order of
+                discovery.
+            separator: Used as separator/delimiter in generated column names in case of
+                multiple `values` columns.
+
+        Examples:
+            >>> import narwhals as nw
+            >>> import pandas as pd
+            >>> import polars as pl
+            >>> data = {
+            ...     "ix": [1, 1, 2, 2, 1, 2],
+            ...     "col": ["a", "a", "a", "a", "b", "b"],
+            ...     "foo": [0, 1, 2, 2, 7, 1],
+            ...     "bar": [0, 2, 0, 0, 9, 4],
+            ... }
+            >>> df_pd = pd.DataFrame(data)
+            >>> df_pl = pl.DataFrame(data)
+
+            Let's define a dataframe-agnostic function:
+
+            >>> @nw.narwhalify
+            ... def func(df):
+            ...     return df.pivot("col", index="ix", aggregate_function="sum")
+
+            We can then pass either pandas or Polars to `func`:
+
+            >>> func(df_pd)
+               ix  foo_a  foo_b  bar_a  bar_b
+            0   1      1      7      2      9
+            1   2      4      1      0      4
+            >>> func(df_pl)
+            shape: (2, 5)
+            ┌─────┬───────┬───────┬───────┬───────┐
+            │ ix  ┆ foo_a ┆ foo_b ┆ bar_a ┆ bar_b │
+            │ --- ┆ ---   ┆ ---   ┆ ---   ┆ ---   │
+            │ i64 ┆ i64   ┆ i64   ┆ i64   ┆ i64   │
+            ╞═════╪═══════╪═══════╪═══════╪═══════╡
+            │ 1   ┆ 1     ┆ 7     ┆ 2     ┆ 9     │
+            │ 2   ┆ 4     ┆ 1     ┆ 0     ┆ 4     │
+            └─────┴───────┴───────┴───────┴───────┘
+        """
+        return self._from_compliant_dataframe(
+            self._compliant_frame.pivot(
+                on=on,
+                index=index,
+                values=values,
+                aggregate_function=aggregate_function,
+                maintain_order=maintain_order,
+                sort_columns=sort_columns,
+                separator=separator,
+            )
+        )
+
     def to_arrow(self: Self) -> pa.Table:
         r"""
         Convert to arrow table.
diff --git a/narwhals/functions.py b/narwhals/functions.py
index 3fd2025a..d00d5941 100644
--- a/narwhals/functions.py
+++ b/narwhals/functions.py
@@ -233,9 +233,11 @@ def _new_series_impl(
                 narwhals_to_native_dtype as polars_narwhals_to_native_dtype,
             )
 
-            dtype = polars_narwhals_to_native_dtype(dtype, dtypes=dtypes)
+            dtype_pl = polars_narwhals_to_native_dtype(dtype, dtypes=dtypes)
+        else:
+            dtype_pl = None
 
-        native_series = native_namespace.Series(name=name, values=values, dtype=dtype)
+        native_series = native_namespace.Series(name=name, values=values, dtype=dtype_pl)
     elif implementation in {
         Implementation.PANDAS,
         Implementation.MODIN,
@@ -376,12 +378,14 @@ def _from_dict_impl(
                 narwhals_to_native_dtype as polars_narwhals_to_native_dtype,
             )
 
-            schema = {
+            schema_pl = {
                 name: polars_narwhals_to_native_dtype(dtype, dtypes=dtypes)
                 for name, dtype in schema.items()
             }
+        else:
+            schema_pl = None
 
-        native_frame = native_namespace.from_dict(data, schema=schema)
+        native_frame = native_namespace.from_dict(data, schema=schema_pl)
     elif implementation in {
         Implementation.PANDAS,
         Implementation.MODIN,
diff --git a/narwhals/utils.py b/narwhals/utils.py
index 85ec5a00..e1a7b812 100644
--- a/narwhals/utils.py
+++ b/narwhals/utils.py
@@ -538,12 +538,11 @@ def is_ordered_categorical(series: Series) -> bool:
 
 
 def generate_unique_token(n_bytes: int, columns: list[str]) -> str:  # pragma: no cover
-    warn(
+    msg = (
         "Use `generate_temporary_column_name` instead. `generate_unique_token` is "
-        "deprecated and it will be removed in future versions",
-        DeprecationWarning,
-        stacklevel=2,
+        "deprecated and it will be removed in future versions"
     )
+    issue_deprecation_warning(msg, _version="1.13.0")
     return generate_temporary_column_name(n_bytes=n_bytes, columns=columns)
 
 
diff --git a/pyproject.toml b/pyproject.toml
index 65322100..9f20bcf7 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -65,6 +65,7 @@ lint.ignore = [
   "ISC001",
   "NPY002",
   "PD901",  # This is a auxiliary library so dataframe variables have no concrete business meaning
+  "PD010",
   "PLR0911",
   "PLR0912",
   "PLR0913",
diff --git a/utils/generate_backend_completeness.py b/utils/generate_backend_completeness.py
index d7d05daa..2aafc567 100644
--- a/utils/generate_backend_completeness.py
+++ b/utils/generate_backend_completeness.py
@@ -68,7 +68,7 @@ def render_table_and_write_to_output(
     results: list[pl.DataFrame], title: str, output_filename: str
 ) -> None:
     results = (
-        pl.concat(results)  # noqa: PD010
+        pl.concat(results)
         .with_columns(supported=pl.lit(":white_check_mark:"))
         .pivot(on="Backend", values="supported", index=["Class", "Method"])
         .filter(pl.col("narwhals").is_not_null())


Test patch:
diff --git a/tests/frame/pivot_test.py b/tests/frame/pivot_test.py
new file mode 100644
index 00000000..76217361
--- /dev/null
+++ b/tests/frame/pivot_test.py
@@ -0,0 +1,240 @@
+from __future__ import annotations
+
+from contextlib import nullcontext as does_not_raise
+from typing import Any
+
+import polars as pl
+import pytest
+
+import narwhals.stable.v1 as nw
+from tests.utils import PANDAS_VERSION
+from tests.utils import POLARS_VERSION
+from tests.utils import assert_equal_data
+
+data = {
+    "ix": [1, 2, 1, 1, 2, 2],
+    "iy": [1, 2, 2, 1, 2, 1],
+    "col": ["b", "b", "a", "a", "a", "a"],
+    "col_b": ["x", "y", "x", "y", "x", "y"],
+    "foo": [7, 1, 0, 1, 2, 2],
+    "bar": [9, 4, 0, 2, 0, 0],
+}
+
+data_no_dups = {
+    "ix": [1, 1, 2, 2],
+    "col": ["a", "b", "a", "b"],
+    "foo": [1, 2, 3, 4],
+    "bar": ["x", "y", "z", "w"],
+}
+
+
+@pytest.mark.parametrize(
+    ("agg_func", "expected"),
+    [
+        (
+            "min",
+            {
+                "ix": [1, 2],
+                "foo_a": [0, 2],
+                "foo_b": [7, 1],
+                "bar_a": [0, 0],
+                "bar_b": [9, 4],
+            },
+        ),
+        (
+            "max",
+            {
+                "ix": [1, 2],
+                "foo_a": [1, 2],
+                "foo_b": [7, 1],
+                "bar_a": [2, 0],
+                "bar_b": [9, 4],
+            },
+        ),
+        (
+            "first",
+            {
+                "ix": [1, 2],
+                "foo_a": [0, 2],
+                "foo_b": [7, 1],
+                "bar_a": [0, 0],
+                "bar_b": [9, 4],
+            },
+        ),
+        (
+            "last",
+            {
+                "ix": [1, 2],
+                "foo_a": [1, 2],
+                "foo_b": [7, 1],
+                "bar_a": [2, 0],
+                "bar_b": [9, 4],
+            },
+        ),
+        (
+            "sum",
+            {
+                "ix": [1, 2],
+                "foo_a": [1, 4],
+                "foo_b": [7, 1],
+                "bar_a": [2, 0],
+                "bar_b": [9, 4],
+            },
+        ),
+        (
+            "mean",
+            {
+                "ix": [1, 2],
+                "foo_a": [0.5, 2.0],
+                "foo_b": [7.0, 1.0],
+                "bar_a": [1.0, 0.0],
+                "bar_b": [9.0, 4.0],
+            },
+        ),
+        (
+            "median",
+            {
+                "ix": [1, 2],
+                "foo_a": [0.5, 2.0],
+                "foo_b": [7.0, 1.0],
+                "bar_a": [1.0, 0.0],
+                "bar_b": [9.0, 4.0],
+            },
+        ),
+        (
+            "len",
+            {
+                "ix": [1, 2],
+                "foo_a": [2, 2],
+                "foo_b": [1, 1],
+                "bar_a": [2, 2],
+                "bar_b": [1, 1],
+            },
+        ),
+    ],
+)
+@pytest.mark.parametrize(("on", "index"), [("col", "ix"), (["col"], ["ix"])])
+def test_pivot(
+    constructor_eager: Any,
+    agg_func: str,
+    expected: dict[str, list[Any]],
+    on: str | list[str],
+    index: str | list[str],
+    request: pytest.FixtureRequest,
+) -> None:
+    if any(x in str(constructor_eager) for x in ("pyarrow_table", "modin")):
+        request.applymarker(pytest.mark.xfail)
+    if ("polars" in str(constructor_eager) and POLARS_VERSION < (1, 0)) or (
+        "pandas" in str(constructor_eager) and PANDAS_VERSION < (1, 1)
+    ):
+        # not implemented
+        request.applymarker(pytest.mark.xfail)
+
+    df = nw.from_native(constructor_eager(data), eager_only=True)
+    result = df.pivot(
+        on=on,
+        index=index,
+        values=["foo", "bar"],
+        aggregate_function=agg_func,  # type: ignore[arg-type]
+    )
+
+    assert_equal_data(result, expected)
+
+
+@pytest.mark.parametrize(
+    ("data_", "context"),
+    [
+        (data_no_dups, does_not_raise()),
+        (data, pytest.raises((ValueError, pl.exceptions.ComputeError))),
+    ],
+)
+def test_pivot_no_agg(
+    request: Any, constructor_eager: Any, data_: Any, context: Any
+) -> None:
+    if any(x in str(constructor_eager) for x in ("pyarrow_table", "modin")):
+        request.applymarker(pytest.mark.xfail)
+    if ("polars" in str(constructor_eager) and POLARS_VERSION < (1, 0)) or (
+        "pandas" in str(constructor_eager) and PANDAS_VERSION < (1, 1)
+    ):
+        # not implemented
+        request.applymarker(pytest.mark.xfail)
+
+    df = nw.from_native(constructor_eager(data_), eager_only=True)
+    with context:
+        df.pivot("col", index="ix", aggregate_function=None)
+
+
+@pytest.mark.parametrize(
+    ("sort_columns", "expected"),
+    [
+        (True, ["ix", "foo_a", "foo_b", "bar_a", "bar_b"]),
+        (False, ["ix", "foo_b", "foo_a", "bar_b", "bar_a"]),
+    ],
+)
+def test_pivot_sort_columns(
+    request: Any, constructor_eager: Any, sort_columns: Any, expected: list[str]
+) -> None:
+    if any(x in str(constructor_eager) for x in ("pyarrow_table", "modin")):
+        request.applymarker(pytest.mark.xfail)
+    if ("polars" in str(constructor_eager) and POLARS_VERSION < (1, 0)) or (
+        "pandas" in str(constructor_eager) and PANDAS_VERSION < (1, 1)
+    ):
+        # not implemented
+        request.applymarker(pytest.mark.xfail)
+
+    df = nw.from_native(constructor_eager(data), eager_only=True)
+    result = df.pivot(
+        on="col",
+        index="ix",
+        values=["foo", "bar"],
+        aggregate_function="sum",
+        sort_columns=sort_columns,
+    )
+    assert result.columns == expected
+
+
+@pytest.mark.parametrize(
+    ("kwargs", "expected"),
+    [
+        ({"on": ["col"], "values": ["foo"]}, ["ix", "b", "a"]),
+        (
+            {"on": ["col"], "values": ["foo", "bar"]},
+            ["ix", "foo_b", "foo_a", "bar_b", "bar_a"],
+        ),
+        (
+            {"on": ["col", "col_b"], "values": ["foo"]},
+            ["ix", '{"b","x"}', '{"b","y"}', '{"a","x"}', '{"a","y"}'],
+        ),
+        (
+            {"on": ["col", "col_b"], "values": ["foo", "bar"]},
+            [
+                "ix",
+                'foo_{"b","x"}',
+                'foo_{"b","y"}',
+                'foo_{"a","x"}',
+                'foo_{"a","y"}',
+                'bar_{"b","x"}',
+                'bar_{"b","y"}',
+                'bar_{"a","x"}',
+                'bar_{"a","y"}',
+            ],
+        ),
+    ],
+)
+def test_pivot_names_out(
+    request: Any, constructor_eager: Any, kwargs: Any, expected: list[str]
+) -> None:
+    if any(x in str(constructor_eager) for x in ("pyarrow_table", "modin")):
+        request.applymarker(pytest.mark.xfail)
+    if ("polars" in str(constructor_eager) and POLARS_VERSION < (1, 0)) or (
+        "pandas" in str(constructor_eager) and PANDAS_VERSION < (1, 1)
+    ):
+        # not implemented
+        request.applymarker(pytest.mark.xfail)
+
+    df = nw.from_native(constructor_eager(data), eager_only=True)
+
+    result = (
+        df.pivot(aggregate_function="min", index="ix", **kwargs).collect_schema().names()
+    )
+    assert result == expected
diff --git a/tests/series_only/to_dummy_test.py b/tests/series_only/to_dummy_test.py
index ce25dd65..9ebbee57 100644
--- a/tests/series_only/to_dummy_test.py
+++ b/tests/series_only/to_dummy_test.py
@@ -40,11 +40,7 @@ def test_to_dummies_with_nulls(constructor_eager: ConstructorEager, sep: str) ->
 
 
 @pytest.mark.parametrize("sep", ["_", "-"])
-def test_to_dummies_drop_first_na(
-    request: pytest.FixtureRequest, constructor_eager: ConstructorEager, sep: str
-) -> None:
-    if "cudf" in str(constructor_eager):
-        request.applymarker(pytest.mark.xfail)
+def test_to_dummies_drop_first_na(constructor_eager: ConstructorEager, sep: str) -> None:
     s = nw.from_native(constructor_eager({"a": data_na}), eager_only=True)["a"].alias("a")
     result = s.to_dummies(drop_first=True, separator=sep)
     expected = {f"a{sep}null": [0, 0, 1], f"a{sep}y": [0, 1, 0]}