Repository: narwhals-dev/narwhals
Pull request #2223 (2025-03-16 13:08:36)

chore: Spec `CompliantDataFrame`
## What type of PR is this? (check all applicable)

- [ ] 💾 Refactor
- [ ] ✨ Feature
- [ ] 🐛 Bug Fix
- [ ] 🔧 Optimization
- [x] 📝 Documentation
- [ ] ✅ Test
- [x] 🐳 Other

## Related issues

- #2202
- #2119
- #2064

## Checklist

- [x] Code follows style guide (ruff)
- [ ] Tests added
- [x] Documented the changes

## If you have comments or can explain your changes, please do so below
- [x] Decide on `PolarsDataFrame.__getattr__` annotations
- [x] Implement `CompliantDataFrame.aggregate` (https://github.com/narwhals-dev/narwhals/pull/2223/commits/4dc0514941065373a129f6b1a6e8c663b9a9fddf)
  - See [thread](https://github.com/narwhals-dev/narwhals/pull/2223#discussion_r1997628660)
- [x] Fix (`FBT001`) issues (https://github.com/narwhals-dev/narwhals/pull/2223/commits/70cc069a0dac5206f8618d40aa5de8e50034df7c)
- [x] Using `CompliantExpr` in `CompliantDataFrame`
  - https://github.com/narwhals-dev/narwhals/pull/2223/commits/deb77c2b0e4ffe56f02aed9ff863fb022088622a
  - https://github.com/narwhals-dev/narwhals/pull/2223/commits/c9b57a20b2fc0e2dcab6e5d3fc65f1f07d69af8e
  - https://github.com/narwhals-dev/narwhals/pull/2223/commits/7198e9576d2af160a54b4d51cb91288dd4dc8e0a


Patch:
diff --git a/narwhals/_arrow/dataframe.py b/narwhals/_arrow/dataframe.py
index 59be276f..e85abbab 100644
--- a/narwhals/_arrow/dataframe.py
+++ b/narwhals/_arrow/dataframe.py
@@ -5,6 +5,7 @@ from typing import TYPE_CHECKING
 from typing import Any
 from typing import Iterator
 from typing import Literal
+from typing import Mapping
 from typing import Sequence
 from typing import cast
 from typing import overload
@@ -26,6 +27,7 @@ from narwhals.utils import check_column_exists
 from narwhals.utils import check_column_names_are_unique
 from narwhals.utils import generate_temporary_column_name
 from narwhals.utils import is_sequence_but_not_str
+from narwhals.utils import not_implemented
 from narwhals.utils import parse_columns_to_drop
 from narwhals.utils import parse_version
 from narwhals.utils import scale_bytes
@@ -71,7 +73,7 @@ from narwhals.typing import CompliantDataFrame
 from narwhals.typing import CompliantLazyFrame
 
 
-class ArrowDataFrame(EagerDataFrame["ArrowSeries"], CompliantLazyFrame):
+class ArrowDataFrame(EagerDataFrame["ArrowSeries", "ArrowExpr"], CompliantLazyFrame):
     # --- not in the spec ---
     def __init__(
         self: Self,
@@ -356,9 +358,6 @@ class ArrowDataFrame(EagerDataFrame["ArrowSeries"], CompliantLazyFrame):
             self._native_frame.select(list(column_names)), validate_column_names=False
         )
 
-    def aggregate(self: ArrowDataFrame, *exprs: ArrowExpr) -> ArrowDataFrame:
-        return self.select(*exprs)
-
     def select(self: ArrowDataFrame, *exprs: ArrowExpr) -> ArrowDataFrame:
         new_series = self._evaluate_into_exprs(*exprs)
         if not new_series:
@@ -449,21 +448,9 @@ class ArrowDataFrame(EagerDataFrame["ArrowSeries"], CompliantLazyFrame):
             ),
         )
 
-    def join_asof(
-        self: Self,
-        other: Self,
-        *,
-        left_on: str | None,
-        right_on: str | None,
-        by_left: list[str] | None,
-        by_right: list[str] | None,
-        strategy: Literal["backward", "forward", "nearest"],
-        suffix: str,
-    ) -> Self:
-        msg = "join_asof is not yet supported on PyArrow tables"  # pragma: no cover
-        raise NotImplementedError(msg)
+    join_asof = not_implemented()
 
-    def drop(self: Self, columns: list[str], strict: bool) -> Self:  # noqa: FBT001
+    def drop(self: Self, columns: Sequence[str], *, strict: bool) -> Self:
         to_drop = parse_columns_to_drop(
             compliant_frame=self, columns=columns, strict=strict
         )
@@ -471,7 +458,7 @@ class ArrowDataFrame(EagerDataFrame["ArrowSeries"], CompliantLazyFrame):
             self._native_frame.drop(to_drop), validate_column_names=False
         )
 
-    def drop_nulls(self: ArrowDataFrame, subset: list[str] | None) -> ArrowDataFrame:
+    def drop_nulls(self: ArrowDataFrame, subset: Sequence[str] | None) -> ArrowDataFrame:
         if subset is None:
             return self._from_native_frame(
                 self._native_frame.drop_null(), validate_column_names=False
@@ -629,7 +616,7 @@ class ArrowDataFrame(EagerDataFrame["ArrowSeries"], CompliantLazyFrame):
         self: Self,
         backend: Implementation | None,
         **kwargs: Any,
-    ) -> CompliantDataFrame[Any]:
+    ) -> CompliantDataFrame[Any, Any]:
         if backend is Implementation.PYARROW or backend is None:
             from narwhals._arrow.dataframe import ArrowDataFrame
 
