feat: Add backend argument to lazy

narwhals/_arrow/dataframe.py

@@ -19,6 +19,7 @@
 from narwhals._expression_parsing import evaluate_into_exprs
 from narwhals.dependencies import is_numpy_array
 from narwhals.utils import Implementation
+from narwhals.utils import Version
 from narwhals.utils import check_column_exists
 from narwhals.utils import generate_temporary_column_name
 from narwhals.utils import is_sequence_but_not_str
@@ -519,8 +520,44 @@ def tail(self: Self, n: int) -> Self:
         else:
             return self._from_native_frame(df.slice(abs(n)))
 
-    def lazy(self: Self) -> Self:
-        return self
+    def lazy(self: Self, *, backend: Implementation | None = None) -> CompliantLazyFrame:
+        from narwhals.utils import parse_version
+
+        if backend is None:
+            return self
+        elif backend is Implementation.DUCKDB:
+            import duckdb  # ignore-banned-import
+
+            from narwhals._duckdb.dataframe import DuckDBLazyFrame
+
+            df = self._native_frame  # noqa: F841
+            return DuckDBLazyFrame(
+                df=duckdb.table("df"),
+                backend_version=parse_version(duckdb.__version__),
+                version=self._version,
+            )
+        elif backend is Implementation.POLARS:
+            import polars as pl  # ignore-banned-import
+
+            from narwhals._polars.dataframe import PolarsLazyFrame
+
+            return PolarsLazyFrame(
+                df=pl.from_arrow(self._native_frame).lazy(),  # type: ignore[union-attr]
+                backend_version=parse_version(pl.__version__),
+                version=self._version,
+            )
+        elif backend is Implementation.DASK:
+            import dask  # ignore-banned-import
+            import dask.dataframe as dd  # ignore-banned-import
+
+            from narwhals._dask.dataframe import DaskLazyFrame
+
+            return DaskLazyFrame(
+                native_dataframe=dd.from_pandas(self._native_frame.to_pandas()),
+                backend_version=parse_version(dask.__version__),
+                version=self._version,
+            )
+        raise AssertionError  # pragma: no cover
 
     def collect(self: Self) -> ArrowDataFrame:
         return ArrowDataFrame(

narwhals/_duckdb/dataframe.py

@@ -142,7 +142,14 @@ def drop(self: Self, columns: list[str], strict: bool) -> Self:  # noqa: FBT001
         selection = (col for col in self.columns if col not in columns_to_drop)
         return self._from_native_frame(self._native_frame.select(*selection))
 
-    def lazy(self: Self) -> Self:
+    def lazy(self: Self, *, backend: Implementation | None = None) -> Self:
+        # 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 = "`backend` argument is not supported for DuckDB"
+            raise ValueError(msg)
         return self
 
     def with_columns(

narwhals/_pandas_like/dataframe.py

@@ -709,8 +709,44 @@ def unique(
         )
 
     # --- lazy-only ---
-    def lazy(self: Self) -> Self:
-        return self
+    def lazy(self: Self, *, backend: Implementation | None = None) -> CompliantLazyFrame:
+        from narwhals.utils import parse_version
+
+        if backend is None:
+            return self
+        elif backend is Implementation.DUCKDB:
+            import duckdb  # ignore-banned-import
+
+            from narwhals._duckdb.dataframe import DuckDBLazyFrame
+
+            df = self._native_frame  # noqa: F841
+            return DuckDBLazyFrame(
+                df=duckdb.table("df"),
+                backend_version=parse_version(duckdb.__version__),
+                version=self._version,
+            )
+        elif backend is Implementation.POLARS:
+            import polars as pl  # ignore-banned-import
+
+            from narwhals._polars.dataframe import PolarsLazyFrame
+
+            return PolarsLazyFrame(
+                df=pl.from_pandas(self._native_frame).lazy(),
+                backend_version=parse_version(pl.__version__),
+                version=self._version,
+            )
+        elif backend is Implementation.DASK:
+            import dask  # ignore-banned-import
+            import dask.dataframe as dd  # ignore-banned-import
+
+            from narwhals._dask.dataframe import DaskLazyFrame
+
+            return DaskLazyFrame(
+                native_dataframe=dd.from_pandas(self._native_frame),
+                backend_version=parse_version(dask.__version__),
+                version=self._version,
+            )
+        raise AssertionError  # pragma: no cover
 
     @property
     def shape(self: Self) -> tuple[int, int]:

narwhals/_polars/dataframe.py

@@ -29,6 +29,7 @@
     from narwhals._polars.group_by import PolarsLazyGroupBy
     from narwhals._polars.series import PolarsSeries
     from narwhals.dtypes import DType
+    from narwhals.typing import CompliantLazyFrame
     from narwhals.utils import Version
 
     T = TypeVar("T")
@@ -231,12 +232,40 @@ def schema(self: Self) -> dict[str, DType]:
             for name, dtype in schema.items()
         }
 
-    def lazy(self: Self) -> PolarsLazyFrame:
-        return PolarsLazyFrame(
-            self._native_frame.lazy(),
-            backend_version=self._backend_version,
-            version=self._version,
-        )
+    def lazy(self: Self, *, backend: Implementation | None = None) -> CompliantLazyFrame:
+        from narwhals.utils import parse_version
+
+        if backend is None or backend is Implementation.POLARS:
+            from narwhals._polars.dataframe import PolarsLazyFrame
+
+            return PolarsLazyFrame(
+                self._native_frame.lazy(),
+                backend_version=self._backend_version,
+                version=self._version,
+            )
+        elif backend is Implementation.DUCKDB:
+            import duckdb  # ignore-banned-import
+
+            from narwhals._duckdb.dataframe import DuckDBLazyFrame
+
+            df = self._native_frame  # noqa: F841
+            return DuckDBLazyFrame(
+                df=duckdb.table("df"),
+                backend_version=parse_version(duckdb.__version__),
+                version=self._version,
+            )
+        elif backend is Implementation.DASK:
+            import dask  # ignore-banned-import
+            import dask.dataframe as dd  # ignore-banned-import
+
+            from narwhals._dask.dataframe import DaskLazyFrame
+
+            return DaskLazyFrame(
+                native_dataframe=dd.from_pandas(self._native_frame.to_pandas()),
+                backend_version=parse_version(dask.__version__),
+                version=self._version,
+            )
+        raise AssertionError  # pragma: no cover
 
     @overload
     def to_dict(self: Self, *, as_series: Literal[True]) -> dict[str, PolarsSeries]: ...

narwhals/dataframe.py

@@ -23,6 +23,7 @@
 from narwhals.exceptions import ShapeError
 from narwhals.schema import Schema
 from narwhals.translate import to_native
+from narwhals.utils import Implementation
 from narwhals.utils import find_stacklevel
 from narwhals.utils import flatten
 from narwhals.utils import generate_repr
@@ -50,7 +51,6 @@
     from narwhals.typing import IntoExpr
     from narwhals.typing import IntoFrame
     from narwhals.typing import SizeUnit
-    from narwhals.utils import Implementation
 
     PS = ParamSpec("PS")
 
@@ -498,10 +498,16 @@ def __arrow_c_stream__(self: Self, requested_schema: object | None = None) -> ob
         pa_table = self.to_arrow()
         return pa_table.__arrow_c_stream__(requested_schema=requested_schema)
 
-    def lazy(self: Self) -> LazyFrame[Any]:
+    def lazy(self: Self, *, backend: Implementation | None = None) -> LazyFrame[Any]:
         """Lazify the DataFrame (if possible).
 
-        If a library does not support lazy execution, then this is a no-op.
+        If `backend` is specified, then a conversion between different backends
+        might be triggered.
+        If a library does not support lazy execution and `backend` is not specified,
+        then this is a no-op.
+
+        Arguments:
+            backend: specifies which Implementation to convert to.
 
         Returns:
             A new LazyFrame.
@@ -545,7 +551,21 @@ def lazy(self: Self) -> LazyFrame[Any]:
             bar: [[6,7,8]]
             ham: [["a","b","c"]]
         """
-        return self._lazyframe(self._compliant_frame.lazy(), level="lazy")
+        supported_backends = (
+            Implementation.DASK,
+            Implementation.DUCKDB,
+            Implementation.POLARS,
+        )
+        if backend is not None and backend not in supported_backends:
+            msg = (
+                "Not supported backend"
+                f"\n\nExpected one of {supported_backends}, got {backend}"
+            )
+            raise ValueError(msg)
+        return self._lazyframe(
+            self._compliant_frame.lazy(backend=backend),
+            level="lazy",
+        )
 
     def to_native(self: Self) -> DataFrameT:
         """Convert Narwhals DataFrame to native one.

narwhals/stable/v1/__init__.py

@@ -167,15 +167,21 @@ def __getitem__(self: Self, item: tuple[slice, slice]) -> Self: ...
     def __getitem__(self: Self, item: Any) -> Any:
         return super().__getitem__(item)
 
-    def lazy(self: Self) -> LazyFrame[Any]:
+    def lazy(self: Self, *, backend: Implementation | None = None) -> LazyFrame[Any]:
         """Lazify the DataFrame (if possible).
 
-        If a library does not support lazy execution, then this is a no-op.
+        If `backend` is specified, then a conversion between different backends
+        might be triggered.
+        If a library does not support lazy execution and `backend` is not specified,
+        then this is a no-op.
+
+        Arguments:
+            backend: specifies which Implementation to convert to.
 
         Returns:
             A new LazyFrame.
         """
-        return super().lazy()  # type: ignore[return-value]
+        return super().lazy(backend=backend)  # type: ignore[return-value]
 
     # Not sure what mypy is complaining about, probably some fancy
     # thing that I need to understand category theory for

tests/frame/lazy_test.py

@@ -2,17 +2,52 @@
 
 from typing import TYPE_CHECKING
 
+import pytest
+
 import narwhals as nw
 import narwhals.stable.v1 as nw_v1
+from narwhals.utils import Implementation
 
 if TYPE_CHECKING:
     from tests.utils import ConstructorEager
 
 
+data = {"a": [1, 2, 3]}
+
+
 def test_lazy(constructor_eager: ConstructorEager) -> None:
-    df = nw.from_native(constructor_eager({"a": [1, 2, 3]}), eager_only=True)
+    df = nw.from_native(constructor_eager(data), eager_only=True)
     result = df.lazy()
     assert isinstance(result, nw.LazyFrame)
-    df = nw_v1.from_native(constructor_eager({"a": [1, 2, 3]}), eager_only=True)
+    df = nw_v1.from_native(constructor_eager(data), eager_only=True)
     result = df.lazy()
     assert isinstance(result, nw_v1.LazyFrame)
+
+
+@pytest.mark.parametrize(
+    "backend", [Implementation.POLARS, Implementation.DUCKDB, Implementation.DASK]
+)
+def test_lazy_backend(
+    request: pytest.FixtureRequest,
+    constructor_eager: ConstructorEager,
+    backend: Implementation,
+) -> None:
+    if "modin" in str(constructor_eager):
+        request.applymarker(pytest.mark.xfail)
+    if backend is Implementation.DASK:
+        pytest.importorskip("dask")
+    if backend is Implementation.DUCKDB:
+        pytest.importorskip("duckdb")
+    df = nw.from_native(constructor_eager(data), eager_only=True)
+    result = df.lazy(backend=backend)
+    assert isinstance(result, nw.LazyFrame)
+    assert result.implementation == backend
+
+
+@pytest.mark.parametrize("backend", [Implementation.PANDAS])
+def test_lazy_backend_invalid(
+    constructor_eager: ConstructorEager, backend: Implementation
+) -> None:
+    df = nw.from_native(constructor_eager(data), eager_only=True)
+    with pytest.raises(ValueError, match="Not supported backend"):
+        df.lazy(backend=backend)