chore: Spec CompliantLazyFrame

narwhals/_arrow/dataframe.py

@@ -52,6 +52,8 @@
     from narwhals._arrow.typing import Mask  # type: ignore[attr-defined]
     from narwhals._arrow.typing import Order  # type: ignore[attr-defined]
     from narwhals.dtypes import DType
+    from narwhals.typing import CompliantDataFrame
+    from narwhals.typing import CompliantLazyFrame
     from narwhals.typing import SizeUnit
     from narwhals.typing import _1DArray
     from narwhals.typing import _2DArray
@@ -69,11 +71,8 @@
     ]
     PromoteOptions: TypeAlias = Literal["none", "default", "permissive"]
 
-from narwhals.typing import CompliantDataFrame
-from narwhals.typing import CompliantLazyFrame
 
-
-class ArrowDataFrame(EagerDataFrame["ArrowSeries", "ArrowExpr"], CompliantLazyFrame):
+class ArrowDataFrame(EagerDataFrame["ArrowSeries", "ArrowExpr", "pa.Table"]):
     # --- not in the spec ---
     def __init__(
         self: Self,
@@ -349,6 +348,8 @@ def estimated_size(self: Self, unit: SizeUnit) -> int | float:
         sz = self._native_frame.nbytes
         return scale_bytes(sz, unit)
 
+    explode = not_implemented()
+
     @property
     def columns(self: Self) -> list[str]:
         return self._native_frame.schema.names
@@ -573,7 +574,9 @@ def tail(self: Self, n: int) -> Self:
         else:
             return self._from_native_frame(df.slice(abs(n)), validate_column_names=False)
 
-    def lazy(self: Self, *, backend: Implementation | None = None) -> CompliantLazyFrame:
+    def lazy(
+        self: Self, *, backend: Implementation | None = None
+    ) -> CompliantLazyFrame[Any, Any]:
         from narwhals.utils import parse_version
 
         if backend is None:

narwhals/_compliant/dataframe.py

@@ -16,7 +16,11 @@
 from narwhals._compliant.typing import CompliantSeriesT
 from narwhals._compliant.typing import EagerExprT_contra
 from narwhals._compliant.typing import EagerSeriesT
+from narwhals._compliant.typing import NativeFrameT_co
 from narwhals._expression_parsing import evaluate_output_names_and_aliases
+from narwhals.utils import Version
+from narwhals.utils import _StoresNative
+from narwhals.utils import deprecated
 
 if TYPE_CHECKING:
     from io import BytesIO
@@ -70,6 +74,7 @@ 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 explode(self: Self, columns: Sequence[str]) -> Self: ...
     def filter(self, predicate: CompliantExprT_contra | Incomplete) -> Self: ...
     def gather_every(self, n: int, offset: int) -> Self: ...
     def get_column(self, name: str) -> CompliantSeriesT: ...
@@ -101,7 +106,7 @@ def join_asof(
         strategy: Literal["backward", "forward", "nearest"],
         suffix: str,
     ) -> Self: ...
-    def lazy(self, *, backend: Implementation | None) -> CompliantLazyFrame: ...
+    def lazy(self, *, backend: Implementation | None) -> CompliantLazyFrame[Any, Any]: ...
     def rename(self, mapping: Mapping[str, str]) -> Self: ...
     def row(self, index: int) -> tuple[Any, ...]: ...
     def rows(
@@ -136,7 +141,7 @@ def unique(
         subset: Sequence[str] | None,
         *,
         keep: Literal["any", "first", "last", "none"],
-        maintain_order: bool | None,
+        maintain_order: bool | None = None,
     ) -> Self: ...
     def unpivot(
         self,
@@ -155,26 +160,112 @@ def write_csv(self, file: str | Path | BytesIO | None) -> str | None: ...
     def write_parquet(self, file: str | Path | BytesIO) -> None: ...
 
 
-class CompliantLazyFrame(Protocol):
+class CompliantLazyFrame(
+    _StoresNative[NativeFrameT_co], Protocol[CompliantExprT_contra, NativeFrameT_co]
+):
+    _native_frame: Any
+    _implementation: Implementation
+    _backend_version: tuple[int, ...]
+    _version: Version
+
     def __narwhals_lazyframe__(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 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).
+        """
+        ...
+
+    def _change_version(self, version: Version) -> Self: ...
+
+    @property
+    def native(self) -> NativeFrameT_co:
+        return self._native_frame  # type: ignore[no-any-return]
 
     @property
     def columns(self) -> Sequence[str]: ...
     @property
     def schema(self) -> Mapping[str, DType]: ...
     def _iter_columns(self) -> Iterator[Any]: ...
+    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 explode(self: Self, columns: Sequence[str]) -> Self: ...
+    def filter(self, predicate: CompliantExprT_contra | Incomplete) -> Self: ...
+    @deprecated(
+        "`LazyFrame.gather_every` is deprecated and will be removed in a future version."
+    )
+    def gather_every(self, n: int, offset: int) -> Self: ...
+    def group_by(self, *keys: str, drop_null_keys: bool) -> Incomplete: ...
+    def head(self, n: int) -> Self: ...
+    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) -> Self | Incomplete:
+        # The `backend`` argument has no effect but we keep it here for
+        # backwards compatibility because in `narwhals.stable.v1`
+        # function `.from_native()` will return a DataFrame for DuckDB.
+
+        if backend is not None:  # pragma: no cover
+            msg = f"`backend` argument is not supported for {self._implementation!r}"
+            raise ValueError(msg)
+        return self
+
+    def rename(self, mapping: Mapping[str, str]) -> Self: ...
+    def select(self, *exprs: CompliantExprT_contra) -> Self: ...
+    def sort(
+        self, *by: str, descending: bool | Sequence[bool], nulls_last: bool
+    ) -> Self: ...
+    @deprecated("`LazyFrame.tail` is deprecated and will be removed in a future version.")
+    def tail(self, n: int) -> Self: ...
+    def unique(
+        self,
+        subset: Sequence[str] | None,
+        *,
+        keep: Literal["any", "first", "last", "none"],
+        maintain_order: bool | None = 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: ...
 
 
 class EagerDataFrame(
     CompliantDataFrame[EagerSeriesT, EagerExprT_contra],
-    Protocol[EagerSeriesT, EagerExprT_contra],
+    CompliantLazyFrame[EagerExprT_contra, NativeFrameT_co],
+    Protocol[EagerSeriesT, EagerExprT_contra, NativeFrameT_co],
 ):
     def _evaluate_expr(self, expr: EagerExprT_contra, /) -> EagerSeriesT:
         """Evaluate `expr` and ensure it has a **single** output."""

narwhals/_compliant/selectors.py

@@ -65,9 +65,11 @@
 SeriesOrExprT = TypeVar("SeriesOrExprT", bound="CompliantSeries | NativeExpr")
 SeriesT = TypeVar("SeriesT", bound="CompliantSeries")
 ExprT = TypeVar("ExprT", bound="NativeExpr")
-FrameT = TypeVar("FrameT", bound="CompliantDataFrame[Any, Any] | CompliantLazyFrame")
+FrameT = TypeVar(
+    "FrameT", bound="CompliantDataFrame[Any, Any] | CompliantLazyFrame[Any, Any]"
+)
 DataFrameT = TypeVar("DataFrameT", bound="CompliantDataFrame[Any, Any]")
-LazyFrameT = TypeVar("LazyFrameT", bound="CompliantLazyFrame")
+LazyFrameT = TypeVar("LazyFrameT", bound="CompliantLazyFrame[Any, Any]")
 SelectorOrExpr: TypeAlias = (
     "CompliantSelector[FrameT, SeriesOrExprT] | CompliantExpr[FrameT, SeriesOrExprT]"
 )
@@ -309,7 +311,7 @@ def __repr__(self: Self) -> str:  # pragma: no cover
 
 
 def _eval_lhs_rhs(
-    df: CompliantDataFrame[Any, Any] | CompliantLazyFrame,
+    df: CompliantDataFrame[Any, Any] | CompliantLazyFrame[Any, Any],
     lhs: CompliantExpr[Any, Any],
     rhs: CompliantExpr[Any, Any],
 ) -> tuple[Sequence[str], Sequence[str]]:

narwhals/_compliant/typing.py

@@ -18,6 +18,7 @@
     from narwhals._compliant.namespace import EagerNamespace
     from narwhals._compliant.series import CompliantSeries
     from narwhals._compliant.series import EagerSeries
+    from narwhals.typing import NativeFrame
 
 __all__ = [
     "AliasName",
@@ -35,26 +36,28 @@
     bound="CompliantSeries | NativeExpr",
     covariant=True,
 )
+
+NativeFrameT_co = TypeVar("NativeFrameT_co", bound="NativeFrame", covariant=True)
 CompliantFrameT = TypeVar(
-    "CompliantFrameT", bound="CompliantDataFrame[Any, Any] | CompliantLazyFrame"
+    "CompliantFrameT", bound="CompliantDataFrame[Any, Any] | CompliantLazyFrame[Any, Any]"
 )
 CompliantDataFrameT = TypeVar("CompliantDataFrameT", bound="CompliantDataFrame[Any, Any]")
-CompliantLazyFrameT = TypeVar("CompliantLazyFrameT", bound="CompliantLazyFrame")
+CompliantLazyFrameT = TypeVar("CompliantLazyFrameT", bound="CompliantLazyFrame[Any, Any]")
 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, Any]")
+EagerDataFrameT = TypeVar("EagerDataFrameT", bound="EagerDataFrame[Any, 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, Any], EagerSeries[Any], EagerExpr[Any, Any]]"
+    "EagerNamespace[EagerDataFrame[Any, Any, Any], EagerSeries[Any], EagerExpr[Any, Any]]"
 )
 
 AliasNames: TypeAlias = Callable[[Sequence[str]], Sequence[str]]

narwhals/_dask/dataframe.py

@@ -4,6 +4,7 @@
 from typing import Any
 from typing import Iterator
 from typing import Literal
+from typing import Mapping
 from typing import Sequence
 
 import dask.dataframe as dd
@@ -18,6 +19,7 @@
 from narwhals.utils import Implementation
 from narwhals.utils import check_column_exists
 from narwhals.utils import generate_temporary_column_name
+from narwhals.utils import not_implemented
 from narwhals.utils import parse_columns_to_drop
 from narwhals.utils import parse_version
 from narwhals.utils import validate_backend_version
@@ -35,7 +37,7 @@
     from narwhals.utils import Version
 
 
-class DaskLazyFrame(CompliantLazyFrame):
+class DaskLazyFrame(CompliantLazyFrame["DaskExpr", "dd.DataFrame"]):
     def __init__(
         self: Self,
         native_dataframe: dd.DataFrame,
@@ -168,7 +170,7 @@ def select(self: Self, *exprs: DaskExpr) -> Self:
         )
         return self._from_native_frame(df)
 
-    def drop_nulls(self: Self, subset: list[str] | None) -> Self:
+    def drop_nulls(self: Self, subset: Sequence[str] | None) -> Self:
         if subset is None:
             return self._from_native_frame(self._native_frame.dropna())
         plx = self.__narwhals_namespace__()
@@ -189,7 +191,7 @@ def schema(self: Self) -> dict[str, DType]:
     def collect_schema(self: Self) -> dict[str, DType]:
         return self.schema
 
-    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
         )
@@ -205,7 +207,7 @@ def with_row_index(self: Self, name: str) -> Self:
             )
         )
 
-    def rename(self: Self, mapping: dict[str, str]) -> Self:
+    def rename(self: Self, mapping: Mapping[str, str]) -> Self:
         return self._from_native_frame(self._native_frame.rename(columns=mapping))
 
     def head(self: Self, n: int) -> Self:
@@ -215,9 +217,10 @@ def head(self: Self, n: int) -> Self:
 
     def unique(
         self: Self,
-        subset: list[str] | None,
+        subset: Sequence[str] | None,
         *,
-        keep: Literal["any", "none"] = "any",
+        keep: Literal["any", "first", "last", "none"] = "any",
+        maintain_order: bool | None = None,
     ) -> Self:
         check_column_exists(self.columns, subset)
         native_frame = self._native_frame
@@ -254,8 +257,8 @@ def join(
         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":
@@ -286,7 +289,7 @@ def join(
             other_native = (
                 select_columns_by_name(
                     other._native_frame,
-                    right_on,
+                    list(right_on),
                     self._backend_version,
                     self._implementation,
                 )
@@ -313,7 +316,7 @@ def join(
             other_native = (
                 select_columns_by_name(
                     other._native_frame,
-                    right_on,
+                    list(right_on),
                     self._backend_version,
                     self._implementation,
                 )
@@ -364,8 +367,8 @@ def join_asof(
         *,
         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:
@@ -412,8 +415,8 @@ def gather_every(self: Self, n: int, offset: int) -> Self:
 
     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:
@@ -425,3 +428,5 @@ def unpivot(
                 value_name=value_name,
             )
         )
+
+    explode = not_implemented()