feat: LazyFrame.collect with backend and **kwargs

narwhals/_arrow/dataframe.py

@@ -17,12 +17,15 @@
 from narwhals._arrow.utils import native_to_narwhals_dtype
 from narwhals._arrow.utils import select_rows
 from narwhals._expression_parsing import evaluate_into_exprs
+from narwhals.dependencies import get_pandas
+from narwhals.dependencies import get_polars
 from narwhals.dependencies import is_numpy_array
 from narwhals.utils import Implementation
 from narwhals.utils import check_column_exists
 from narwhals.utils import generate_temporary_column_name
 from narwhals.utils import is_sequence_but_not_str
 from narwhals.utils import parse_columns_to_drop
+from narwhals.utils import parse_version
 from narwhals.utils import scale_bytes
 from narwhals.utils import validate_backend_version
 
@@ -522,12 +525,46 @@ def tail(self: Self, n: int) -> Self:
     def lazy(self: Self) -> Self:
         return self
 
-    def collect(self: Self) -> ArrowDataFrame:
-        return ArrowDataFrame(
-            self._native_frame,
-            backend_version=self._backend_version,
-            version=self._version,
-        )
+    def collect(
+        self: Self,
+        backend: ModuleType | Implementation | str | None,
+        **kwargs: Any,
+    ) -> CompliantDataFrame:
+        if backend in (None, "pyarrow", Implementation.PYARROW, pa):
+            from narwhals._arrow.dataframe import ArrowDataFrame
+
+            return ArrowDataFrame(
+                native_dataframe=self._native_frame,
+                backend_version=self._backend_version,
+                version=self._version,
+            )
+
+        elif backend in ("pandas", Implementation.PANDAS, get_pandas()):
+            import pandas as pd  # ignore-banned-import
+
+            from narwhals._pandas_like.dataframe import PandasLikeDataFrame
+
+            return PandasLikeDataFrame(
+                native_dataframe=self._native_frame.to_pandas(),
+                implementation=Implementation.PANDAS,
+                backend_version=parse_version(pd.__version__),
+                version=self._version,
+            )
+
+        elif backend in ("polars", Implementation.POLARS, get_polars()):
+            import polars as pl  # ignore-banned-import
+
+            from narwhals._polars.dataframe import PolarsDataFrame
+
+            return PolarsDataFrame(
+                df=pl.from_arrow(self._native_frame),  # type: ignore[arg-type]
+                backend_version=parse_version(pl.__version__),
+                version=self._version,
+            )
+
+        else:
+            msg = f"Unsupported `backend` value: {backend}"
+            raise ValueError(msg)
 
     def clone(self: Self) -> Self:
         msg = "clone is not yet supported on PyArrow tables"

narwhals/_dask/dataframe.py

@@ -13,6 +13,9 @@
 from narwhals._dask.utils import parse_exprs_and_named_exprs
 from narwhals._pandas_like.utils import native_to_narwhals_dtype
 from narwhals._pandas_like.utils import select_columns_by_name
+from narwhals.dependencies import get_polars
+from narwhals.dependencies import get_pyarrow
+from narwhals.typing import CompliantDataFrame
 from narwhals.typing import CompliantLazyFrame
 from narwhals.utils import Implementation
 from narwhals.utils import check_column_exists
@@ -29,7 +32,6 @@
     from narwhals._dask.expr import DaskExpr
     from narwhals._dask.group_by import DaskLazyGroupBy
     from narwhals._dask.namespace import DaskNamespace
-    from narwhals._pandas_like.dataframe import PandasLikeDataFrame
     from narwhals.dtypes import DType
     from narwhals.utils import Version
 
@@ -79,16 +81,50 @@ def with_columns(self: Self, *exprs: DaskExpr, **named_exprs: DaskExpr) -> Self:
         df = df.assign(**new_series)
         return self._from_native_frame(df)
 
-    def collect(self: Self) -> PandasLikeDataFrame:
-        from narwhals._pandas_like.dataframe import PandasLikeDataFrame
+    def collect(
+        self: Self,
+        backend: ModuleType | Implementation | str | None,
+        **kwargs: Any,
+    ) -> CompliantDataFrame:
+        import pandas as pd
 
-        result = self._native_frame.compute()
-        return PandasLikeDataFrame(
-            result,
-            implementation=Implementation.PANDAS,
-            backend_version=parse_version(pd.__version__),
-            version=self._version,
-        )
+        result = self._native_frame.compute(**kwargs)
+
+        if backend in (None, "pandas", Implementation.PANDAS, pd):
+            from narwhals._pandas_like.dataframe import PandasLikeDataFrame
+
+            return PandasLikeDataFrame(
+                result,
+                implementation=Implementation.PANDAS,
+                backend_version=parse_version(pd.__version__),
+                version=self._version,
+            )
+
+        elif backend in ("polars", Implementation.POLARS, get_polars()):
+            import polars as pl  # ignore-banned-import
+
+            from narwhals._polars.dataframe import PolarsDataFrame
+
+            return PolarsDataFrame(
+                pl.from_pandas(result),
+                backend_version=parse_version(pl.__version__),
+                version=self._version,
+            )
+
+        elif backend in ("pyarrow", Implementation.PYARROW, get_pyarrow()):
+            import pyarrow as pa  # ignore-banned-import
+
+            from narwhals._arrow.dataframe import ArrowDataFrame
+
+            return ArrowDataFrame(
+                pa.Table.from_pandas(result),
+                backend_version=parse_version(pa.__version__),
+                version=self._version,
+            )
+
+        else:
+            msg = f"Unsupported `backend` value: {backend}"
+            raise ValueError(msg)
 
     @property
     def columns(self: Self) -> list[str]:

