feat(expr-ir): Add concat

Child of #2572
Compliant-level was already there - just need to get the tests in place now

Author: dangotbanned

narwhals/_plan/dataframe.py

@@ -6,6 +6,8 @@
 from narwhals._plan._expansion import expand_selector_irs_names, prepare_projection
 from narwhals._plan._guards import is_series
 from narwhals._plan.common import ensure_seq_str, normalize_target_file, temp
+from narwhals._plan.compliant.dataframe import EagerDataFrame
+from narwhals._plan.compliant.namespace import EagerNamespace
 from narwhals._plan.group_by import GroupBy, Grouped
 from narwhals._plan.options import ExplodeOptions, SortMultipleOptions
 from narwhals._plan.series import Series
@@ -48,11 +50,7 @@
 
     from narwhals._native import NativeSeries
     from narwhals._plan.arrow.typing import NativeArrowDataFrame
-    from narwhals._plan.compliant.dataframe import (
-        CompliantDataFrame,
-        CompliantFrame,
-        EagerDataFrame,
-    )
+    from narwhals._plan.compliant.dataframe import CompliantFrame, EagerDataFrame
     from narwhals._plan.compliant.namespace import EagerNamespace
     from narwhals._plan.compliant.series import CompliantSeries
     from narwhals._typing import Arrow, _EagerAllowedImpl
@@ -224,7 +222,17 @@ def _dataframe_from_dict(
 class DataFrame(
     BaseFrame[NativeDataFrameT_co], Generic[NativeDataFrameT_co, NativeSeriesT]
 ):
-    _compliant: CompliantDataFrame[IncompleteCyclic, NativeDataFrameT_co, NativeSeriesT]
+    _compliant: EagerDataFrame[IncompleteCyclic, NativeDataFrameT_co, NativeSeriesT]
+
+    def __narwhals_namespace__(
+        self,
+    ) -> EagerNamespace[
+        EagerDataFrame[Any, NativeDataFrameT_co, NativeSeriesT],
+        CompliantSeries[NativeSeriesT],
+        Any,
+        Any,
+    ]:
+        return self._compliant.__narwhals_namespace__()
 
     @property
     def implementation(self) -> _EagerAllowedImpl:

narwhals/_plan/functions.py

@@ -9,6 +9,7 @@
 from narwhals._plan import _guards, _parse, common, expressions as ir, selectors as cs
 from narwhals._plan._dispatch import get_dispatch_name
 from narwhals._plan.compliant import io as _io
+from narwhals._plan.compliant.typing import namespace
 from narwhals._plan.exceptions import (
     list_literal_error,
     unsupported_backend_operation_error,
@@ -35,6 +36,8 @@
 from narwhals.exceptions import ComputeError, InvalidOperationError
 
 if TYPE_CHECKING:
+    from collections.abc import Iterable
+
     import pyarrow as pa
     from typing_extensions import TypeAlias
 
@@ -50,6 +53,7 @@
     from narwhals._plan.expr import Expr
     from narwhals._plan.series import Series
     from narwhals._plan.typing import (
+        DataFrameT,
         IntoExpr,
         IntoExprColumn,
         NativeDataFrameT,
@@ -61,6 +65,7 @@
     from narwhals.typing import (
         Backend,
         ClosedInterval,
+        ConcatMethod,
         EagerAllowed,
         FileSource,
         IntoBackend,
@@ -78,6 +83,9 @@
         t.Any,
     ]
     CompliantDF: TypeAlias = CompliantDataFrame[t.Any, NativeDataFrameT, NativeSeriesT]
+
+    T = t.TypeVar("T")
+
 Incomplete: TypeAlias = t.Any
 _dtypes: Final = Version.MAIN.dtypes
 
@@ -429,6 +437,27 @@ def linear_space(
     )
 
 
+def _ensure_same_frame(items: list[T], /) -> list[T]:
+    item_0_tp = type(items[0])
+    if builtins.all(isinstance(item, item_0_tp) for item in items):
+        return items
+    msg = f"The items to concatenate should either all be eager, or all lazy, got: {[type(item) for item in items]}"
+    raise TypeError(msg)
+
+
+def concat(items: Iterable[DataFrameT], *, how: ConcatMethod = "vertical") -> DataFrameT:
+    elems = list(items)
+    if not elems:
+        msg = "Cannot concat empty list"
+        raise ValueError(msg)
+    if how not in {"horizontal", "vertical", "diagonal"}:  # pragma: no cover
+        msg = "Only vertical, horizontal and diagonal concatenations are supported."
+        raise NotImplementedError(msg)
+    elems = _ensure_same_frame(elems)
+    compliant = namespace(elems[0]).concat((df._compliant for df in elems), how=how)
+    return elems[0]._with_compliant(compliant)
+
+
 @t.overload
 def read_csv(
     source: FileSource, *, backend: Arrow, **kwds: t.Any