@@ -667,9 +654,7 @@ class ArrowDataFrame(EagerDataFrame["ArrowSeries"], CompliantLazyFrame):
         msg = f"Unsupported `backend` value: {backend}"  # pragma: no cover
         raise AssertionError(msg)  # pragma: no cover
 
-    def clone(self: Self) -> Self:
-        msg = "clone is not yet supported on PyArrow tables"
-        raise NotImplementedError(msg)
+    clone = not_implemented()
 
     def item(self: Self, row: int | None, column: int | str | None) -> Any:
         from narwhals._arrow.series import maybe_extract_py_scalar
@@ -695,7 +680,7 @@ class ArrowDataFrame(EagerDataFrame["ArrowSeries"], CompliantLazyFrame):
             self._native_frame[_col][row], return_py_scalar=True
         )
 
-    def rename(self: Self, mapping: dict[str, str]) -> Self:
+    def rename(self: Self, mapping: Mapping[str, str]) -> Self:
         df = self._native_frame
         new_cols = [mapping.get(c, c) for c in df.column_names]
         return self._from_native_frame(df.rename_columns(new_cols))
@@ -746,7 +731,7 @@ class ArrowDataFrame(EagerDataFrame["ArrowSeries"], CompliantLazyFrame):
 
     def unique(
         self: ArrowDataFrame,
-        subset: list[str] | None,
+        subset: Sequence[str] | None,
         *,
         keep: Literal["any", "first", "last", "none"],
         maintain_order: bool | None = None,
@@ -757,7 +742,7 @@ class ArrowDataFrame(EagerDataFrame["ArrowSeries"], CompliantLazyFrame):
 
         df = self._native_frame
         check_column_exists(self.columns, subset)
-        subset = subset or self.columns
+        subset = list(subset or self.columns)
 
         if keep in {"any", "first", "last"}:
             agg_func_map = {"any": "min", "first": "min", "last": "max"}
@@ -808,18 +793,15 @@ class ArrowDataFrame(EagerDataFrame["ArrowSeries"], CompliantLazyFrame):
 
     def unpivot(
         self: Self,
-        on: list[str] | None,
-        index: list[str] | None,
+        on: Sequence[str] | None,
+        index: Sequence[str] | None,
         variable_name: str,
         value_name: str,
     ) -> Self:
         native_frame = self._native_frame
         n_rows = len(self)
-
-        index_: list[str] = [] if index is None else index
-        on_: list[str] = (
-            [c for c in self.columns if c not in index_] if on is None else on
-        )
+        index_ = [] if index is None else index
+        on_ = [c for c in self.columns if c not in index_] if on is None else on
         concat = (
             partial(pa.concat_tables, promote_options="permissive")
             if self._backend_version >= (14, 0, 0)
diff --git a/narwhals/_arrow/utils.py b/narwhals/_arrow/utils.py
index 2d4f8544..7edbe473 100644
--- a/narwhals/_arrow/utils.py
+++ b/narwhals/_arrow/utils.py
@@ -13,6 +13,7 @@ from typing import overload
 import pyarrow as pa
 import pyarrow.compute as pc
 
+from narwhals.exceptions import ShapeError
 from narwhals.utils import _SeriesNamespace
 from narwhals.utils import import_dtypes_module
 from narwhals.utils import isinstance_or_issubclass
@@ -278,6 +279,9 @@ def extract_dataframe_comparand(
 ) -> ArrowChunkedArray:
     """Extract native Series, broadcasting to `length` if necessary."""
     if not other._broadcast:
+        if (len_other := len(other)) != length:
+            msg = f"Expected object of length {length}, got: {len_other}."
+            raise ShapeError(msg)
         return other.native
 
     import numpy as np  # ignore-banned-import
diff --git a/narwhals/_compliant/__init__.py b/narwhals/_compliant/__init__.py
index a81bd06f..e47d1a9d 100644
--- a/narwhals/_compliant/__init__.py
+++ b/narwhals/_compliant/__init__.py
@@ -19,7 +19,7 @@ from narwhals._compliant.series import EagerSeries
 from narwhals._compliant.typing import CompliantExprT
 from narwhals._compliant.typing import CompliantFrameT
 from narwhals._compliant.typing import CompliantSeriesOrNativeExprT_co
-from narwhals._compliant.typing import CompliantSeriesT_co
+from narwhals._compliant.typing import CompliantSeriesT
 from narwhals._compliant.typing import EagerDataFrameT
 from narwhals._compliant.typing import EagerSeriesT
 from narwhals._compliant.typing import IntoCompliantExpr
@@ -35,7 +35,7 @@ __all__ = [
     "CompliantSelectorNamespace",
     "CompliantSeries",
     "CompliantSeriesOrNativeExprT_co",
-    "CompliantSeriesT_co",
+    "CompliantSeriesT",
     "EagerDataFrame",
     "EagerDataFrameT",
     "EagerExpr",
diff --git a/narwhals/_compliant/dataframe.py b/narwhals/_compliant/dataframe.py
index ed1d83b9..b64c1388 100644
--- a/narwhals/_compliant/dataframe.py
+++ b/narwhals/_compliant/dataframe.py
@@ -4,43 +4,155 @@ from itertools import chain
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Iterator
+from typing import Literal
 from typing import Mapping
 from typing import Protocol
 from typing import Sequence
+from typing import Sized
 from typing import TypeVar
+from typing import overload
 
-from narwhals._compliant.typing import CompliantSeriesT_co
+from narwhals._compliant.typing import CompliantExprT_contra
+from narwhals._compliant.typing import CompliantSeriesT
+from narwhals._compliant.typing import EagerExprT_contra
 from narwhals._compliant.typing import EagerSeriesT
 from narwhals._expression_parsing import evaluate_output_names_and_aliases
 
 if TYPE_CHECKING:
+    from io import BytesIO
+    from pathlib import Path
+
+    import pandas as pd
+    import polars as pl
+    import pyarrow as pa
     from typing_extensions import Self
-    from typing_extensions import TypeIs
+    from typing_extensions import TypeAlias
 
-    from narwhals._compliant.expr import EagerExpr
     from narwhals.dtypes import DType
+    from narwhals.typing import SizeUnit
+    from narwhals.typing import _2DArray
+    from narwhals.utils import Implementation
+
+    Incomplete: TypeAlias = Any
 
 __all__ = ["CompliantDataFrame", "CompliantLazyFrame", "EagerDataFrame"]
 
 T = TypeVar("T")
 
 
-class CompliantDataFrame(Protocol[CompliantSeriesT_co]):
+class CompliantDataFrame(Sized, Protocol[CompliantSeriesT, CompliantExprT_contra]):
     def __narwhals_dataframe__(self) -> Self: ...
     def __narwhals_namespace__(self) -> Any: ...
-    def simple_select(
-        self, *column_names: str
-    ) -> Self: ...  # `select` where all args are column names.
-    def aggregate(self, *exprs: Any) -> Self:  # pragma: no cover
-        ...  # `select` where all args are aggregations or literals
-        # (so, no broadcasting is necessary).
+    def __array__(self, dtype: Any, *, copy: bool | None) -> _2DArray: ...
+    def __getitem__(self, item: Any) -> CompliantSeriesT | Self: ...
+    def simple_select(self, *column_names: str) -> Self:
+        """`select` where all args are column names."""
+        ...
+
+    def aggregate(self, *exprs: CompliantExprT_contra) -> Self:
+        """`select` where all args are aggregations or literals.
+
+        (so, no broadcasting is necessary).
+        """
+        return self.select(*exprs)
 
     @property
     def columns(self) -> Sequence[str]: ...
     @property
     def schema(self) -> Mapping[str, DType]: ...
-    def get_column(self, name: str) -> CompliantSeriesT_co: ...
-    def iter_columns(self) -> Iterator[CompliantSeriesT_co]: ...
+    @property
+    def shape(self) -> tuple[int, int]: ...
+    def clone(self) -> Self: ...
+    def collect(
+        self, backend: Implementation | None, **kwargs: Any
+    ) -> CompliantDataFrame[Any, Any]: ...
+    def collect_schema(self) -> Mapping[str, DType]: ...
+    def drop(self, columns: Sequence[str], *, strict: bool) -> Self: ...
+    def drop_nulls(self, subset: Sequence[str] | None) -> Self: ...
+    def estimated_size(self, unit: SizeUnit) -> int | float: ...
+    def filter(self, predicate: CompliantExprT_contra | Incomplete) -> Self: ...
+    def gather_every(self, n: int, offset: int) -> Self: ...
+    def get_column(self, name: str) -> CompliantSeriesT: ...
+    def group_by(self, *keys: str, drop_null_keys: bool) -> Incomplete: ...
+    def head(self, n: int) -> Self: ...
+    def item(self, row: int | None, column: int | str | None) -> Any: ...
+    def iter_columns(self) -> Iterator[CompliantSeriesT]: ...
+    def iter_rows(
+        self, *, named: bool, buffer_size: int
+    ) -> Iterator[tuple[Any, ...]] | Iterator[Mapping[str, Any]]: ...
+    def is_unique(self) -> CompliantSeriesT: ...
+    def join(
+        self: Self,
+        other: Self,
+        *,
+        how: Literal["left", "inner", "cross", "anti", "semi"],
+        left_on: Sequence[str] | None,
+        right_on: Sequence[str] | None,
+        suffix: str,
+    ) -> Self: ...
+    def join_asof(
+        self: Self,
+        other: Self,
+        *,
+        left_on: str | None,
+        right_on: str | None,
+        by_left: Sequence[str] | None,
+        by_right: Sequence[str] | None,
+        strategy: Literal["backward", "forward", "nearest"],
+        suffix: str,
+    ) -> Self: ...
+    def lazy(self, *, backend: Implementation | None) -> CompliantLazyFrame: ...
+    def rename(self, mapping: Mapping[str, str]) -> Self: ...
+    def row(self, index: int) -> tuple[Any, ...]: ...
+    def rows(
+        self, *, named: bool
+    ) -> Sequence[tuple[Any, ...]] | Sequence[Mapping[str, Any]]: ...
+    def sample(
+        self,
+        n: int | None,
+        *,
+        fraction: float | None,
+        with_replacement: bool,
+        seed: int | None,
+    ) -> Self: ...
+    def select(self, *exprs: CompliantExprT_contra) -> Self: ...
+    def sort(
+        self, *by: str, descending: bool | Sequence[bool], nulls_last: bool
+    ) -> Self: ...
+    def tail(self, n: int) -> Self: ...
+    def to_arrow(self) -> pa.Table: ...
+    def to_numpy(self) -> _2DArray: ...
+    def to_pandas(self) -> pd.DataFrame: ...
+    def to_polars(self) -> pl.DataFrame: ...
+    @overload
+    def to_dict(self, *, as_series: Literal[True]) -> dict[str, CompliantSeriesT]: ...
+    @overload
+    def to_dict(self, *, as_series: Literal[False]) -> dict[str, list[Any]]: ...
+    def to_dict(
+        self, *, as_series: bool
+    ) -> dict[str, CompliantSeriesT] | dict[str, list[Any]]: ...
+    def unique(
+        self,
+        subset: Sequence[str] | None,
+        *,
+        keep: Literal["any", "first", "last", "none"],
+        maintain_order: bool | None,
+    ) -> Self: ...
+    def unpivot(
+        self,
+        on: Sequence[str] | None,
+        index: Sequence[str] | None,
+        variable_name: str,
+        value_name: str,
+    ) -> Self: ...
+    def with_columns(self, *exprs: CompliantExprT_contra) -> Self: ...
+    def with_row_index(self, name: str) -> Self: ...
+    @overload
+    def write_csv(self, file: None) -> str: ...
+    @overload
+    def write_csv(self, file: str | Path | BytesIO) -> None: ...
+    def write_csv(self, file: str | Path | BytesIO | None) -> str | None: ...
+    def write_parquet(self, file: str | Path | BytesIO) -> None: ...
 
 
 class CompliantLazyFrame(Protocol):
@@ -60,29 +172,25 @@ class CompliantLazyFrame(Protocol):
     def _iter_columns(self) -> Iterator[Any]: ...
 
 
-class EagerDataFrame(CompliantDataFrame[EagerSeriesT], Protocol[EagerSeriesT]):
-    def _maybe_evaluate_expr(
-        self, expr: EagerExpr[Self, EagerSeriesT] | T, /
-    ) -> EagerSeriesT | T:
-        if is_eager_expr(expr):
-            result: Sequence[EagerSeriesT] = expr(self)
-            if len(result) > 1:
-                msg = (
-                    "Multi-output expressions (e.g. `nw.all()` or `nw.col('a', 'b')`) "
-                    "are not supported in this context"
-                )
-                raise ValueError(msg)
-            return result[0]
-        return expr
-
-    def _evaluate_into_exprs(
-        self, *exprs: EagerExpr[Self, EagerSeriesT]
-    ) -> Sequence[EagerSeriesT]:
+class EagerDataFrame(
+    CompliantDataFrame[EagerSeriesT, EagerExprT_contra],
+    Protocol[EagerSeriesT, EagerExprT_contra],
+):
+    def _evaluate_expr(self, expr: EagerExprT_contra, /) -> EagerSeriesT:
+        """Evaluate `expr` and ensure it has a **single** output."""
+        result: Sequence[EagerSeriesT] = expr(self)
+        if len(result) > 1:
+            msg = (
+                "Multi-output expressions (e.g. `nw.all()` or `nw.col('a', 'b')`) "
+                "are not supported in this context"
+            )
+            raise ValueError(msg)
+        return result[0]
+
+    def _evaluate_into_exprs(self, *exprs: EagerExprT_contra) -> Sequence[EagerSeriesT]:
         return list(chain.from_iterable(self._evaluate_into_expr(expr) for expr in exprs))
 
-    def _evaluate_into_expr(
-        self, expr: EagerExpr[Self, EagerSeriesT], /
-    ) -> Sequence[EagerSeriesT]:
+    def _evaluate_into_expr(self, expr: EagerExprT_contra, /) -> Sequence[EagerSeriesT]:
         """Return list of raw columns.
 
         For eager backends we alias operations at each step.
@@ -98,11 +206,3 @@ class EagerDataFrame(CompliantDataFrame[EagerSeriesT], Protocol[EagerSeriesT]):
             msg = f"Safety assertion failed, expected {aliases}, got {result}"
             raise AssertionError(msg)
         return result
-
-
-# NOTE: `mypy` is requiring the gymnastics here and is very fragile
-# DON'T CHANGE THIS or `EagerDataFrame._maybe_evaluate_expr`
-def is_eager_expr(
-    obj: EagerExpr[Any, EagerSeriesT] | Any,
-) -> TypeIs[EagerExpr[Any, EagerSeriesT]]:
-    return hasattr(obj, "__narwhals_expr__")
diff --git a/narwhals/_compliant/expr.py b/narwhals/_compliant/expr.py
index 03e2343e..dece1372 100644
--- a/narwhals/_compliant/expr.py
+++ b/narwhals/_compliant/expr.py
@@ -51,6 +51,7 @@ if TYPE_CHECKING:
     from typing import Mapping
 
     from typing_extensions import Self
+    from typing_extensions import TypeIs
 
     from narwhals._compliant.namespace import CompliantNamespace
     from narwhals._compliant.namespace import EagerNamespace
@@ -390,6 +391,10 @@ class EagerExpr(
     ) -> dict[str, Any]:
         return {}
 
+    @classmethod
+    def _is_expr(cls, obj: Self | Any) -> TypeIs[Self]:
+        return hasattr(obj, "__narwhals_expr__")
+
     def _reuse_series_inner(
         self,
         df: EagerDataFrameT,
@@ -402,8 +407,8 @@ class EagerExpr(
         kwargs = {
             **call_kwargs,
             **{
-                arg_name: df._maybe_evaluate_expr(arg_value)
-                for arg_name, arg_value in expressifiable_args.items()
+                name: df._evaluate_expr(value) if self._is_expr(value) else value
+                for name, value in expressifiable_args.items()
             },
         }
         method = methodcaller(
diff --git a/narwhals/_compliant/selectors.py b/narwhals/_compliant/selectors.py
index 969d4f46..38c02026 100644
--- a/narwhals/_compliant/selectors.py
+++ b/narwhals/_compliant/selectors.py
@@ -65,8 +65,8 @@ __all__ = [
 SeriesOrExprT = TypeVar("SeriesOrExprT", bound="CompliantSeries | NativeExpr")
 SeriesT = TypeVar("SeriesT", bound="CompliantSeries")
 ExprT = TypeVar("ExprT", bound="NativeExpr")
-FrameT = TypeVar("FrameT", bound="CompliantDataFrame[Any] | CompliantLazyFrame")
-DataFrameT = TypeVar("DataFrameT", bound="CompliantDataFrame[Any]")
+FrameT = TypeVar("FrameT", bound="CompliantDataFrame[Any, Any] | CompliantLazyFrame")
+DataFrameT = TypeVar("DataFrameT", bound="CompliantDataFrame[Any, Any]")
 LazyFrameT = TypeVar("LazyFrameT", bound="CompliantLazyFrame")
 SelectorOrExpr: TypeAlias = (
     "CompliantSelector[FrameT, SeriesOrExprT] | CompliantExpr[FrameT, SeriesOrExprT]"
@@ -309,7 +309,7 @@ class CompliantSelector(
 
 
 def _eval_lhs_rhs(
-    df: CompliantDataFrame[Any] | CompliantLazyFrame,
+    df: CompliantDataFrame[Any, Any] | CompliantLazyFrame,
     lhs: CompliantExpr[Any, Any],
     rhs: CompliantExpr[Any, Any],
 ) -> tuple[Sequence[str], Sequence[str]]:
diff --git a/narwhals/_compliant/typing.py b/narwhals/_compliant/typing.py
index 6b0fc0dc..5e728e96 100644
--- a/narwhals/_compliant/typing.py
+++ b/narwhals/_compliant/typing.py
@@ -25,32 +25,36 @@ __all__ = [
     "CompliantDataFrameT",
     "CompliantFrameT",
     "CompliantLazyFrameT",
-    "CompliantSeriesT_co",
+    "CompliantSeriesT",
     "IntoCompliantExpr",
 ]
 NativeExprT_co = TypeVar("NativeExprT_co", bound="NativeExpr", covariant=True)
-CompliantSeriesT_co = TypeVar(
-    "CompliantSeriesT_co", bound="CompliantSeries", covariant=True
-)
+CompliantSeriesT = TypeVar("CompliantSeriesT", bound="CompliantSeries")
 CompliantSeriesOrNativeExprT_co = TypeVar(
     "CompliantSeriesOrNativeExprT_co",
     bound="CompliantSeries | NativeExpr",
     covariant=True,
 )
 CompliantFrameT = TypeVar(
-    "CompliantFrameT", bound="CompliantDataFrame[Any] | CompliantLazyFrame"
+    "CompliantFrameT", bound="CompliantDataFrame[Any, Any] | CompliantLazyFrame"
 )
-CompliantDataFrameT = TypeVar("CompliantDataFrameT", bound="CompliantDataFrame[Any]")
+CompliantDataFrameT = TypeVar("CompliantDataFrameT", bound="CompliantDataFrame[Any, Any]")
 CompliantLazyFrameT = TypeVar("CompliantLazyFrameT", bound="CompliantLazyFrame")
 IntoCompliantExpr: TypeAlias = "CompliantExpr[CompliantFrameT, CompliantSeriesOrNativeExprT_co] | CompliantSeriesOrNativeExprT_co"
 CompliantExprT = TypeVar("CompliantExprT", bound="CompliantExpr[Any, Any]")
+CompliantExprT_contra = TypeVar(
+    "CompliantExprT_contra", bound="CompliantExpr[Any, Any]", contravariant=True
+)
 
-EagerDataFrameT = TypeVar("EagerDataFrameT", bound="EagerDataFrame[Any]")
+EagerDataFrameT = TypeVar("EagerDataFrameT", bound="EagerDataFrame[Any, Any]")
 EagerSeriesT = TypeVar("EagerSeriesT", bound="EagerSeries[Any]")
 EagerSeriesT_co = TypeVar("EagerSeriesT_co", bound="EagerSeries[Any]", covariant=True)
 EagerExprT = TypeVar("EagerExprT", bound="EagerExpr[Any, Any]")
+EagerExprT_contra = TypeVar(
+    "EagerExprT_contra", bound="EagerExpr[Any, Any]", contravariant=True
+)
 EagerNamespaceAny: TypeAlias = (
-    "EagerNamespace[EagerDataFrame[Any], EagerSeries[Any], EagerExpr[Any, Any]]"
+    "EagerNamespace[EagerDataFrame[Any, Any], EagerSeries[Any], EagerExpr[Any, Any]]"
 )
 
 AliasNames: TypeAlias = Callable[[Sequence[str]], Sequence[str]]
diff --git a/narwhals/_dask/dataframe.py b/narwhals/_dask/dataframe.py
index e90cc234..42ef5d4c 100644
--- a/narwhals/_dask/dataframe.py
+++ b/narwhals/_dask/dataframe.py
@@ -93,7 +93,7 @@ class DaskLazyFrame(CompliantLazyFrame):
         self: Self,
         backend: Implementation | None,
         **kwargs: Any,
-    ) -> CompliantDataFrame[Any]:
+    ) -> CompliantDataFrame[Any, Any]:
         result = self._native_frame.compute(**kwargs)
 
         if backend is None or backend is Implementation.PANDAS:
diff --git a/narwhals/_duckdb/dataframe.py b/narwhals/_duckdb/dataframe.py
index 88f27509..d23bf550 100644
--- a/narwhals/_duckdb/dataframe.py
+++ b/narwhals/_duckdb/dataframe.py
@@ -92,7 +92,7 @@ class DuckDBLazyFrame(CompliantLazyFrame):
         self: Self,
         backend: ModuleType | Implementation | str | None,
         **kwargs: Any,
-    ) -> CompliantDataFrame[Any]:
+    ) -> CompliantDataFrame[Any, Any]:
         if backend is None or backend is Implementation.PYARROW:
             import pyarrow as pa  # ignore-banned-import
 
diff --git a/narwhals/_expression_parsing.py b/narwhals/_expression_parsing.py
index 41945232..08d2a969 100644
--- a/narwhals/_expression_parsing.py
+++ b/narwhals/_expression_parsing.py
@@ -109,7 +109,7 @@ def extract_compliant(
 
 def evaluate_output_names_and_aliases(
     expr: CompliantExpr[Any, Any],
-    df: CompliantDataFrame[Any] | CompliantLazyFrame,
+    df: CompliantDataFrame[Any, Any] | CompliantLazyFrame,
     exclude: Sequence[str],
 ) -> tuple[Sequence[str], Sequence[str]]:
     output_names = expr._evaluate_output_names(df)
diff --git a/narwhals/_pandas_like/dataframe.py b/narwhals/_pandas_like/dataframe.py
index 190d3ce2..e5a80e20 100644
--- a/narwhals/_pandas_like/dataframe.py
+++ b/narwhals/_pandas_like/dataframe.py
@@ -4,6 +4,7 @@ from typing import TYPE_CHECKING
 from typing import Any
 from typing import Iterator
 from typing import Literal
+from typing import Mapping
 from typing import Sequence
 from typing import cast
 from typing import overload
@@ -82,7 +83,9 @@ CLASSICAL_NUMPY_DTYPES: frozenset[np.dtype[Any]] = frozenset(
 )
 
 
-class PandasLikeDataFrame(EagerDataFrame["PandasLikeSeries"], CompliantLazyFrame):
+class PandasLikeDataFrame(
+    EagerDataFrame["PandasLikeSeries", "PandasLikeExpr"], CompliantLazyFrame
+):
     # --- not in the spec ---
     def __init__(
         self: Self,
@@ -356,17 +359,15 @@ class PandasLikeDataFrame(EagerDataFrame["PandasLikeSeries"], CompliantLazyFrame
         *,
         named: bool,
         buffer_size: int,
-    ) -> Iterator[list[tuple[Any, ...]]] | Iterator[list[dict[str, Any]]]:
+    ) -> Iterator[tuple[Any, ...]] | Iterator[dict[str, Any]]:
         # The param ``buffer_size`` is only here for compatibility with the Polars API
         # and has no effect on the output.
         if not named:
             yield from self._native_frame.itertuples(index=False, name=None)
         else:
             col_names = self._native_frame.columns
-            yield from (
-                dict(zip(col_names, row))
-                for row in self._native_frame.itertuples(index=False)
-            )  # type: ignore[misc]
+            for row in self._native_frame.itertuples(index=False):
+                yield dict(zip(col_names, row))
 
     @property
     def schema(self: Self) -> dict[str, DType]:
@@ -397,11 +398,6 @@ class PandasLikeDataFrame(EagerDataFrame["PandasLikeSeries"], CompliantLazyFrame
             validate_column_names=False,
         )
 
-    def aggregate(
-        self: PandasLikeDataFrame, *exprs: PandasLikeExpr
-    ) -> PandasLikeDataFrame:
-        return self.select(*exprs)
-
     def select(self: PandasLikeDataFrame, *exprs: PandasLikeExpr) -> PandasLikeDataFrame:
         new_series = self._evaluate_into_exprs(*exprs)
         if not new_series:
@@ -418,7 +414,7 @@ class PandasLikeDataFrame(EagerDataFrame["PandasLikeSeries"], CompliantLazyFrame
         return self._from_native_frame(df, validate_column_names=True)
 
     def drop_nulls(
-        self: PandasLikeDataFrame, subset: list[str] | None
+        self: PandasLikeDataFrame, subset: Sequence[str] | None
     ) -> PandasLikeDataFrame:
         if subset is None:
             return self._from_native_frame(
@@ -445,8 +441,8 @@ class PandasLikeDataFrame(EagerDataFrame["PandasLikeSeries"], CompliantLazyFrame
             )
         )
 
-    def row(self: Self, row: int) -> tuple[Any, ...]:
-        return tuple(x for x in self._native_frame.iloc[row])
+    def row(self: Self, index: int) -> tuple[Any, ...]:
+        return tuple(x for x in self._native_frame.iloc[index])
 
     def filter(
         self: PandasLikeDataFrame, predicate: PandasLikeExpr | list[bool]
@@ -494,7 +490,7 @@ class PandasLikeDataFrame(EagerDataFrame["PandasLikeSeries"], CompliantLazyFrame
         )
         return self._from_native_frame(df, validate_column_names=False)
 
-    def rename(self: Self, mapping: dict[str, str]) -> Self:
+    def rename(self: Self, mapping: Mapping[str, str]) -> Self:
         return self._from_native_frame(
             rename(
                 self._native_frame,
@@ -504,7 +500,7 @@ class PandasLikeDataFrame(EagerDataFrame["PandasLikeSeries"], CompliantLazyFrame
             )
         )
 
-    def drop(self: Self, columns: list[str], strict: bool) -> Self:  # noqa: FBT001
+    def drop(self: Self, columns: Sequence[str], *, strict: bool) -> Self:
         to_drop = parse_columns_to_drop(
             compliant_frame=self, columns=columns, strict=strict
         )
@@ -535,7 +531,7 @@ class PandasLikeDataFrame(EagerDataFrame["PandasLikeSeries"], CompliantLazyFrame
         self: Self,
         backend: Implementation | None,
         **kwargs: Any,
-    ) -> CompliantDataFrame[Any]:
+    ) -> CompliantDataFrame[Any, Any]:
         if backend is None:
             return PandasLikeDataFrame(
                 self._native_frame,
@@ -597,8 +593,8 @@ class PandasLikeDataFrame(EagerDataFrame["PandasLikeSeries"], CompliantLazyFrame
         other: Self,
         *,
         how: Literal["left", "inner", "cross", "anti", "semi"],
-        left_on: list[str] | None,
-        right_on: list[str] | None,
+        left_on: Sequence[str] | None,
+        right_on: Sequence[str] | None,
         suffix: str,
     ) -> Self:
         if how == "cross":
@@ -655,7 +651,7 @@ class PandasLikeDataFrame(EagerDataFrame["PandasLikeSeries"], CompliantLazyFrame
                 other_native = rename(
                     select_columns_by_name(
                         other._native_frame,
-                        right_on,
+                        list(right_on),
                         self._backend_version,
                         self._implementation,
                     ),
@@ -684,7 +680,7 @@ class PandasLikeDataFrame(EagerDataFrame["PandasLikeSeries"], CompliantLazyFrame
                 rename(
                     select_columns_by_name(
                         other._native_frame,
-                        right_on,
+                        list(right_on),
                         self._backend_version,
                         self._implementation,
                     ),
@@ -735,8 +731,8 @@ class PandasLikeDataFrame(EagerDataFrame["PandasLikeSeries"], CompliantLazyFrame
         *,
         left_on: str | None,
         right_on: str | None,
-        by_left: list[str] | None,
-        by_right: list[str] | None,
+        by_left: Sequence[str] | None,
+        by_right: Sequence[str] | None,
         strategy: Literal["backward", "forward", "nearest"],
         suffix: str,
     ) -> Self:
@@ -768,7 +764,7 @@ class PandasLikeDataFrame(EagerDataFrame["PandasLikeSeries"], CompliantLazyFrame
 
     def unique(
         self: Self,
-        subset: list[str] | None,
+        subset: Sequence[str] | None,
         *,
         keep: Literal["any", "first", "last", "none"],
         maintain_order: bool | None = None,
@@ -1068,8 +1064,8 @@ class PandasLikeDataFrame(EagerDataFrame["PandasLikeSeries"], CompliantLazyFrame
 
     def unpivot(
         self: Self,
-        on: list[str] | None,
-        index: list[str] | None,
+        on: Sequence[str] | None,
+        index: Sequence[str] | None,
         variable_name: str,
         value_name: str,
     ) -> Self:
diff --git a/narwhals/_pandas_like/utils.py b/narwhals/_pandas_like/utils.py
index 0fb7b352..30bf6c18 100644
--- a/narwhals/_pandas_like/utils.py
+++ b/narwhals/_pandas_like/utils.py
@@ -7,6 +7,7 @@ from contextlib import suppress
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Iterable
+from typing import Sized
 from typing import TypeVar
 from typing import cast
 
@@ -14,12 +15,13 @@ import pandas as pd
 
 from narwhals.exceptions import ColumnNotFoundError
 from narwhals.exceptions import DuplicateError
+from narwhals.exceptions import ShapeError
 from narwhals.utils import Implementation
 from narwhals.utils import Version
 from narwhals.utils import import_dtypes_module
 from narwhals.utils import isinstance_or_issubclass
 
-T = TypeVar("T")
+T = TypeVar("T", bound=Sized)
 
 if TYPE_CHECKING:
     from pandas._typing import Dtype as PandasDtype
@@ -137,6 +139,9 @@ def extract_dataframe_comparand(
     if other._broadcast:
         s = other._native_series
         return s.__class__(s.iloc[0], index=index, dtype=s.dtype, name=s.name)
+    if (len_other := len(other)) != (len_idx := len(index)):
+        msg = f"Expected object of length {len_idx}, got: {len_other}."
+        raise ShapeError(msg)
     if other._native_series.index is not index:
         return set_index(
             other._native_series,
diff --git a/narwhals/_polars/dataframe.py b/narwhals/_polars/dataframe.py
index e26b6631..439528bf 100644
--- a/narwhals/_polars/dataframe.py
+++ b/narwhals/_polars/dataframe.py
@@ -4,6 +4,7 @@ from typing import TYPE_CHECKING
 from typing import Any
 from typing import Iterator
 from typing import Literal
+from typing import Mapping
 from typing import Sequence
 from typing import overload
 
@@ -23,9 +24,13 @@ from narwhals.utils import validate_backend_version
 
 if TYPE_CHECKING:
     from types import ModuleType
+    from typing import Callable
     from typing import TypeVar
 
+    import pandas as pd
+    import pyarrow as pa
     from typing_extensions import Self
+    from typing_extensions import TypeAlias
 
     from narwhals._polars.group_by import PolarsGroupBy
     from narwhals._polars.group_by import PolarsLazyGroupBy
@@ -37,9 +42,43 @@ if TYPE_CHECKING:
     from narwhals.utils import Version
 
     T = TypeVar("T")
+    R = TypeVar("R")
+
+Method: TypeAlias = "Callable[..., R]"
+"""Generic alias representing all methods implemented via `__getattr__`.
+
+Where `R` is the return type.
+"""
 
 
 class PolarsDataFrame:
+    clone: Method[Self]
+    collect: Method[CompliantDataFrame[Any, Any]]
+    drop_nulls: Method[Self]
+    estimated_size: Method[int | float]
+    filter: Method[Self]
+    gather_every: Method[Self]
+    item: Method[Any]
+    iter_rows: Method[Iterator[tuple[Any, ...]] | Iterator[Mapping[str, Any]]]
+    is_unique: Method[PolarsSeries]
+    join: Method[Self]
+    join_asof: Method[Self]
+    rename: Method[Self]
+    row: Method[tuple[Any, ...]]
+    rows: Method[Sequence[tuple[Any, ...]] | Sequence[Mapping[str, Any]]]
+    sample: Method[Self]
+    select: Method[Self]
+    sort: Method[Self]
+    to_arrow: Method[pa.Table]
+    to_numpy: Method[_2DArray]
+    to_pandas: Method[pd.DataFrame]
+    unique: Method[Self]
+    with_columns: Method[Self]
+    # NOTE: `write_csv` requires an `@overload` for `str | None`
+    # Can't do that here 😟
+    write_csv: Method[Any]
+    write_parquet: Method[None]
+
     def __init__(
         self: Self,
         df: pl.DataFrame,
@@ -222,7 +261,7 @@ class PolarsDataFrame:
         return self._from_native_frame(self._native_frame.select(*column_names))
 
     def aggregate(self: Self, *exprs: Any) -> Self:
-        return self.select(*exprs)  # type: ignore[no-any-return]
+        return self.select(*exprs)
 
     def get_column(self: Self, name: str) -> PolarsSeries:
         from narwhals._polars.series import PolarsSeries
@@ -321,7 +360,7 @@ class PolarsDataFrame:
             return self._from_native_frame(self._native_frame.with_row_count(name))
         return self._from_native_frame(self._native_frame.with_row_index(name))
 
-    def drop(self: Self, columns: list[str], strict: bool) -> Self:  # noqa: FBT001
+    def drop(self: Self, columns: Sequence[str], *, strict: bool) -> Self:
         to_drop = parse_columns_to_drop(
             compliant_frame=self, columns=columns, strict=strict
         )
@@ -329,8 +368,8 @@ class PolarsDataFrame:
 
     def unpivot(
         self: Self,
-        on: list[str] | None,
-        index: list[str] | None,
+        on: Sequence[str] | None,
+        index: Sequence[str] | None,
         variable_name: str,
         value_name: str,
     ) -> Self:
@@ -475,7 +514,7 @@ class PolarsLazyFrame:
         self: Self,
         backend: Implementation | None,
         **kwargs: Any,
-    ) -> CompliantDataFrame[Any]:
+    ) -> CompliantDataFrame[Any, Any]:
         try:
             result = self._native_frame.collect(**kwargs)
         except Exception as e:  # noqa: BLE001
diff --git a/narwhals/_spark_like/dataframe.py b/narwhals/_spark_like/dataframe.py
index 8bd1319d..fa44d440 100644
--- a/narwhals/_spark_like/dataframe.py
+++ b/narwhals/_spark_like/dataframe.py
@@ -172,7 +172,7 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
         self: Self,
         backend: ModuleType | Implementation | str | None,
         **kwargs: Any,
-    ) -> CompliantDataFrame[Any]:
+    ) -> CompliantDataFrame[Any, Any]:
         if backend is Implementation.PANDAS:
             import pandas as pd  # ignore-banned-import
 
diff --git a/narwhals/utils.py b/narwhals/utils.py
index 87e5d63e..95722578 100644
--- a/narwhals/utils.py
+++ b/narwhals/utils.py
@@ -1399,13 +1399,13 @@ def generate_repr(header: str, native_repr: str) -> str:
     )
 
 
-def check_column_exists(columns: list[str], subset: list[str] | None) -> None:
+def check_column_exists(columns: Sequence[str], subset: Sequence[str] | None) -> None:
     if subset is not None and (missing := set(subset).difference(columns)):
         msg = f"Column(s) {sorted(missing)} not found in {columns}"
         raise ColumnNotFoundError(msg)
 
 
-def check_column_names_are_unique(columns: list[str]) -> None:
+def check_column_names_are_unique(columns: Sequence[str]) -> None:
     len_unique_columns = len(set(columns))
     if len(columns) != len_unique_columns:
         from collections import Counter
@@ -1472,8 +1472,8 @@ def _hasattr_static(obj: Any, attr: str) -> bool:
 
 
 def is_compliant_dataframe(
-    obj: CompliantDataFrame[CompliantSeriesT_co] | Any,
-) -> TypeIs[CompliantDataFrame[CompliantSeriesT_co]]:
+    obj: CompliantDataFrame[CompliantSeriesT_co, CompliantExprT_co] | Any,
+) -> TypeIs[CompliantDataFrame[CompliantSeriesT_co, CompliantExprT_co]]:
     return _hasattr_static(obj, "__narwhals_dataframe__")
 
 


Test patch:
diff --git a/tests/frame/with_columns_test.py b/tests/frame/with_columns_test.py
index 5756bb9b..c793a3e0 100644
--- a/tests/frame/with_columns_test.py
+++ b/tests/frame/with_columns_test.py
@@ -5,6 +5,7 @@ import pandas as pd
 import pytest
 
 import narwhals.stable.v1 as nw
+from narwhals.exceptions import ShapeError
 from tests.utils import PYARROW_VERSION
 from tests.utils import Constructor
 from tests.utils import ConstructorEager
@@ -68,3 +69,12 @@ def test_with_columns_dtypes_single_row(
     df = nw.from_native(constructor(data)).with_columns(nw.col("a").cast(nw.Categorical))
     result = df.with_columns(nw.col("a"))
     assert result.collect_schema() == {"a": nw.Categorical}
+
+
+def test_with_columns_series_shape_mismatch(constructor_eager: ConstructorEager) -> None:
+    df1 = nw.from_native(constructor_eager({"first": [1, 2, 3]}), eager_only=True)
+    second = nw.from_native(constructor_eager({"second": [1, 2, 3, 4]}), eager_only=True)[
+        "second"
+    ]
+    with pytest.raises(ShapeError):
+        df1.with_columns(second=second)