narwhals/_duckdb/dataframe.py

@@ -13,7 +13,11 @@
 from narwhals._duckdb.utils import native_to_narwhals_dtype
 from narwhals._duckdb.utils import parse_exprs_and_named_exprs
 from narwhals.dependencies import get_duckdb
+from narwhals.dependencies import get_pandas
+from narwhals.dependencies import get_polars
+from narwhals.dependencies import get_pyarrow
 from narwhals.exceptions import ColumnNotFoundError
+from narwhals.typing import CompliantDataFrame
 from narwhals.utils import Implementation
 from narwhals.utils import Version
 from narwhals.utils import generate_temporary_column_name
@@ -79,22 +83,50 @@ def __getitem__(self: Self, item: str) -> DuckDBInterchangeSeries:
             self._native_frame.select(item), version=self._version
         )
 
-    def collect(self: Self) -> pa.Table:
-        try:
+    def collect(
+        self: Self,
+        backend: ModuleType | Implementation | str | None,
+        **kwargs: Any,
+    ) -> CompliantDataFrame:
+        if backend in (None, "pyarrow", Implementation.PYARROW, get_pyarrow()):
             import pyarrow as pa  # ignore-banned-import
-        except ModuleNotFoundError as exc:  # pragma: no cover
-            msg = "PyArrow>=11.0.0 is required to collect `LazyFrame` backed by DuckDcollect `LazyFrame` backed by DuckDB"
-            raise ModuleNotFoundError(msg) from exc
 
-        from narwhals._arrow.dataframe import ArrowDataFrame
+            from narwhals._arrow.dataframe import ArrowDataFrame
 
-        return ArrowDataFrame(
-            native_dataframe=self._native_frame.arrow(),
-            backend_version=parse_version(pa.__version__),
-            version=self._version,
-        )
+            return ArrowDataFrame(
+                native_dataframe=self._native_frame.arrow(),
+                backend_version=parse_version(pa.__version__),
+                version=self._version,
+            )
+
+        elif backend in ("pandas", Implementation.PANDAS, get_pandas()):
+            import pandas as pd  # ignore-banned-import
+
+            from narwhals._pandas_like.dataframe import PandasLikeDataFrame
+
+            return PandasLikeDataFrame(
+                native_dataframe=self._native_frame.df(),
+                implementation=Implementation.PANDAS,
+                backend_version=parse_version(pd.__version__),
+                version=self._version,
+            )
+
+        elif backend in ("polars", Implementation.POLARS, get_polars()):
+            import polars as pl  # ignore-banned-import
+
+            from narwhals._polars.dataframe import PolarsDataFrame
+
+            return PolarsDataFrame(
+                df=self._native_frame.pl(),
+                backend_version=parse_version(pl.__version__),
+                version=self._version,
+            )
+
+        else:
+            msg = f"Unsupported `backend` value: {backend}"
+            raise ValueError(msg)
 
-    def head(self: Self, n: int) -> Self:
+    def head(self, n: int) -> Self:
         return self._from_native_frame(self._native_frame.limit(n))
 
     def simple_select(self, *column_names: str) -> Self:

narwhals/_pandas_like/dataframe.py

@@ -18,13 +18,17 @@
 from narwhals._pandas_like.utils import pivot_table
 from narwhals._pandas_like.utils import rename
 from narwhals._pandas_like.utils import select_columns_by_name
+from narwhals.dependencies import get_pandas
+from narwhals.dependencies import get_polars
+from narwhals.dependencies import get_pyarrow
 from narwhals.dependencies import is_numpy_array
 from narwhals.utils import Implementation
 from narwhals.utils import check_column_exists
 from narwhals.utils import generate_temporary_column_name
 from narwhals.utils import import_dtypes_module
 from narwhals.utils import is_sequence_but_not_str
 from narwhals.utils import parse_columns_to_drop
+from narwhals.utils import parse_version
 from narwhals.utils import scale_bytes
 from narwhals.utils import validate_backend_version
 
@@ -501,13 +505,66 @@ def sort(
         )
 
     # --- convert ---
-    def collect(self: Self) -> PandasLikeDataFrame:
-        return PandasLikeDataFrame(
-            self._native_frame,
-            implementation=self._implementation,
-            backend_version=self._backend_version,
-            version=self._version,
-        )
+    def collect(
+        self: Self,
+        backend: ModuleType | Implementation | str | None,
+        **kwargs: Any,
+    ) -> CompliantDataFrame:
+        if backend is None:
+            return PandasLikeDataFrame(
+                self._native_frame,
+                implementation=self._implementation,
+                backend_version=self._backend_version,
+                version=self._version,
+            )
+
+        elif backend in ("pandas", Implementation.PANDAS, get_pandas()):
+            import pandas as pd  # ignore-banned-import
+
+            return PandasLikeDataFrame(
+                self.to_pandas(),
+                implementation=Implementation.PANDAS,
+                backend_version=parse_version(pd.__version__),
+                version=self._version,
+            )
+
+        elif backend in ("pyarrow", Implementation.PYARROW, get_pyarrow()):
+            import pyarrow as pa  # ignore-banned-import
+
+            from narwhals._arrow.dataframe import ArrowDataFrame
+
+            if self._implementation is Implementation.CUDF:
+                pa_native = self._native_frame.to_arrow(preserve_index=False)
+            else:
+                pa_native = pa.Table.from_pandas(self._native_frame)
+
+            return ArrowDataFrame(
+                native_dataframe=pa_native,
+                backend_version=parse_version(pa.__version__),
+                version=self._version,
+            )
+
+        elif backend in ("polars", Implementation.POLARS, get_polars()):
+            import polars as pl  # ignore-banned-import
+
+            from narwhals._polars.dataframe import PolarsDataFrame
+
+            if self._implementation is Implementation.PANDAS:
+                pl_native = pl.from_pandas(self._native_frame)
+            elif self._implementation is Implementation.CUDF:  # pragma: no cover
+                pl_native = pl.from_pandas(self._native_frame.to_pandas())
+            elif self._implementation is Implementation.MODIN:
+                pl_native = pl.from_pandas(self._native_frame._to_pandas())
+
+            return PolarsDataFrame(
+                df=pl_native,
+                backend_version=parse_version(pl.__version__),
+                version=self._version,
+            )
+
+        else:
+            msg = f"Unsupported `backend` value: {backend}"
+            raise ValueError(msg)
 
     # --- actions ---
     def group_by(self: Self, *keys: str, drop_null_keys: bool) -> PandasLikeGroupBy:

narwhals/_polars/dataframe.py

@@ -12,10 +12,13 @@
 from narwhals._polars.utils import convert_str_slice_to_int_slice
 from narwhals._polars.utils import extract_args_kwargs
 from narwhals._polars.utils import native_to_narwhals_dtype
+from narwhals.dependencies import get_pandas
+from narwhals.dependencies import get_pyarrow
 from narwhals.exceptions import ColumnNotFoundError
 from narwhals.utils import Implementation
 from narwhals.utils import is_sequence_but_not_str
 from narwhals.utils import parse_columns_to_drop
+from narwhals.utils import parse_version
 from narwhals.utils import validate_backend_version
 
 if TYPE_CHECKING:
@@ -29,6 +32,7 @@
     from narwhals._polars.group_by import PolarsLazyGroupBy
     from narwhals._polars.series import PolarsSeries
     from narwhals.dtypes import DType
+    from narwhals.typing import CompliantDataFrame
     from narwhals.utils import Version
 
     T = TypeVar("T")
@@ -411,19 +415,53 @@ def collect_schema(self: Self) -> dict[str, DType]:
                 for name, dtype in self._native_frame.collect_schema().items()
             }
 
-    def collect(self: Self) -> PolarsDataFrame:
+    def collect(
+        self: Self,
+        backend: ModuleType | Implementation | str | None,
+        **kwargs: Any,
+    ) -> CompliantDataFrame:
         import polars as pl
 
         try:
-            result = self._native_frame.collect()
+            result = self._native_frame.collect(**kwargs)
         except pl.exceptions.ColumnNotFoundError as e:
             raise ColumnNotFoundError(str(e)) from e
 
-        return PolarsDataFrame(
-            result,
-            backend_version=self._backend_version,
-            version=self._version,
-        )
+        if backend in (None, "polars", Implementation.POLARS, pl):
+            from narwhals._polars.dataframe import PolarsDataFrame
+
+            return PolarsDataFrame(
+                result,
+                backend_version=self._backend_version,
+                version=self._version,
+            )
+
+        elif backend in ("pandas", Implementation.PANDAS, get_pandas()):
+            import pandas as pd  # ignore-banned-import
+
+            from narwhals._pandas_like.dataframe import PandasLikeDataFrame
+
+            return PandasLikeDataFrame(
+                result.to_pandas(),
+                implementation=Implementation.PANDAS,
+                backend_version=parse_version(pd.__version__),
+                version=self._version,
+            )
+
+        elif backend in ("pyarrow", Implementation.PYARROW, get_pyarrow()):
+            import pyarrow as pa  # ignore-banned-import
+
+            from narwhals._arrow.dataframe import ArrowDataFrame
+
+            return ArrowDataFrame(
+                result.to_arrow(),
+                backend_version=parse_version(pa.__version__),
+                version=self._version,
+            )
+
+        else:
+            msg = f"Unsupported `backend` value: {backend}"
+            raise ValueError(msg)
 
     def group_by(self: Self, *by: str, drop_null_keys: bool) -> PolarsLazyGroupBy:
         from narwhals._polars.group_by import PolarsLazyGroupBy