Repository: narwhals-dev/narwhals
Pull request #1856 (2025-01-23 14:09:31)

chore: parse strings as columns names at narwhals-level
closes https://github.com/narwhals-dev/narwhals/issues/1390 (completely by accident 😳 😄 )

<!--
# Thanks for contributing a pull request! 
## Please make sure you see our contribution guidelines: https://github.com/narwhals-dev/narwhals/blob/main/CONTRIBUTING.md
-->

## What type of PR is this? (check all applicable)

- [ ] 💾 Refactor
- [ ] ✨ Feature
- [ ] 🐛 Bug Fix
- [ ] 🔧 Optimization
- [ ] 📝 Documentation
- [ ] ✅ Test
- [ ] 🐳 Other

## Related issues

- Related issue #\<issue number\>
- Closes #\<issue number\>

## Checklist

- [ ] Code follows style guide (ruff)
- [ ] Tests added
- [ ] Documented the changes

## If you have comments or can explain your changes, please do so below


MarcoGorelli: thanks for your review!

> The asymmetry in some method specs, between Narwhals and Compliant objects. The expected simple_select method (just needs documentation I guess)

It's true that there's some asymmetry, but I think that's OK - in my head, the separation is:
- compliant level: strict and minimal
- narwhals (user-facing) level: more ergonomic

> Are we adding any extra overhead?

From tests I've done locally it doesn't look like it. Theoretically i'd expect it to be slightly lower, if we're just doing a single `isinstance(expr, str)` check upfront as opposed to repeatedly when parsing expressions at the compliant level

Patch:
diff --git a/docs/extending.md b/docs/extending.md
index 588e234f..c0076c58 100644
--- a/docs/extending.md
+++ b/docs/extending.md
@@ -107,22 +107,20 @@ the next section for what else you can do.
 We love open source, but we're not "open source absolutists". If you're unable to open
 source you library, then this is how you can make your library compatible with Narwhals.
 
-Make sure that, in addition to the public Narwhals API, you also define:
-
-  - `DataFrame.__narwhals_dataframe__`: return an object which implements public methods
-    from `Narwhals.DataFrame`
-  - `DataFrame.__narwhals_namespace__`: return an object which implements public top-level
-    functions from `narwhals` (e.g. `narwhals.col`, `narwhals.concat`, ...)
-  - `DataFrame.__native_namespace__`: return a native namespace object which must have a
-    `from_dict` method
-  - `LazyFrame.__narwhals_lazyframe__`: return an object which implements public methods
-    from `Narwhals.LazyFrame`
-  - `LazyFrame.__narwhals_namespace__`: return an object which implements public top-level
-    functions from `narwhals` (e.g. `narwhals.col`, `narwhals.concat`, ...)
-  - `LazyFrame.__native_namespace__`: return a native namespace object which must have a
-    `from_dict` method
-  - `Series.__narwhals_series__`: return an object which implements public methods
-    from `Narwhals.Series`
+Make sure that you also define:
+
+  - `DataFrame.__narwhals_dataframe__`: return an object which implements methods from the
+    `CompliantDataFrame` protocol in  `narwhals/typing.py`.
+  - `DataFrame.__narwhals_namespace__`: return an object which implements methods from the
+    `CompliantNamespace` protocol in `narwhals/typing.py`.
+  - `DataFrame.__native_namespace__`: return the object's native namespace.
+  - `LazyFrame.__narwhals_lazyframe__`: return an object which implements methods from the
+    `CompliantLazyFrame` protocol in  `narwhals/typing.py`.
+  - `LazyFrame.__narwhals_namespace__`: return an object which implements methods from the
+    `CompliantNamespace` protocol in `narwhals/typing.py`.
+  - `LazyFrame.__native_namespace__`: return the object's native namespace.
+  - `Series.__narwhals_series__`: return an object which implements methods from the
+    `CompliantSeries` protocol in `narwhals/typing.py`.
 
   If your library doesn't distinguish between lazy and eager, then it's OK for your dataframe
   object to implement both `__narwhals_dataframe__` and `__narwhals_lazyframe__`. In fact,
diff --git a/narwhals/_arrow/dataframe.py b/narwhals/_arrow/dataframe.py
index c75f4502..6e3d73c6 100644
--- a/narwhals/_arrow/dataframe.py
+++ b/narwhals/_arrow/dataframe.py
@@ -3,7 +3,6 @@ from __future__ import annotations
 from itertools import chain
 from typing import TYPE_CHECKING
 from typing import Any
-from typing import Iterable
 from typing import Iterator
 from typing import Literal
 from typing import Sequence
@@ -21,7 +20,6 @@ from narwhals._expression_parsing import evaluate_into_exprs
 from narwhals.dependencies import is_numpy_array
 from narwhals.utils import Implementation
 from narwhals.utils import check_column_exists
-from narwhals.utils import flatten
 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
@@ -292,6 +290,9 @@ class ArrowDataFrame(CompliantDataFrame, CompliantLazyFrame):
     def columns(self: Self) -> list[str]:
         return self._native_frame.schema.names  # type: ignore[no-any-return]
 
+    def simple_select(self, *column_names: str) -> Self:
+        return self._from_native_frame(self._native_frame.select(list(column_names)))
+
     def select(self: Self, *exprs: IntoArrowExpr, **named_exprs: IntoArrowExpr) -> Self:
         new_series = evaluate_into_exprs(self, *exprs, **named_exprs)
         if not new_series:
@@ -405,21 +406,19 @@ class ArrowDataFrame(CompliantDataFrame, CompliantLazyFrame):
 
     def sort(
         self: Self,
-        by: str | Iterable[str],
-        *more_by: str,
+        *by: str,
         descending: bool | Sequence[bool],
         nulls_last: bool,
     ) -> Self:
-        flat_keys = flatten([*flatten([by]), *more_by])
         df = self._native_frame
 
         if isinstance(descending, bool):
             order = "descending" if descending else "ascending"
-            sorting = [(key, order) for key in flat_keys]
+            sorting = [(key, order) for key in by]
         else:
             sorting = [
                 (key, "descending" if is_descending else "ascending")
-                for key, is_descending in zip(flat_keys, descending)
+                for key, is_descending in zip(by, descending)
             ]
 
         null_placement = "at_end" if nulls_last else "at_start"
@@ -663,7 +662,7 @@ class ArrowDataFrame(CompliantDataFrame, CompliantLazyFrame):
 
             return self._from_native_frame(pc.take(df, keep_idx))
 
-        keep_idx = self.select(*subset).is_unique()
+        keep_idx = self.simple_select(*subset).is_unique()
         return self.filter(keep_idx)
 
     def gather_every(self: Self, n: int, offset: int) -> Self:
diff --git a/narwhals/_arrow/expr.py b/narwhals/_arrow/expr.py
index 733cc665..2d911c77 100644
--- a/narwhals/_arrow/expr.py
+++ b/narwhals/_arrow/expr.py
@@ -395,7 +395,7 @@ class ArrowExpr(CompliantExpr[ArrowSeries]):
                 raise AnonymousExprError.from_expr_name(msg)
 
             tmp = df.group_by(*keys, drop_null_keys=False).agg(self)
-            tmp = df.select(*keys).join(
+            tmp = df.simple_select(*keys).join(
                 tmp, how="left", left_on=keys, right_on=keys, suffix="_right"
             )
             return [tmp[name] for name in self._output_names]
diff --git a/narwhals/_arrow/group_by.py b/narwhals/_arrow/group_by.py
index fff5889d..86ce73cb 100644
--- a/narwhals/_arrow/group_by.py
+++ b/narwhals/_arrow/group_by.py
@@ -11,7 +11,6 @@ import pyarrow as pa
 import pyarrow.compute as pc
 
 from narwhals._expression_parsing import is_simple_aggregation
-from narwhals._expression_parsing import parse_into_exprs
 from narwhals.exceptions import AnonymousExprError
 from narwhals.utils import generate_temporary_column_name
 from narwhals.utils import remove_prefix
@@ -20,8 +19,8 @@ if TYPE_CHECKING:
     from typing_extensions import Self
 
     from narwhals._arrow.dataframe import ArrowDataFrame
+    from narwhals._arrow.expr import ArrowExpr
     from narwhals._arrow.series import ArrowSeries
-    from narwhals._arrow.typing import IntoArrowExpr
     from narwhals.typing import CompliantExpr
 
 POLARS_TO_ARROW_AGGREGATIONS = {
@@ -49,16 +48,7 @@ class ArrowGroupBy:
         self._keys = list(keys)
         self._grouped = pa.TableGroupBy(self._df._native_frame, list(self._keys))
 
-    def agg(
-        self: Self,
-        *aggs: IntoArrowExpr,
-        **named_aggs: IntoArrowExpr,
-    ) -> ArrowDataFrame:
-        exprs = parse_into_exprs(
-            *aggs,
-            namespace=self._df.__narwhals_namespace__(),
-            **named_aggs,
-        )
+    def agg(self: Self, *exprs: ArrowExpr) -> ArrowDataFrame:
         for expr in exprs:
             if expr._output_names is None:
                 msg = "group_by.agg"
@@ -93,7 +83,7 @@ class ArrowGroupBy:
                             table.filter(pc.equal(table[col_token], v)).drop([col_token])
                         )
                     )
-                    .select(*self._keys)
+                    .simple_select(*self._keys)
                     .head(1)
                     .iter_rows(named=False, buffer_size=512)
                 ),
diff --git a/narwhals/_arrow/namespace.py b/narwhals/_arrow/namespace.py
index 984fc354..f02ee44c 100644
--- a/narwhals/_arrow/namespace.py
+++ b/narwhals/_arrow/namespace.py
@@ -361,15 +361,11 @@ class ArrowNamespace(CompliantNamespace[ArrowSeries]):
 
     def concat_str(
         self: Self,
-        exprs: Iterable[IntoArrowExpr],
-        *more_exprs: IntoArrowExpr,
+        *exprs: IntoArrowExpr,
         separator: str,
         ignore_nulls: bool,
     ) -> ArrowExpr:
-        parsed_exprs = [
-            *parse_into_exprs(*exprs, namespace=self),
-            *parse_into_exprs(*more_exprs, namespace=self),
-        ]
+        parsed_exprs = parse_into_exprs(*exprs, namespace=self)
         dtypes = import_dtypes_module(self._version)
 
         def func(df: ArrowDataFrame) -> list[ArrowSeries]:
@@ -399,7 +395,6 @@ class ArrowNamespace(CompliantNamespace[ArrowSeries]):
             output_names=reduce_output_names(parsed_exprs),
             kwargs={
                 "exprs": exprs,
-                "more_exprs": more_exprs,
                 "separator": separator,
                 "ignore_nulls": ignore_nulls,
             },
diff --git a/narwhals/_arrow/series.py b/narwhals/_arrow/series.py
index e9a82d1d..fb29311e 100644
--- a/narwhals/_arrow/series.py
+++ b/narwhals/_arrow/series.py
@@ -745,7 +745,7 @@ class ArrowSeries(CompliantSeries):
             pa.Table.from_arrays(columns, names=cols),
             backend_version=self._backend_version,
             version=self._version,
-        ).select(*output_order)
+        ).simple_select(*output_order)
 
     def quantile(
         self: Self,
diff --git a/narwhals/_arrow/typing.py b/narwhals/_arrow/typing.py
index 9d7130a6..38b8c9be 100644
--- a/narwhals/_arrow/typing.py
+++ b/narwhals/_arrow/typing.py
@@ -14,4 +14,4 @@ if TYPE_CHECKING:
     from narwhals._arrow.expr import ArrowExpr
     from narwhals._arrow.series import ArrowSeries
 
-    IntoArrowExpr: TypeAlias = Union[ArrowExpr, str, ArrowSeries]
+    IntoArrowExpr: TypeAlias = Union[ArrowExpr, ArrowSeries]
diff --git a/narwhals/_dask/dataframe.py b/narwhals/_dask/dataframe.py
index 010dbc4f..165014d8 100644
--- a/narwhals/_dask/dataframe.py
+++ b/narwhals/_dask/dataframe.py
@@ -3,7 +3,6 @@ from __future__ import annotations
 from itertools import chain
 from typing import TYPE_CHECKING
 from typing import Any
-from typing import Iterable
 from typing import Literal
 from typing import Sequence
 
@@ -17,7 +16,6 @@ from narwhals._pandas_like.utils import select_columns_by_name
 from narwhals.typing import CompliantLazyFrame
 from narwhals.utils import Implementation
 from narwhals.utils import check_column_exists
-from narwhals.utils import flatten
 from narwhals.utils import generate_temporary_column_name
 from narwhals.utils import parse_columns_to_drop
 from narwhals.utils import parse_version
@@ -31,7 +29,6 @@ if TYPE_CHECKING:
     from narwhals._dask.expr import DaskExpr
     from narwhals._dask.group_by import DaskLazyGroupBy
     from narwhals._dask.namespace import DaskNamespace
-    from narwhals._dask.typing import IntoDaskExpr
     from narwhals._pandas_like.dataframe import PandasLikeDataFrame
     from narwhals.dtypes import DType
     from narwhals.utils import Version
@@ -107,22 +104,17 @@ class DaskLazyFrame(CompliantLazyFrame):
 
         return self._from_native_frame(self._native_frame.loc[mask])
 
-    def select(
-        self: Self,
-        *exprs: IntoDaskExpr,
-        **named_exprs: IntoDaskExpr,
-    ) -> Self:
-        if exprs and all(isinstance(x, str) for x in exprs) and not named_exprs:
-            # This is a simple slice => fastpath!
-            return self._from_native_frame(
-                select_columns_by_name(
-                    self._native_frame,
-                    list(exprs),  # type: ignore[arg-type]
-                    self._backend_version,
-                    self._implementation,
-                )
+    def simple_select(self: Self, *column_names: str) -> Self:
+        return self._from_native_frame(
+            select_columns_by_name(
+                self._native_frame,
+                list(column_names),
+                self._backend_version,
+                self._implementation,
             )
+        )
 
+    def select(self: Self, *exprs: DaskExpr, **named_exprs: DaskExpr) -> Self:
         new_series = parse_exprs_and_named_exprs(self, *exprs, **named_exprs)
 
         if not new_series:
@@ -211,12 +203,10 @@ class DaskLazyFrame(CompliantLazyFrame):
 
     def sort(
         self: Self,
-        by: str | Iterable[str],
-        *more_by: str,
+        *by: str,
         descending: bool | Sequence[bool],
         nulls_last: bool,
     ) -> Self:
-        flat_keys = flatten([*flatten([by]), *more_by])
         df = self._native_frame
         if isinstance(descending, bool):
             ascending: bool | list[bool] = not descending
@@ -224,7 +214,7 @@ class DaskLazyFrame(CompliantLazyFrame):
             ascending = [not d for d in descending]
         na_position = "last" if nulls_last else "first"
         return self._from_native_frame(
-            df.sort_values(flat_keys, ascending=ascending, na_position=na_position)
+            df.sort_values(list(by), ascending=ascending, na_position=na_position)
         )
 
     def join(
diff --git a/narwhals/_dask/expr.py b/narwhals/_dask/expr.py
index 4306724d..77791628 100644
--- a/narwhals/_dask/expr.py
+++ b/narwhals/_dask/expr.py
@@ -666,7 +666,7 @@ class DaskExpr(CompliantExpr["dx.Series"]):
             if df._native_frame.npartitions == 1:  # pragma: no cover
                 tmp = df.group_by(*keys, drop_null_keys=False).agg(self)
                 tmp_native = (
-                    df.select(*keys)
+                    df.simple_select(*keys)
                     .join(tmp, how="left", left_on=keys, right_on=keys, suffix="_right")
                     ._native_frame
                 )
diff --git a/narwhals/_dask/group_by.py b/narwhals/_dask/group_by.py
index 7b113079..7fd434af 100644
--- a/narwhals/_dask/group_by.py
+++ b/narwhals/_dask/group_by.py
@@ -9,7 +9,6 @@ from typing import Sequence
 import dask.dataframe as dd
 
 from narwhals._expression_parsing import is_simple_aggregation
-from narwhals._expression_parsing import parse_into_exprs
 from narwhals.exceptions import AnonymousExprError
 from narwhals.utils import remove_prefix
 
@@ -23,7 +22,7 @@ if TYPE_CHECKING:
     from typing_extensions import Self
 
     from narwhals._dask.dataframe import DaskLazyFrame
-    from narwhals._dask.typing import IntoDaskExpr
+    from narwhals._dask.expr import DaskExpr
     from narwhals.typing import CompliantExpr
 
 
@@ -89,14 +88,8 @@ class DaskLazyGroupBy:
 
     def agg(
         self: Self,
-        *aggs: IntoDaskExpr,
-        **named_aggs: IntoDaskExpr,
+        *exprs: DaskExpr,
     ) -> DaskLazyFrame:
-        exprs = parse_into_exprs(
-            *aggs,
-            namespace=self._df.__narwhals_namespace__(),
-            **named_aggs,
-        )
         output_names: list[str] = copy(self._keys)
         for expr in exprs:
             if expr._output_names is None:
@@ -135,7 +128,7 @@ def agg_dask(
     """
     if not exprs:
         # No aggregation provided
-        return df.select(*keys).unique(subset=keys)
+        return df.simple_select(*keys).unique(subset=keys)
 
     all_simple_aggs = True
     for expr in exprs:
diff --git a/narwhals/_dask/namespace.py b/narwhals/_dask/namespace.py
index 713d2afe..9c8a48e8 100644
--- a/narwhals/_dask/namespace.py
+++ b/narwhals/_dask/namespace.py
@@ -19,7 +19,6 @@ from narwhals._dask.utils import name_preserving_sum
 from narwhals._dask.utils import narwhals_to_native_dtype
 from narwhals._dask.utils import validate_comparand
 from narwhals._expression_parsing import combine_root_names
-from narwhals._expression_parsing import parse_into_exprs
 from narwhals._expression_parsing import reduce_output_names
 from narwhals.typing import CompliantNamespace
 
@@ -31,7 +30,6 @@ if TYPE_CHECKING:
 
     from typing_extensions import Self
 
-    from narwhals._dask.typing import IntoDaskExpr
     from narwhals.dtypes import DType
     from narwhals.utils import Version
 
@@ -126,57 +124,51 @@ class DaskNamespace(CompliantNamespace["dx.Series"]):
             kwargs={},
         )
 
-    def all_horizontal(self: Self, *exprs: IntoDaskExpr) -> DaskExpr:
-        parsed_exprs = parse_into_exprs(*exprs, namespace=self)
-
+    def all_horizontal(self: Self, *exprs: DaskExpr) -> DaskExpr:
         def func(df: DaskLazyFrame) -> list[dx.Series]:
-            series = [s for _expr in parsed_exprs for s in _expr(df)]
+            series = [s for _expr in exprs for s in _expr(df)]
             return [reduce(lambda x, y: x & y, series).rename(series[0].name)]
 
         return DaskExpr(
             call=func,
-            depth=max(x._depth for x in parsed_exprs) + 1,
+            depth=max(x._depth for x in exprs) + 1,
             function_name="all_horizontal",
-            root_names=combine_root_names(parsed_exprs),
-            output_names=reduce_output_names(parsed_exprs),
+            root_names=combine_root_names(exprs),
+            output_names=reduce_output_names(exprs),
             returns_scalar=False,
             backend_version=self._backend_version,
             version=self._version,
             kwargs={"exprs": exprs},
         )
 
-    def any_horizontal(self: Self, *exprs: IntoDaskExpr) -> DaskExpr:
-        parsed_exprs = parse_into_exprs(*exprs, namespace=self)
-
+    def any_horizontal(self: Self, *exprs: DaskExpr) -> DaskExpr:
         def func(df: DaskLazyFrame) -> list[dx.Series]:
-            series = [s for _expr in parsed_exprs for s in _expr(df)]
+            series = [s for _expr in exprs for s in _expr(df)]
             return [reduce(lambda x, y: x | y, series).rename(series[0].name)]
 
         return DaskExpr(
             call=func,
-            depth=max(x._depth for x in parsed_exprs) + 1,
+            depth=max(x._depth for x in exprs) + 1,
             function_name="any_horizontal",
-            root_names=combine_root_names(parsed_exprs),
-            output_names=reduce_output_names(parsed_exprs),
+            root_names=combine_root_names(exprs),
+            output_names=reduce_output_names(exprs),
             returns_scalar=False,
             backend_version=self._backend_version,
             version=self._version,
             kwargs={"exprs": exprs},
         )
 
-    def sum_horizontal(self: Self, *exprs: IntoDaskExpr) -> DaskExpr:
-        parsed_exprs = parse_into_exprs(*exprs, namespace=self)
-
+    def sum_horizontal(self: Self, *exprs: DaskExpr) -> DaskExpr:
         def func(df: DaskLazyFrame) -> list[dx.Series]:
-            series = [s.fillna(0) for _expr in parsed_exprs for s in _expr(df)]
+            series = [s.fillna(0) for _expr in exprs for s in _expr(df)]
             return [reduce(lambda x, y: x + y, series).rename(series[0].name)]
 
         return DaskExpr(
             call=func,
-            depth=max(x._depth for x in parsed_exprs) + 1,
+            depth=max(x._depth for x in exprs) + 1,
             function_name="sum_horizontal",
-            root_names=combine_root_names(parsed_exprs),
-            output_names=reduce_output_names(parsed_exprs),
+            root_names=combine_root_names(exprs),
+            output_names=reduce_output_names(exprs),
             returns_scalar=False,
             backend_version=self._backend_version,
             version=self._version,
@@ -238,12 +230,10 @@ class DaskNamespace(CompliantNamespace["dx.Series"]):
 
         raise NotImplementedError
 
-    def mean_horizontal(self: Self, *exprs: IntoDaskExpr) -> DaskExpr:
-        parsed_exprs = parse_into_exprs(*exprs, namespace=self)
-
+    def mean_horizontal(self: Self, *exprs: DaskExpr) -> DaskExpr:
         def func(df: DaskLazyFrame) -> list[dx.Series]:
-            series = (s.fillna(0) for _expr in parsed_exprs for s in _expr(df))
-            non_na = (1 - s.isna() for _expr in parsed_exprs for s in _expr(df))
+            series = (s.fillna(0) for _expr in exprs for s in _expr(df))
+            non_na = (1 - s.isna() for _expr in exprs for s in _expr(df))
             return [
                 name_preserving_div(
                     reduce(name_preserving_sum, series),
@@ -253,50 +243,46 @@ class DaskNamespace(CompliantNamespace["dx.Series"]):
 
         return DaskExpr(
             call=func,
-            depth=max(x._depth for x in parsed_exprs) + 1,
+            depth=max(x._depth for x in exprs) + 1,
             function_name="mean_horizontal",
-            root_names=combine_root_names(parsed_exprs),
-            output_names=reduce_output_names(parsed_exprs),
+            root_names=combine_root_names(exprs),
+            output_names=reduce_output_names(exprs),
             returns_scalar=False,
             backend_version=self._backend_version,
             version=self._version,
             kwargs={"exprs": exprs},
         )
 
-    def min_horizontal(self: Self, *exprs: IntoDaskExpr) -> DaskExpr:
-        parsed_exprs = parse_into_exprs(*exprs, namespace=self)
-
+    def min_horizontal(self: Self, *exprs: DaskExpr) -> DaskExpr:
         def func(df: DaskLazyFrame) -> list[dx.Series]:
-            series = [s for _expr in parsed_exprs for s in _expr(df)]
+            series = [s for _expr in exprs for s in _expr(df)]
 
             return [dd.concat(series, axis=1).min(axis=1).rename(series[0].name)]
 
         return DaskExpr(
             call=func,
-            depth=max(x._depth for x in parsed_exprs) + 1,
+            depth=max(x._depth for x in exprs) + 1,
             function_name="min_horizontal",
-            root_names=combine_root_names(parsed_exprs),
-            output_names=reduce_output_names(parsed_exprs),
+            root_names=combine_root_names(exprs),
+            output_names=reduce_output_names(exprs),
             returns_scalar=False,
             backend_version=self._backend_version,
             version=self._version,
             kwargs={"exprs": exprs},
         )
 
-    def max_horizontal(self: Self, *exprs: IntoDaskExpr) -> DaskExpr:
-        parsed_exprs = parse_into_exprs(*exprs, namespace=self)
-
+    def max_horizontal(self: Self, *exprs: DaskExpr) -> DaskExpr:
         def func(df: DaskLazyFrame) -> list[dx.Series]:
-            series = [s for _expr in parsed_exprs for s in _expr(df)]
+            series = [s for _expr in exprs for s in _expr(df)]
 
             return [dd.concat(series, axis=1).max(axis=1).rename(series[0].name)]
 
         return DaskExpr(
             call=func,
-            depth=max(x._depth for x in parsed_exprs) + 1,
+            depth=max(x._depth for x in exprs) + 1,
             function_name="max_horizontal",
-            root_names=combine_root_names(parsed_exprs),
-            output_names=reduce_output_names(parsed_exprs),
+            root_names=combine_root_names(exprs),
+            output_names=reduce_output_names(exprs),
             returns_scalar=False,
             backend_version=self._backend_version,
             version=self._version,
@@ -305,7 +291,7 @@ class DaskNamespace(CompliantNamespace["dx.Series"]):
 
     def when(
         self: Self,
-        *predicates: IntoDaskExpr,
+        *predicates: DaskExpr,
     ) -> DaskWhen:
         plx = self.__class__(backend_version=self._backend_version, version=self._version)
         condition = plx.all_horizontal(*predicates)
@@ -315,19 +301,13 @@ class DaskNamespace(CompliantNamespace["dx.Series"]):
 
     def concat_str(
         self: Self,
-        exprs: Iterable[IntoDaskExpr],
-        *more_exprs: IntoDaskExpr,
+        *exprs: DaskExpr,
         separator: str,
         ignore_nulls: bool,
     ) -> DaskExpr:
-        parsed_exprs = [
-            *parse_into_exprs(*exprs, namespace=self),
-            *parse_into_exprs(*more_exprs, namespace=self),
-        ]
-
         def func(df: DaskLazyFrame) -> list[dx.Series]:
-            series = (s.astype(str) for _expr in parsed_exprs for s in _expr(df))
-            null_mask = [s for _expr in parsed_exprs for s in _expr.is_null()(df)]
+            series = (s.astype(str) for _expr in exprs for s in _expr(df))
+            null_mask = [s for _expr in exprs for s in _expr.is_null()(df)]
 
             if not ignore_nulls:
                 null_mask_result = reduce(lambda x, y: x | y, null_mask)
@@ -353,16 +333,15 @@ class DaskNamespace(CompliantNamespace["dx.Series"]):
 
         return DaskExpr(
             call=func,
-            depth=max(x._depth for x in parsed_exprs) + 1,
+            depth=max(x._depth for x in exprs) + 1,
             function_name="concat_str",
-            root_names=combine_root_names(parsed_exprs),
-            output_names=reduce_output_names(parsed_exprs),
+            root_names=combine_root_names(exprs),
+            output_names=reduce_output_names(exprs),
             returns_scalar=False,
             backend_version=self._backend_version,
             version=self._version,
             kwargs={
                 "exprs": exprs,
-                "more_exprs": more_exprs,
                 "separator": separator,
                 "ignore_nulls": ignore_nulls,
             },
@@ -388,21 +367,16 @@ class DaskWhen:
         self._version = version
 
     def __call__(self: Self, df: DaskLazyFrame) -> Sequence[dx.Series]:
-        from narwhals._expression_parsing import parse_into_expr
-
-        plx = df.__narwhals_namespace__()
-        condition = parse_into_expr(self._condition, namespace=plx)(df)[0]
+        condition = self._condition(df)[0]
         condition = cast("dx.Series", condition)
 
-        try:
-            then_expr = parse_into_expr(self._then_value, namespace=plx)
-        except TypeError:
-            # `self._then_value` is a scalar and can't be converted to an expression
-            value_sequence: Sequence[Any] = [self._then_value]
-            is_scalar = True
+        if isinstance(self._then_value, DaskExpr):
+            is_scalar = self._then_value._returns_scalar
+            value_sequence: Sequence[Any] = self._then_value(df)[0]
         else:
-            is_scalar = then_expr._returns_scalar  # type: ignore[attr-defined]
-            value_sequence = then_expr(df)[0]
+            # `self._then_value` is a scalar
+            value_sequence = [self._then_value]
+            is_scalar = True
 
         if is_scalar:
             _df = condition.to_frame("a")
@@ -416,15 +390,16 @@ class DaskWhen:
 
         if self._otherwise_value is None:
             return [value_series.where(condition)]
-        try:
-            otherwise_expr = parse_into_expr(self._otherwise_value, namespace=plx)
-        except TypeError:
-            # `self._otherwise_value` is a scalar and can't be converted to an expression
+        if not isinstance(self._otherwise_value, DaskExpr):
+            # `self._otherwise_value` is a scalar
             return [value_series.where(condition, self._otherwise_value)]
+
+        otherwise_expr = self._otherwise_value
         otherwise_series = otherwise_expr(df)[0]
 
-        if otherwise_expr._returns_scalar:  # type: ignore[attr-defined]
+        if otherwise_expr._returns_scalar:
             return [value_series.where(condition, otherwise_series[0])]
+
         validate_comparand(condition, otherwise_series)
         return [value_series.where(condition, otherwise_series)]
 
diff --git a/narwhals/_dask/typing.py b/narwhals/_dask/typing.py
deleted file mode 100644
index 23719eac..00000000
--- a/narwhals/_dask/typing.py
+++ /dev/null
@@ -1,16 +0,0 @@
-from __future__ import annotations  # pragma: no cover
-
-from typing import TYPE_CHECKING  # pragma: no cover
-from typing import Union  # pragma: no cover
-
-if TYPE_CHECKING:
-    import sys
-
-    if sys.version_info >= (3, 10):
-        from typing import TypeAlias
-    else:
-        from typing_extensions import TypeAlias
-
-    from narwhals._dask.expr import DaskExpr
-
-    IntoDaskExpr: TypeAlias = Union[DaskExpr, str]
diff --git a/narwhals/_dask/utils.py b/narwhals/_dask/utils.py
index d1438386..c7140e96 100644
--- a/narwhals/_dask/utils.py
+++ b/narwhals/_dask/utils.py
@@ -6,7 +6,6 @@ from typing import Any
 from narwhals._pandas_like.utils import select_columns_by_name
 from narwhals.dependencies import get_pandas
 from narwhals.dependencies import get_pyarrow
-from narwhals.exceptions import InvalidIntoExprError
 from narwhals.utils import Implementation
 from narwhals.utils import import_dtypes_module
 from narwhals.utils import isinstance_or_issubclass
@@ -45,16 +44,11 @@ def maybe_evaluate(df: DaskLazyFrame, obj: Any) -> Any:
 
 
 def parse_exprs_and_named_exprs(
-    df: DaskLazyFrame, *exprs: Any, **named_exprs: Any
+    df: DaskLazyFrame, *exprs: DaskExpr, **named_exprs: DaskExpr
 ) -> dict[str, dx.Series]:
     results = {}
     for expr in exprs:
-        if hasattr(expr, "__narwhals_expr__"):
-            _results = expr._call(df)
-        elif isinstance(expr, str):
-            _results = [df._native_frame[expr]]
-        else:
-            raise InvalidIntoExprError.from_invalid_type(type(expr))
+        _results = expr._call(df)
         return_scalar = getattr(expr, "_returns_scalar", False)
         for _result in _results:
             results[_result.name] = _result[0] if return_scalar else _result
diff --git a/narwhals/_duckdb/dataframe.py b/narwhals/_duckdb/dataframe.py
index f5a7fe7d..00325b28 100644
--- a/narwhals/_duckdb/dataframe.py
+++ b/narwhals/_duckdb/dataframe.py
@@ -3,7 +3,6 @@ from __future__ import annotations
 from itertools import chain
 from typing import TYPE_CHECKING
 from typing import Any
-from typing import Iterable
 from typing import Literal
 from typing import Sequence
 
@@ -15,7 +14,6 @@ from narwhals.dependencies import get_duckdb
 from narwhals.exceptions import ColumnNotFoundError
 from narwhals.utils import Implementation
 from narwhals.utils import Version
-from narwhals.utils import flatten
 from narwhals.utils import generate_temporary_column_name
 from narwhals.utils import parse_columns_to_drop
 from narwhals.utils import parse_version
@@ -33,11 +31,12 @@ if TYPE_CHECKING:
     from narwhals._duckdb.group_by import DuckDBGroupBy
     from narwhals._duckdb.namespace import DuckDBNamespace
     from narwhals._duckdb.series import DuckDBInterchangeSeries
-    from narwhals._duckdb.typing import IntoDuckDBExpr
     from narwhals.dtypes import DType
 
+from narwhals.typing import CompliantLazyFrame
 
-class DuckDBLazyFrame:
+
+class DuckDBLazyFrame(CompliantLazyFrame):
     _implementation = Implementation.DUCKDB
 
     def __init__(
@@ -97,10 +96,13 @@ class DuckDBLazyFrame:
     def head(self: Self, n: int) -> Self:
         return self._from_native_frame(self._native_frame.limit(n))
 
+    def simple_select(self, *column_names: str) -> Self:
+        return self._from_native_frame(self._native_frame.select(*column_names))
+
     def select(
         self: Self,
-        *exprs: IntoDuckDBExpr,
-        **named_exprs: IntoDuckDBExpr,
+        *exprs: DuckDBExpr,
+        **named_exprs: DuckDBExpr,
     ) -> Self:
         new_columns_map = parse_exprs_and_named_exprs(self, *exprs, **named_exprs)
         if not new_columns_map:
@@ -134,8 +136,8 @@ class DuckDBLazyFrame:
 
     def with_columns(
         self: Self,
-        *exprs: IntoDuckDBExpr,
-        **named_exprs: IntoDuckDBExpr,
+        *exprs: DuckDBExpr,
+        **named_exprs: DuckDBExpr,
     ) -> Self:
         new_columns_map = parse_exprs_and_named_exprs(self, *exprs, **named_exprs)
         result = []
@@ -297,14 +299,12 @@ class DuckDBLazyFrame:
 
     def sort(
         self: Self,
-        by: str | Iterable[str],
-        *more_by: str,
+        *by: str,
         descending: bool | Sequence[bool],
         nulls_last: bool,
     ) -> Self:
-        flat_by = flatten([*flatten([by]), *more_by])
         if isinstance(descending, bool):
-            descending = [descending] * len(flat_by)
+            descending = [descending] * len(by)
         descending_str = ["desc" if x else "" for x in descending]
 
         result = self._native_frame.order(
@@ -313,7 +313,7 @@ class DuckDBLazyFrame:
                     f'"{col}" {desc} nulls last'
                     if nulls_last
                     else f'"{col}" {desc} nulls first'
-                    for col, desc in zip(flat_by, descending_str)
+                    for col, desc in zip(by, descending_str)
                 )
             )
         )
diff --git a/narwhals/_duckdb/expr.py b/narwhals/_duckdb/expr.py
index 3571aadb..031e48e6 100644
--- a/narwhals/_duckdb/expr.py
+++ b/narwhals/_duckdb/expr.py
@@ -460,27 +460,32 @@ class DuckDBExpr(CompliantExpr["duckdb.Expression"]):
         )
 
     def std(self: Self, ddof: int) -> Self:
-        if ddof == 1:
-            func = "stddev_samp"
-        elif ddof == 0:
-            func = "stddev_pop"
-        else:
-            msg = f"std with ddof {ddof} is not supported in DuckDB"
-            raise NotImplementedError(msg)
+        def _std(_input: duckdb.Expression, ddof: int) -> duckdb.Expression:
+            n_samples = FunctionExpression("count", _input)
+
+            return (
+                FunctionExpression("stddev_pop", _input)
+                * FunctionExpression("sqrt", n_samples)
+                / (FunctionExpression("sqrt", (n_samples - ddof)))
+            )
+
         return self._from_call(
-            lambda _input: FunctionExpression(func, _input), "std", returns_scalar=True
+            _std,
+            "std",
+            ddof=ddof,
+            returns_scalar=True,
         )
 
     def var(self: Self, ddof: int) -> Self:
-        if ddof == 1:
-            func = "var_samp"
-        elif ddof == 0:
-            func = "var_pop"
-        else:
-            msg = f"var with ddof {ddof} is not supported in DuckDB"
-            raise NotImplementedError(msg)
+        def _var(_input: duckdb.Expression, ddof: int) -> duckdb.Expression:
+            n_samples = FunctionExpression("count", _input)
+            return FunctionExpression("var_pop", _input) * n_samples / (n_samples - ddof)
+
         return self._from_call(
-            lambda _input: FunctionExpression(func, _input), "var", returns_scalar=True
+            _var,
+            "var",
+            ddof=ddof,
+            returns_scalar=True,
         )
 
     def max(self: Self) -> Self:
diff --git a/narwhals/_duckdb/group_by.py b/narwhals/_duckdb/group_by.py
index ff3dae9b..11f629be 100644
--- a/narwhals/_duckdb/group_by.py
+++ b/narwhals/_duckdb/group_by.py
@@ -3,14 +3,13 @@ from __future__ import annotations
 from copy import copy
 from typing import TYPE_CHECKING
 
-from narwhals._expression_parsing import parse_into_exprs
 from narwhals.exceptions import AnonymousExprError
 
 if TYPE_CHECKING:
     from typing_extensions import Self
 
     from narwhals._duckdb.dataframe import DuckDBLazyFrame
-    from narwhals._duckdb.typing import IntoDuckDBExpr
+    from narwhals._duckdb.expr import DuckDBExpr
 
 
 class DuckDBGroupBy:
@@ -28,14 +27,8 @@ class DuckDBGroupBy:
 
     def agg(
         self: Self,
-        *aggs: IntoDuckDBExpr,
-        **named_aggs: IntoDuckDBExpr,
+        *exprs: DuckDBExpr,
     ) -> DuckDBLazyFrame:
-        exprs = parse_into_exprs(
-            *aggs,
-            namespace=self._compliant_frame.__narwhals_namespace__(),
-            **named_aggs,
-        )
         output_names: list[str] = copy(self._keys)
         for expr in exprs:
             if expr._output_names is None:  # pragma: no cover
diff --git a/narwhals/_duckdb/namespace.py b/narwhals/_duckdb/namespace.py
index bbe6d961..2ceffc4e 100644
--- a/narwhals/_duckdb/namespace.py
+++ b/narwhals/_duckdb/namespace.py
@@ -5,7 +5,6 @@ import operator
 from functools import reduce
 from typing import TYPE_CHECKING
 from typing import Any
-from typing import Iterable
 from typing import Literal
 from typing import Sequence
 from typing import cast
@@ -19,7 +18,6 @@ from duckdb import FunctionExpression
 from narwhals._duckdb.expr import DuckDBExpr
 from narwhals._duckdb.utils import narwhals_to_native_dtype
 from narwhals._expression_parsing import combine_root_names
-from narwhals._expression_parsing import parse_into_exprs
 from narwhals._expression_parsing import reduce_output_names
 from narwhals.typing import CompliantNamespace
 
@@ -28,7 +26,6 @@ if TYPE_CHECKING:
     from typing_extensions import Self
 
     from narwhals._duckdb.dataframe import DuckDBLazyFrame
-    from narwhals._duckdb.typing import IntoDuckDBExpr
     from narwhals.dtypes import DType
     from narwhals.utils import Version
 
@@ -84,19 +81,13 @@ class DuckDBNamespace(CompliantNamespace["duckdb.Expression"]):
 
     def concat_str(
         self: Self,
-        exprs: Iterable[IntoDuckDBExpr],
-        *more_exprs: IntoDuckDBExpr,
+        *exprs: DuckDBExpr,
         separator: str,
         ignore_nulls: bool,
     ) -> DuckDBExpr:
-        parsed_exprs = [
-            *parse_into_exprs(*exprs, namespace=self),
-            *parse_into_exprs(*more_exprs, namespace=self),
-        ]
-
         def func(df: DuckDBLazyFrame) -> list[duckdb.Expression]:
-            cols = [s for _expr in parsed_exprs for s in _expr(df)]
-            null_mask = [s.isnull() for _expr in parsed_exprs for s in _expr(df)]
+            cols = [s for _expr in exprs for s in _expr(df)]
+            null_mask = [s.isnull() for _expr in exprs for s in _expr(df)]
             first_column_name = get_column_name(df, cols[0])
 
             if not ignore_nulls:
@@ -141,106 +132,95 @@ class DuckDBNamespace(CompliantNamespace["duckdb.Expression"]):
 
         return DuckDBExpr(
             call=func,
-            depth=max(x._depth for x in parsed_exprs) + 1,
+            depth=max(x._depth for x in exprs) + 1,
             function_name="concat_str",
-            root_names=combine_root_names(parsed_exprs),
-            output_names=reduce_output_names(parsed_exprs),
+            root_names=combine_root_names(exprs),
+            output_names=reduce_output_names(exprs),
             returns_scalar=False,
             backend_version=self._backend_version,
             version=self._version,
             kwargs={
                 "exprs": exprs,
-                "more_exprs": more_exprs,
                 "separator": separator,
                 "ignore_nulls": ignore_nulls,
             },
         )
 
-    def all_horizontal(self: Self, *exprs: IntoDuckDBExpr) -> DuckDBExpr:
-        parsed_exprs = parse_into_exprs(*exprs, namespace=self)
-
+    def all_horizontal(self: Self, *exprs: DuckDBExpr) -> DuckDBExpr:
         def func(df: DuckDBLazyFrame) -> list[duckdb.Expression]:
-            cols = [c for _expr in parsed_exprs for c in _expr(df)]
+            cols = [c for _expr in exprs for c in _expr(df)]
             col_name = get_column_name(df, cols[0])
             return [reduce(operator.and_, cols).alias(col_name)]
 
         return DuckDBExpr(
             call=func,
-            depth=max(x._depth for x in parsed_exprs) + 1,
+            depth=max(x._depth for x in exprs) + 1,
             function_name="all_horizontal",
-            root_names=combine_root_names(parsed_exprs),
-            output_names=reduce_output_names(parsed_exprs),
+            root_names=combine_root_names(exprs),
+            output_names=reduce_output_names(exprs),
             returns_scalar=False,
             backend_version=self._backend_version,
             version=self._version,
             kwargs={"exprs": exprs},
         )
 
-    def any_horizontal(self: Self, *exprs: IntoDuckDBExpr) -> DuckDBExpr:
-        parsed_exprs = parse_into_exprs(*exprs, namespace=self)
-
+    def any_horizontal(self: Self, *exprs: DuckDBExpr) -> DuckDBExpr:
         def func(df: DuckDBLazyFrame) -> list[duckdb.Expression]:
-            cols = [c for _expr in parsed_exprs for c in _expr(df)]
+            cols = [c for _expr in exprs for c in _expr(df)]
             col_name = get_column_name(df, cols[0])
             return [reduce(operator.or_, cols).alias(col_name)]
 
         return DuckDBExpr(
             call=func,
-            depth=max(x._depth for x in parsed_exprs) + 1,
+            depth=max(x._depth for x in exprs) + 1,
             function_name="or_horizontal",
-            root_names=combine_root_names(parsed_exprs),
-            output_names=reduce_output_names(parsed_exprs),
+            root_names=combine_root_names(exprs),
+            output_names=reduce_output_names(exprs),
             returns_scalar=False,
             backend_version=self._backend_version,
             version=self._version,
             kwargs={"exprs": exprs},
         )
 
-    def max_horizontal(self: Self, *exprs: IntoDuckDBExpr) -> DuckDBExpr:
-        parsed_exprs = parse_into_exprs(*exprs, namespace=self)
-
+    def max_horizontal(self: Self, *exprs: DuckDBExpr) -> DuckDBExpr:
         def func(df: DuckDBLazyFrame) -> list[duckdb.Expression]:
-            cols = [c for _expr in parsed_exprs for c in _expr(df)]
+            cols = [c for _expr in exprs for c in _expr(df)]
             col_name = get_column_name(df, cols[0])
             return [FunctionExpression("greatest", *cols).alias(col_name)]
 
         return DuckDBExpr(
             call=func,
-            depth=max(x._depth for x in parsed_exprs) + 1,
+            depth=max(x._depth for x in exprs) + 1,
             function_name="max_horizontal",
-            root_names=combine_root_names(parsed_exprs),
-            output_names=reduce_output_names(parsed_exprs),
+            root_names=combine_root_names(exprs),
+            output_names=reduce_output_names(exprs),
             returns_scalar=False,
             backend_version=self._backend_version,
             version=self._version,
             kwargs={"exprs": exprs},
         )
 
-    def min_horizontal(self: Self, *exprs: IntoDuckDBExpr) -> DuckDBExpr:
-        parsed_exprs = parse_into_exprs(*exprs, namespace=self)
-
+    def min_horizontal(self: Self, *exprs: DuckDBExpr) -> DuckDBExpr:
         def func(df: DuckDBLazyFrame) -> list[duckdb.Expression]:
-            cols = [c for _expr in parsed_exprs for c in _expr(df)]
+            cols = [c for _expr in exprs for c in _expr(df)]
             col_name = get_column_name(df, cols[0])
             return [FunctionExpression("least", *cols).alias(col_name)]
 
         return DuckDBExpr(
             call=func,
-            depth=max(x._depth for x in parsed_exprs) + 1,
+            depth=max(x._depth for x in exprs) + 1,
             function_name="min_horizontal",
-            root_names=combine_root_names(parsed_exprs),
-            output_names=reduce_output_names(parsed_exprs),
+            root_names=combine_root_names(exprs),
+            output_names=reduce_output_names(exprs),
             returns_scalar=False,
             backend_version=self._backend_version,
             version=self._version,
             kwargs={"exprs": exprs},
         )
 
-    def sum_horizontal(self: Self, *exprs: IntoDuckDBExpr) -> DuckDBExpr:
-        parsed_exprs = parse_into_exprs(*exprs, namespace=self)
-
+    def sum_horizontal(self: Self, *exprs: DuckDBExpr) -> DuckDBExpr:
         def func(df: DuckDBLazyFrame) -> list[duckdb.Expression]:
-            cols = [c for _expr in parsed_exprs for c in _expr(df)]
+            cols = [c for _expr in exprs for c in _expr(df)]
             col_name = get_column_name(df, cols[0])
             return [
                 reduce(
@@ -251,21 +231,19 @@ class DuckDBNamespace(CompliantNamespace["duckdb.Expression"]):
 
         return DuckDBExpr(
             call=func,
-            depth=max(x._depth for x in parsed_exprs) + 1,
+            depth=max(x._depth for x in exprs) + 1,
             function_name="sum_horizontal",
-            root_names=combine_root_names(parsed_exprs),
-            output_names=reduce_output_names(parsed_exprs),
+            root_names=combine_root_names(exprs),
+            output_names=reduce_output_names(exprs),
             returns_scalar=False,
             backend_version=self._backend_version,
             version=self._version,
             kwargs={"exprs": exprs},
         )
 
-    def mean_horizontal(self: Self, *exprs: IntoDuckDBExpr) -> DuckDBExpr:
-        parsed_exprs = parse_into_exprs(*exprs, namespace=self)
-
+    def mean_horizontal(self: Self, *exprs: DuckDBExpr) -> DuckDBExpr:
         def func(df: DuckDBLazyFrame) -> list[duckdb.Expression]:
-            cols = [c for _expr in parsed_exprs for c in _expr(df)]
+            cols = [c for _expr in exprs for c in _expr(df)]
             col_name = get_column_name(df, cols[0])
             return [
                 (
@@ -279,10 +257,10 @@ class DuckDBNamespace(CompliantNamespace["duckdb.Expression"]):
 
         return DuckDBExpr(
             call=func,
-            depth=max(x._depth for x in parsed_exprs) + 1,
+            depth=max(x._depth for x in exprs) + 1,
             function_name="mean_horizontal",
-            root_names=combine_root_names(parsed_exprs),
-            output_names=reduce_output_names(parsed_exprs),
+            root_names=combine_root_names(exprs),
+            output_names=reduce_output_names(exprs),
             returns_scalar=False,
             backend_version=self._backend_version,
             version=self._version,
@@ -291,7 +269,7 @@ class DuckDBNamespace(CompliantNamespace["duckdb.Expression"]):
 
     def when(
         self: Self,
-        *predicates: IntoDuckDBExpr,
+        *predicates: DuckDBExpr,
     ) -> DuckDBWhen:
         plx = self.__class__(backend_version=self._backend_version, version=self._version)
         condition = plx.all_horizontal(*predicates)
@@ -367,32 +345,27 @@ class DuckDBWhen:
         self._version = version
 
     def __call__(self: Self, df: DuckDBLazyFrame) -> Sequence[duckdb.Expression]:
-        from narwhals._expression_parsing import parse_into_expr
-
-        plx = df.__narwhals_namespace__()
-        condition = parse_into_expr(self._condition, namespace=plx)(df)[0]
+        condition = self._condition(df)[0]
         condition = cast("duckdb.Expression", condition)
 
-        try:
-            value = parse_into_expr(self._then_value, namespace=plx)(df)[0]
-        except TypeError:
-            # `self._otherwise_value` is a scalar and can't be converted to an expression
+        if isinstance(self._then_value, DuckDBExpr):
+            value = self._then_value(df)[0]
+        else:
+            # `self._otherwise_value` is a scalar
             value = ConstantExpression(self._then_value).alias("literal")
         value = cast("duckdb.Expression", value)
         value_name = get_column_name(df, value)
 
         if self._otherwise_value is None:
             return [CaseExpression(condition=condition, value=value).alias(value_name)]
-        try:
-            otherwise_expr = parse_into_expr(self._otherwise_value, namespace=plx)
-        except TypeError:
-            # `self._otherwise_value` is a scalar and can't be converted to an expression
+        if not isinstance(self._otherwise_value, DuckDBExpr):
+            # `self._otherwise_value` is a scalar
             return [
                 CaseExpression(condition=condition, value=value)
                 .otherwise(ConstantExpression(self._otherwise_value))
                 .alias(value_name)
             ]
-        otherwise = otherwise_expr(df)[0]
+        otherwise = self._otherwise_value(df)[0]
         return [
             CaseExpression(condition=condition, value=value)
             .otherwise(otherwise)
diff --git a/narwhals/_duckdb/typing.py b/narwhals/_duckdb/typing.py
deleted file mode 100644
index 65d1ba3a..00000000
--- a/narwhals/_duckdb/typing.py
+++ /dev/null
@@ -1,16 +0,0 @@
-from __future__ import annotations  # pragma: no cover
-
-from typing import TYPE_CHECKING  # pragma: no cover
-from typing import Union  # pragma: no cover
-
-if TYPE_CHECKING:
-    import sys
-
-    if sys.version_info >= (3, 10):
-        from typing import TypeAlias
-    else:
-        from typing_extensions import TypeAlias
-
-    from narwhals._duckdb.expr import DuckDBExpr
-
-    IntoDuckDBExpr: TypeAlias = Union[DuckDBExpr, str]
diff --git a/narwhals/_duckdb/utils.py b/narwhals/_duckdb/utils.py
index 5e7f6221..d4512326 100644
--- a/narwhals/_duckdb/utils.py
+++ b/narwhals/_duckdb/utils.py
@@ -6,10 +6,7 @@ from typing import TYPE_CHECKING
 from typing import Any
 from typing import Sequence
 
-from duckdb import ColumnExpression
-
 from narwhals.dtypes import DType
-from narwhals.exceptions import InvalidIntoExprError
 from narwhals.utils import import_dtypes_module
 from narwhals.utils import isinstance_or_issubclass
 
@@ -18,7 +15,6 @@ if TYPE_CHECKING:
 
     from narwhals._duckdb.dataframe import DuckDBLazyFrame
     from narwhals._duckdb.expr import DuckDBExpr
-    from narwhals._duckdb.typing import IntoDuckDBExpr
     from narwhals.utils import Version
 
 
@@ -52,15 +48,13 @@ def maybe_evaluate(df: DuckDBLazyFrame, obj: Any) -> Any:
 
 def parse_exprs_and_named_exprs(
     df: DuckDBLazyFrame,
-    *exprs: IntoDuckDBExpr,
-    **named_exprs: IntoDuckDBExpr,
+    *exprs: DuckDBExpr,
+    **named_exprs: DuckDBExpr,
 ) -> dict[str, duckdb.Expression]:
     result_columns: dict[str, list[duckdb.Expression]] = {}
     for expr in exprs:
         column_list = _columns_from_expr(df, expr)
-        if isinstance(expr, str):  # pragma: no cover
-            output_names = [expr]
-        elif expr._output_names is None:
+        if expr._output_names is None:
             output_names = [
                 get_column_name(df, col, returns_scalar=expr._returns_scalar)
                 for col in column_list
@@ -78,20 +72,15 @@ def parse_exprs_and_named_exprs(
 
 
 def _columns_from_expr(
-    df: DuckDBLazyFrame, expr: IntoDuckDBExpr
+    df: DuckDBLazyFrame, expr: DuckDBExpr
 ) -> Sequence[duckdb.Expression]:
-    if isinstance(expr, str):  # pragma: no cover
-        return [ColumnExpression(expr)]
-    elif hasattr(expr, "__narwhals_expr__"):
-        col_output_list = expr._call(df)
-        if expr._output_names is not None and (
-            len(col_output_list) != len(expr._output_names)
-        ):  # pragma: no cover
-            msg = "Safety assertion failed, please report a bug to https://github.com/narwhals-dev/narwhals/issues"
-            raise AssertionError(msg)
-        return col_output_list
-    else:
-        raise InvalidIntoExprError.from_invalid_type(type(expr))
+    col_output_list = expr._call(df)
+    if expr._output_names is not None and (
+        len(col_output_list) != len(expr._output_names)
+    ):  # pragma: no cover
+        msg = "Safety assertion failed, please report a bug to https://github.com/narwhals-dev/narwhals/issues"
+        raise AssertionError(msg)
+    return col_output_list
 
 
 @lru_cache(maxsize=16)
@@ -163,7 +152,7 @@ def narwhals_to_native_dtype(dtype: DType | type[DType], version: Version) -> st
     if isinstance_or_issubclass(dtype, dtypes.Int64):
         return "BIGINT"
     if isinstance_or_issubclass(dtype, dtypes.Int32):
-        return "INT"
+        return "INTEGER"
     if isinstance_or_issubclass(dtype, dtypes.Int16):
         return "SMALLINT"
     if isinstance_or_issubclass(dtype, dtypes.Int8):
@@ -171,7 +160,7 @@ def narwhals_to_native_dtype(dtype: DType | type[DType], version: Version) -> st
     if isinstance_or_issubclass(dtype, dtypes.UInt64):
         return "UBIGINT"
     if isinstance_or_issubclass(dtype, dtypes.UInt32):
-        return "UINT"
+        return "UINTEGER"
     if isinstance_or_issubclass(dtype, dtypes.UInt16):  # pragma: no cover
         return "USMALLINT"
     if isinstance_or_issubclass(dtype, dtypes.UInt8):  # pragma: no cover
diff --git a/narwhals/_expression_parsing.py b/narwhals/_expression_parsing.py
index a11db68e..96f1b00c 100644
--- a/narwhals/_expression_parsing.py
+++ b/narwhals/_expression_parsing.py
@@ -18,8 +18,6 @@ from narwhals.exceptions import LengthChangingExprError
 from narwhals.utils import Implementation
 
 if TYPE_CHECKING:
-    from typing_extensions import TypeAlias
-
     from narwhals._arrow.expr import ArrowExpr
     from narwhals._pandas_like.expr import PandasLikeExpr
     from narwhals.typing import CompliantDataFrame
@@ -28,13 +26,9 @@ if TYPE_CHECKING:
     from narwhals.typing import CompliantNamespace
     from narwhals.typing import CompliantSeries
     from narwhals.typing import CompliantSeriesT_co
+    from narwhals.typing import IntoCompliantExpr
     from narwhals.typing import IntoExpr
 
-    IntoCompliantExpr: TypeAlias = (
-        CompliantExpr[CompliantSeriesT_co] | str | CompliantSeriesT_co
-    )
-    CompliantExprT = TypeVar("CompliantExprT", bound=CompliantExpr[Any])
-
     ArrowOrPandasLikeExpr = TypeVar(
         "ArrowOrPandasLikeExpr", bound=Union[ArrowExpr, PandasLikeExpr]
     )
@@ -119,11 +113,9 @@ def parse_into_expr(
         return into_expr  # type: ignore[return-value]
     if hasattr(into_expr, "__narwhals_series__"):
         return namespace._create_expr_from_series(into_expr)  # type: ignore[no-any-return, attr-defined]
-    if isinstance(into_expr, str):
-        return namespace.col(into_expr)
     if is_numpy_array(into_expr):
-        series = namespace._create_compliant_series(into_expr)
-        return namespace._create_expr_from_series(series)
+        series = namespace._create_compliant_series(into_expr)  # type: ignore[attr-defined]
+        return namespace._create_expr_from_series(series)  # type: ignore[no-any-return, attr-defined]
     raise InvalidIntoExprError.from_invalid_type(type(into_expr))
 
 
@@ -326,13 +318,18 @@ def reduce_output_names(parsed_exprs: Sequence[CompliantExpr[Any]]) -> list[str]
 
 
 def extract_compliant(
-    plx: CompliantNamespace[CompliantSeriesT_co], other: Any
+    plx: CompliantNamespace[CompliantSeriesT_co],
+    other: Any,
+    *,
+    parse_column_name_as_expr: bool,
 ) -> CompliantExpr[CompliantSeriesT_co] | CompliantSeriesT_co | Any:
     from narwhals.expr import Expr
     from narwhals.series import Series
 
     if isinstance(other, Expr):
         return other._to_compliant_expr(plx)
+    if parse_column_name_as_expr and isinstance(other, str):
+        return plx.col(other)
     if isinstance(other, Series):
         return other._compliant_series
     return other
diff --git a/narwhals/_ibis/dataframe.py b/narwhals/_ibis/dataframe.py
index 6adb0f08..d1fc4728 100644
--- a/narwhals/_ibis/dataframe.py
+++ b/narwhals/_ibis/dataframe.py
@@ -4,6 +4,8 @@ from functools import lru_cache
 from typing import TYPE_CHECKING
 from typing import Any
 
+import ibis.selectors as s
+
 from narwhals.dependencies import get_ibis
 from narwhals.utils import Implementation
 from narwhals.utils import Version
@@ -105,23 +107,21 @@ class IbisLazyFrame:
     def to_arrow(self: Self) -> pa.Table:
         return self._native_frame.to_pyarrow()
 
+    def simple_select(self, *column_names: str) -> Self:
+        return self._from_native_frame(self._native_frame.select(s.cols(*column_names)))
+
     def select(
         self: Self,
         *exprs: Any,
         **named_exprs: Any,
     ) -> Self:
-        if named_exprs or not all(isinstance(x, str) for x in exprs):  # pragma: no cover
-            msg = (
-                "`select`-ing not by name is not supported for Ibis backend.\n\n"
-                "If you would like to see this kind of object better supported in "
-                "Narwhals, please open a feature request "
-                "at https://github.com/narwhals-dev/narwhals/issues."
-            )
-            raise NotImplementedError(msg)
-
-        import ibis.selectors as s
-
-        return self._from_native_frame(self._native_frame.select(s.cols(*exprs)))
+        msg = (
+            "`select`-ing not by name is not supported for Ibis backend.\n\n"
+            "If you would like to see this kind of object better supported in "
+            "Narwhals, please open a feature request "
+            "at https://github.com/narwhals-dev/narwhals/issues."
+        )
+        raise NotImplementedError(msg)
 
     def __getattr__(self, attr: str) -> Any:
         if attr == "schema":
diff --git a/narwhals/_interchange/dataframe.py b/narwhals/_interchange/dataframe.py
index 06cab9b0..947cfa26 100644
--- a/narwhals/_interchange/dataframe.py
+++ b/narwhals/_interchange/dataframe.py
@@ -153,25 +153,25 @@ class InterchangeFrame:
         )
         raise NotImplementedError(msg)
 
-    def select(
-        self: Self,
-        *exprs: str,
-        **named_exprs: str,
-    ) -> Self:
-        if named_exprs or not all(isinstance(x, str) for x in exprs):  # pragma: no cover
-            msg = (
-                "`select`-ing not by name is not supported for interchange-only level.\n\n"
-                "If you would like to see this kind of object better supported in "
-                "Narwhals, please open a feature request "
-                "at https://github.com/narwhals-dev/narwhals/issues."
-            )
-            raise NotImplementedError(msg)
-
-        frame = self._interchange_frame.select_columns_by_name(exprs)
+    def simple_select(self: Self, *column_names: str) -> Self:
+        frame = self._interchange_frame.select_columns_by_name(list(column_names))
         if not hasattr(frame, "_df"):  # pragma: no cover
             msg = (
                 "Expected interchange object to implement `_df` property to allow for recovering original object.\n"
                 "See https://github.com/data-apis/dataframe-api/issues/360."
             )
-            raise NotImplementedError(frame)
+            raise NotImplementedError(msg)
         return self.__class__(frame._df, version=self._version)
+
+    def select(
+        self: Self,
+        *exprs: str,
+        **named_exprs: str,
+    ) -> Self:  # pragma: no cover
+        msg = (
+            "`select`-ing not by name is not supported for interchange-only level.\n\n"
+            "If you would like to see this kind of object better supported in "
+            "Narwhals, please open a feature request "
+            "at https://github.com/narwhals-dev/narwhals/issues."
+        )
+        raise NotImplementedError(msg)
diff --git a/narwhals/_pandas_like/dataframe.py b/narwhals/_pandas_like/dataframe.py
index dc1d69bc..0aec90bd 100644
--- a/narwhals/_pandas_like/dataframe.py
+++ b/narwhals/_pandas_like/dataframe.py
@@ -3,7 +3,6 @@ from __future__ import annotations
 from itertools import chain
 from typing import TYPE_CHECKING
 from typing import Any
-from typing import Iterable
 from typing import Iterator
 from typing import Literal
 from typing import Sequence
@@ -22,7 +21,6 @@ from narwhals._pandas_like.utils import select_columns_by_name
 from narwhals.dependencies import is_numpy_array
 from narwhals.utils import Implementation
 from narwhals.utils import check_column_exists
-from narwhals.utils import flatten
 from narwhals.utils import generate_temporary_column_name
 from narwhals.utils import import_dtypes_module
 from narwhals.utils import is_sequence_but_not_str
@@ -48,8 +46,11 @@ if TYPE_CHECKING:
     from narwhals.typing import SizeUnit
     from narwhals.utils import Version
 
+from narwhals.typing import CompliantDataFrame
+from narwhals.typing import CompliantLazyFrame
 
-class PandasLikeDataFrame:
+
+class PandasLikeDataFrame(CompliantDataFrame, CompliantLazyFrame):
     # --- not in the spec ---
     def __init__(
         self: Self,
@@ -347,22 +348,21 @@ class PandasLikeDataFrame:
         return self.schema
 
     # --- reshape ---
+    def simple_select(self: Self, *column_names: str) -> Self:
+        return self._from_native_frame(
+            select_columns_by_name(
+                self._native_frame,
+                list(column_names),
+                self._backend_version,
+                self._implementation,
+            )
+        )
+
     def select(
         self: Self,
         *exprs: IntoPandasLikeExpr,
         **named_exprs: IntoPandasLikeExpr,
     ) -> Self:
-        if exprs and all(isinstance(x, str) for x in exprs) and not named_exprs:
-            # This is a simple slice => fastpath!
-            column_names = list(exprs)
-            return self._from_native_frame(
-                select_columns_by_name(
-                    self._native_frame,
-                    column_names,  # type: ignore[arg-type]
-                    self._backend_version,
-                    self._implementation,
-                )
-            )
         new_series = evaluate_into_exprs(self, *exprs, **named_exprs)
         if not new_series:
             # return empty dataframe, like Polars does
@@ -482,12 +482,10 @@ class PandasLikeDataFrame:
     # --- transform ---
     def sort(
         self: Self,
-        by: str | Iterable[str],
-        *more_by: str,
+        *by: str,
         descending: bool | Sequence[bool],
         nulls_last: bool,
     ) -> Self:
-        flat_keys = flatten([*flatten([by]), *more_by])
         df = self._native_frame
         if isinstance(descending, bool):
             ascending: bool | list[bool] = not descending
@@ -495,7 +493,7 @@ class PandasLikeDataFrame:
             ascending = [not d for d in descending]
         na_position = "last" if nulls_last else "first"
         return self._from_native_frame(
-            df.sort_values(flat_keys, ascending=ascending, na_position=na_position)
+            df.sort_values(list(by), ascending=ascending, na_position=na_position)
         )
 
     # --- convert ---
diff --git a/narwhals/_pandas_like/expr.py b/narwhals/_pandas_like/expr.py
index 4138b1a8..17852408 100644
--- a/narwhals/_pandas_like/expr.py
+++ b/narwhals/_pandas_like/expr.py
@@ -450,7 +450,7 @@ class PandasLikeExpr(CompliantExpr[PandasLikeSeries]):
                     msg = "over"
                     raise AnonymousExprError.from_expr_name(msg)
                 tmp = df.group_by(*keys, drop_null_keys=False).agg(self)
-                tmp = df.select(*keys).join(
+                tmp = df.simple_select(*keys).join(
                     tmp, how="left", left_on=keys, right_on=keys, suffix="_right"
                 )
                 return [tmp[name] for name in self._output_names]
diff --git a/narwhals/_pandas_like/group_by.py b/narwhals/_pandas_like/group_by.py
index ad20af42..b817f3dd 100644
--- a/narwhals/_pandas_like/group_by.py
+++ b/narwhals/_pandas_like/group_by.py
@@ -10,7 +10,6 @@ from typing import Iterator
 from typing import Sequence
 
 from narwhals._expression_parsing import is_simple_aggregation
-from narwhals._expression_parsing import parse_into_exprs
 from narwhals._pandas_like.utils import horizontal_concat
 from narwhals._pandas_like.utils import native_series_from_iterable
 from narwhals._pandas_like.utils import select_columns_by_name
@@ -24,8 +23,8 @@ if TYPE_CHECKING:
     from typing_extensions import Self
 
     from narwhals._pandas_like.dataframe import PandasLikeDataFrame
+    from narwhals._pandas_like.expr import PandasLikeExpr
     from narwhals._pandas_like.series import PandasLikeSeries
-    from narwhals._pandas_like.typing import IntoPandasLikeExpr
     from narwhals.typing import CompliantExpr
 
 POLARS_TO_PANDAS_AGGREGATIONS = {
@@ -83,14 +82,8 @@ class PandasLikeGroupBy:
 
     def agg(
         self: Self,
-        *aggs: IntoPandasLikeExpr,
-        **named_aggs: IntoPandasLikeExpr,
+        *exprs: PandasLikeExpr,
     ) -> PandasLikeDataFrame:
-        exprs = parse_into_exprs(
-            *aggs,
-            namespace=self._df.__narwhals_namespace__(),
-            **named_aggs,
-        )
         implementation: Implementation = self._df._implementation
         output_names: list[str] = copy(self._keys)
         for expr in exprs:
diff --git a/narwhals/_pandas_like/namespace.py b/narwhals/_pandas_like/namespace.py
index 6f098674..8ace4351 100644
--- a/narwhals/_pandas_like/namespace.py
+++ b/narwhals/_pandas_like/namespace.py
@@ -377,15 +377,11 @@ class PandasLikeNamespace(CompliantNamespace[PandasLikeSeries]):
 
     def concat_str(
         self: Self,
-        exprs: Iterable[IntoPandasLikeExpr],
-        *more_exprs: IntoPandasLikeExpr,
+        *exprs: IntoPandasLikeExpr,
         separator: str,
         ignore_nulls: bool,
     ) -> PandasLikeExpr:
-        parsed_exprs = [
-            *parse_into_exprs(*exprs, namespace=self),
-            *parse_into_exprs(*more_exprs, namespace=self),
-        ]
+        parsed_exprs = parse_into_exprs(*exprs, namespace=self)
         dtypes = import_dtypes_module(self._version)
 
         def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
@@ -429,7 +425,6 @@ class PandasLikeNamespace(CompliantNamespace[PandasLikeSeries]):
             output_names=reduce_output_names(parsed_exprs),
             kwargs={
                 "exprs": exprs,
-                "more_exprs": more_exprs,
                 "separator": separator,
                 "ignore_nulls": ignore_nulls,
             },
diff --git a/narwhals/_pandas_like/typing.py b/narwhals/_pandas_like/typing.py
index 3fe6dcd5..9def2506 100644
--- a/narwhals/_pandas_like/typing.py
+++ b/narwhals/_pandas_like/typing.py
@@ -14,4 +14,4 @@ if TYPE_CHECKING:
     from narwhals._pandas_like.expr import PandasLikeExpr
     from narwhals._pandas_like.series import PandasLikeSeries
 
-    IntoPandasLikeExpr: TypeAlias = Union[PandasLikeExpr, str, PandasLikeSeries]
+    IntoPandasLikeExpr: TypeAlias = Union[PandasLikeExpr, PandasLikeSeries]
diff --git a/narwhals/_pandas_like/utils.py b/narwhals/_pandas_like/utils.py
index 1946f97e..883af4bb 100644
--- a/narwhals/_pandas_like/utils.py
+++ b/narwhals/_pandas_like/utils.py
@@ -868,7 +868,7 @@ def pivot_table(
     if df._implementation is Implementation.CUDF:
         if any(
             x == dtypes.Categorical
-            for x in df.select(*[*values, *index, *columns]).schema.values()
+            for x in df.simple_select(*[*values, *index, *columns]).schema.values()
         ):
             msg = "`pivot` with Categoricals is not implemented for cuDF backend"
             raise NotImplementedError(msg)
diff --git a/narwhals/_polars/dataframe.py b/narwhals/_polars/dataframe.py
index d07e485b..727912f1 100644
--- a/narwhals/_polars/dataframe.py
+++ b/narwhals/_polars/dataframe.py
@@ -13,7 +13,6 @@ 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.exceptions import ColumnNotFoundError
-from narwhals.exceptions import InvalidIntoExprError
 from narwhals.utils import Implementation
 from narwhals.utils import is_sequence_but_not_str
 from narwhals.utils import parse_columns_to_drop
@@ -109,17 +108,9 @@ class PolarsDataFrame:
                 return self._from_native_object(
                     getattr(self._native_frame, attr)(*args, **kwargs)
                 )
-            except pl.exceptions.ColumnNotFoundError as e:
+            except pl.exceptions.ColumnNotFoundError as e:  # pragma: no cover
                 msg = f"{e!s}\n\nHint: Did you mean one of these columns: {self.columns}?"
                 raise ColumnNotFoundError(msg) from e
-            except TypeError as e:
-                e_str = str(e)
-                if (
-                    "cannot create expression literal" in e_str
-                    or "invalid literal" in e_str
-                ):
-                    raise InvalidIntoExprError(e_str) from e
-                raise
 
         return func
 
@@ -213,6 +204,9 @@ class PolarsDataFrame:
                 )
             return self._from_native_object(result)
 
+    def simple_select(self, *column_names: str) -> Self:
+        return self._from_native_frame(self._native_frame.select(*column_names))
+
     def get_column(self: Self, name: str) -> PolarsSeries:
         from narwhals._polars.series import PolarsSeries
 
@@ -386,14 +380,6 @@ class PolarsLazyFrame:
                 )
             except pl.exceptions.ColumnNotFoundError as e:  # pragma: no cover
                 raise ColumnNotFoundError(str(e)) from e
-            except TypeError as e:
-                e_str = str(e)
-                if (
-                    "cannot create expression literal" in e_str
-                    or "invalid literal" in e_str
-                ):
-                    raise InvalidIntoExprError(e_str) from e
-                raise
 
         return func
 
@@ -475,3 +461,6 @@ class PolarsLazyFrame:
                 on=on, index=index, variable_name=variable_name, value_name=value_name
             )
         )
+
+    def simple_select(self, *column_names: str) -> Self:
+        return self._from_native_frame(self._native_frame.select(*column_names))
diff --git a/narwhals/_polars/group_by.py b/narwhals/_polars/group_by.py
index 5bb1b58f..26b19cae 100644
--- a/narwhals/_polars/group_by.py
+++ b/narwhals/_polars/group_by.py
@@ -24,11 +24,9 @@ class PolarsGroupBy:
         else:
             self._grouped = df._native_frame.group_by(keys)
 
-    def agg(self: Self, *aggs: PolarsExpr, **named_aggs: PolarsExpr) -> PolarsDataFrame:
-        aggs, named_aggs = extract_args_kwargs(aggs, named_aggs)  # type: ignore[assignment]
-        return self._compliant_frame._from_native_frame(
-            self._grouped.agg(*aggs, **named_aggs),
-        )
+    def agg(self: Self, *aggs: PolarsExpr) -> PolarsDataFrame:
+        aggs, _ = extract_args_kwargs(aggs, {})  # type: ignore[assignment]
+        return self._compliant_frame._from_native_frame(self._grouped.agg(*aggs))
 
     def __iter__(self: Self) -> Iterator[tuple[tuple[str, ...], PolarsDataFrame]]:
         for key, df in self._grouped:
@@ -46,8 +44,6 @@ class PolarsLazyGroupBy:
         else:
             self._grouped = df._native_frame.group_by(keys)
 
-    def agg(self: Self, *aggs: PolarsExpr, **named_aggs: PolarsExpr) -> PolarsLazyFrame:
-        aggs, named_aggs = extract_args_kwargs(aggs, named_aggs)  # type: ignore[assignment]
-        return self._compliant_frame._from_native_frame(
-            self._grouped.agg(*aggs, **named_aggs),
-        )
+    def agg(self: Self, *aggs: PolarsExpr) -> PolarsLazyFrame:
+        aggs, _ = extract_args_kwargs(aggs, {})  # type: ignore[assignment]
+        return self._compliant_frame._from_native_frame(self._grouped.agg(*aggs))
diff --git a/narwhals/_polars/namespace.py b/narwhals/_polars/namespace.py
index dd5d2a0a..32e53b37 100644
--- a/narwhals/_polars/namespace.py
+++ b/narwhals/_polars/namespace.py
@@ -151,8 +151,7 @@ class PolarsNamespace:
 
     def concat_str(
         self: Self,
-        exprs: Iterable[IntoPolarsExpr],
-        *more_exprs: IntoPolarsExpr,
+        *exprs: IntoPolarsExpr,
         separator: str,
         ignore_nulls: bool,
     ) -> PolarsExpr:
@@ -162,10 +161,7 @@ class PolarsNamespace:
 
         pl_exprs: list[pl.Expr] = [
             expr._native_expr  # type: ignore[attr-defined]
-            for expr in (
-                *parse_into_exprs(*exprs, namespace=self),
-                *parse_into_exprs(*more_exprs, namespace=self),
-            )
+            for expr in parse_into_exprs(*exprs, namespace=self)
         ]
 
         if self._backend_version < (0, 20, 6):
diff --git a/narwhals/_polars/typing.py b/narwhals/_polars/typing.py
index 3aed2899..3323a114 100644
--- a/narwhals/_polars/typing.py
+++ b/narwhals/_polars/typing.py
@@ -14,4 +14,4 @@ if TYPE_CHECKING:
     from narwhals._polars.expr import PolarsExpr
     from narwhals._polars.series import PolarsSeries
 
-    IntoPolarsExpr: TypeAlias = Union[PolarsExpr, str, PolarsSeries]
+    IntoPolarsExpr: TypeAlias = Union[PolarsExpr, PolarsSeries]
diff --git a/narwhals/_spark_like/dataframe.py b/narwhals/_spark_like/dataframe.py
index bee9d634..3c4ac2fe 100644
--- a/narwhals/_spark_like/dataframe.py
+++ b/narwhals/_spark_like/dataframe.py
@@ -3,7 +3,6 @@ from __future__ import annotations
 from itertools import chain
 from typing import TYPE_CHECKING
 from typing import Any
-from typing import Iterable
 from typing import Literal
 from typing import Sequence
 
@@ -11,7 +10,6 @@ from narwhals._spark_like.utils import native_to_narwhals_dtype
 from narwhals._spark_like.utils import parse_exprs_and_named_exprs
 from narwhals.utils import Implementation
 from narwhals.utils import check_column_exists
-from narwhals.utils import flatten
 from narwhals.utils import parse_columns_to_drop
 from narwhals.utils import parse_version
 from narwhals.utils import validate_backend_version
@@ -26,12 +24,12 @@ if TYPE_CHECKING:
     from narwhals._spark_like.expr import SparkLikeExpr
     from narwhals._spark_like.group_by import SparkLikeLazyGroupBy
     from narwhals._spark_like.namespace import SparkLikeNamespace
-    from narwhals._spark_like.typing import IntoSparkLikeExpr
     from narwhals.dtypes import DType
     from narwhals.utils import Version
+from narwhals.typing import CompliantLazyFrame
 
 
-class SparkLikeLazyFrame:
+class SparkLikeLazyFrame(CompliantLazyFrame):
     def __init__(
         self: Self,
         native_dataframe: DataFrame,
@@ -88,15 +86,14 @@ class SparkLikeLazyFrame:
             version=self._version,
         )
 
+    def simple_select(self: Self, *column_names: str) -> Self:
+        return self._from_native_frame(self._native_frame.select(*column_names))
+
     def select(
         self: Self,
-        *exprs: IntoSparkLikeExpr,
-        **named_exprs: IntoSparkLikeExpr,
+        *exprs: SparkLikeExpr,
+        **named_exprs: SparkLikeExpr,
     ) -> Self:
-        if exprs and all(isinstance(x, str) for x in exprs) and not named_exprs:
-            # This is a simple select
-            return self._from_native_frame(self._native_frame.select(*exprs))
-
         new_columns = parse_exprs_and_named_exprs(self, *exprs, **named_exprs)
 
         if not new_columns:
@@ -135,8 +132,8 @@ class SparkLikeLazyFrame:
 
     def with_columns(
         self: Self,
-        *exprs: IntoSparkLikeExpr,
-        **named_exprs: IntoSparkLikeExpr,
+        *exprs: SparkLikeExpr,
+        **named_exprs: SparkLikeExpr,
     ) -> Self:
         new_columns_map = parse_exprs_and_named_exprs(self, *exprs, **named_exprs)
         return self._from_native_frame(self._native_frame.withColumns(new_columns_map))
@@ -163,16 +160,14 @@ class SparkLikeLazyFrame:
 
     def sort(
         self: Self,
-        by: str | Iterable[str],
-        *more_by: str,
+        *by: str,
         descending: bool | Sequence[bool],
         nulls_last: bool,
     ) -> Self:
         import pyspark.sql.functions as F  # noqa: N812
 
-        flat_by = flatten([*flatten([by]), *more_by])
         if isinstance(descending, bool):
-            descending = [descending] * len(flat_by)
+            descending = [descending] * len(by)
 
         if nulls_last:
             sort_funcs = (
@@ -183,7 +178,7 @@ class SparkLikeLazyFrame:
                 F.desc_nulls_first if d else F.asc_nulls_first for d in descending
             )
 
-        sort_cols = [sort_f(col) for col, sort_f in zip(flat_by, sort_funcs)]
+        sort_cols = [sort_f(col) for col, sort_f in zip(by, sort_funcs)]
         return self._from_native_frame(self._native_frame.sort(*sort_cols))
 
     def drop_nulls(self: Self, subset: list[str] | None) -> Self:
diff --git a/narwhals/_spark_like/group_by.py b/narwhals/_spark_like/group_by.py
index 4d1c9882..319a95dc 100644
--- a/narwhals/_spark_like/group_by.py
+++ b/narwhals/_spark_like/group_by.py
@@ -10,7 +10,6 @@ from typing import Sequence
 from pyspark.sql import functions as F  # noqa: N812
 
 from narwhals._expression_parsing import is_simple_aggregation
-from narwhals._expression_parsing import parse_into_exprs
 from narwhals._spark_like.utils import _std
 from narwhals._spark_like.utils import _var
 from narwhals.exceptions import AnonymousExprError
@@ -23,7 +22,7 @@ if TYPE_CHECKING:
     from typing_extensions import Self
 
     from narwhals._spark_like.dataframe import SparkLikeLazyFrame
-    from narwhals._spark_like.typing import IntoSparkLikeExpr
+    from narwhals._spark_like.typing import SparkLikeExpr
     from narwhals.typing import CompliantExpr
 
 
@@ -45,14 +44,8 @@ class SparkLikeLazyGroupBy:
 
     def agg(
         self: Self,
-        *aggs: IntoSparkLikeExpr,
-        **named_aggs: IntoSparkLikeExpr,
+        *exprs: SparkLikeExpr,
     ) -> SparkLikeLazyFrame:
-        exprs = parse_into_exprs(
-            *aggs,
-            namespace=self._df.__narwhals_namespace__(),
-            **named_aggs,
-        )
         output_names: list[str] = copy(self._keys)
         for expr in exprs:
             if expr._output_names is None:  # pragma: no cover
diff --git a/narwhals/_spark_like/namespace.py b/narwhals/_spark_like/namespace.py
index 029f77d0..59a15dde 100644
--- a/narwhals/_spark_like/namespace.py
+++ b/narwhals/_spark_like/namespace.py
@@ -10,8 +10,6 @@ from typing import Literal
 from pyspark.sql import functions as F  # noqa: N812
 
 from narwhals._expression_parsing import combine_root_names
-from narwhals._expression_parsing import parse_into_expr
-from narwhals._expression_parsing import parse_into_exprs
 from narwhals._expression_parsing import reduce_output_names
 from narwhals._spark_like.dataframe import SparkLikeLazyFrame
 from narwhals._spark_like.expr import SparkLikeExpr
@@ -23,7 +21,6 @@ if TYPE_CHECKING:
     from pyspark.sql import DataFrame
     from typing_extensions import Self
 
-    from narwhals._spark_like.typing import IntoSparkLikeExpr
     from narwhals.dtypes import DType
     from narwhals.utils import Version
 
@@ -103,53 +100,47 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
             kwargs={},
         )
 
-    def all_horizontal(self: Self, *exprs: IntoSparkLikeExpr) -> SparkLikeExpr:
-        parsed_exprs = parse_into_exprs(*exprs, namespace=self)
-
+    def all_horizontal(self: Self, *exprs: SparkLikeExpr) -> SparkLikeExpr:
         def func(df: SparkLikeLazyFrame) -> list[Column]:
-            cols = [c for _expr in parsed_exprs for c in _expr(df)]
+            cols = [c for _expr in exprs for c in _expr(df)]
             col_name = get_column_name(df, cols[0])
             return [reduce(operator.and_, cols).alias(col_name)]
 
         return SparkLikeExpr(
             call=func,
-            depth=max(x._depth for x in parsed_exprs) + 1,
+            depth=max(x._depth for x in exprs) + 1,
             function_name="all_horizontal",
-            root_names=combine_root_names(parsed_exprs),
-            output_names=reduce_output_names(parsed_exprs),
+            root_names=combine_root_names(exprs),
+            output_names=reduce_output_names(exprs),
             returns_scalar=False,
             backend_version=self._backend_version,
             version=self._version,
             kwargs={"exprs": exprs},
         )
 
-    def any_horizontal(self: Self, *exprs: IntoSparkLikeExpr) -> SparkLikeExpr:
-        parsed_exprs = parse_into_exprs(*exprs, namespace=self)
-
+    def any_horizontal(self: Self, *exprs: SparkLikeExpr) -> SparkLikeExpr:
         def func(df: SparkLikeLazyFrame) -> list[Column]:
-            cols = [c for _expr in parsed_exprs for c in _expr(df)]
+            cols = [c for _expr in exprs for c in _expr(df)]
             col_name = get_column_name(df, cols[0])
             return [reduce(operator.or_, cols).alias(col_name)]
 
         return SparkLikeExpr(
             call=func,
-            depth=max(x._depth for x in parsed_exprs) + 1,
+            depth=max(x._depth for x in exprs) + 1,
             function_name="any_horizontal",
-            root_names=combine_root_names(parsed_exprs),
-            output_names=reduce_output_names(parsed_exprs),
+            root_names=combine_root_names(exprs),
+            output_names=reduce_output_names(exprs),
             returns_scalar=False,
             backend_version=self._backend_version,
             version=self._version,
             kwargs={"exprs": exprs},
         )
 
-    def sum_horizontal(self: Self, *exprs: IntoSparkLikeExpr) -> SparkLikeExpr:
-        parsed_exprs = parse_into_exprs(*exprs, namespace=self)
-
+    def sum_horizontal(self: Self, *exprs: SparkLikeExpr) -> SparkLikeExpr:
         def func(df: SparkLikeLazyFrame) -> list[Column]:
             import pyspark.sql.functions as F  # noqa: N812
 
-            cols = [c for _expr in parsed_exprs for c in _expr(df)]
+            cols = [c for _expr in exprs for c in _expr(df)]
             col_name = get_column_name(df, cols[0])
             return [
                 reduce(
@@ -160,23 +151,21 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
 
         return SparkLikeExpr(
             call=func,
-            depth=max(x._depth for x in parsed_exprs) + 1,
+            depth=max(x._depth for x in exprs) + 1,
             function_name="sum_horizontal",
-            root_names=combine_root_names(parsed_exprs),
-            output_names=reduce_output_names(parsed_exprs),
+            root_names=combine_root_names(exprs),
+            output_names=reduce_output_names(exprs),
             returns_scalar=False,
             backend_version=self._backend_version,
             version=self._version,
             kwargs={"exprs": exprs},
         )
 
-    def mean_horizontal(self: Self, *exprs: IntoSparkLikeExpr) -> SparkLikeExpr:
+    def mean_horizontal(self: Self, *exprs: SparkLikeExpr) -> SparkLikeExpr:
         from pyspark.sql.types import IntegerType
 
-        parsed_exprs = parse_into_exprs(*exprs, namespace=self)
-
         def func(df: SparkLikeLazyFrame) -> list[Column]:
-            cols = [c for _expr in parsed_exprs for c in _expr(df)]
+            cols = [c for _expr in exprs for c in _expr(df)]
             col_name = get_column_name(df, cols[0])
             return [
                 (
@@ -190,50 +179,46 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
 
         return SparkLikeExpr(
             call=func,
-            depth=max(x._depth for x in parsed_exprs) + 1,
+            depth=max(x._depth for x in exprs) + 1,
             function_name="mean_horizontal",
-            root_names=combine_root_names(parsed_exprs),
-            output_names=reduce_output_names(parsed_exprs),
+            root_names=combine_root_names(exprs),
+            output_names=reduce_output_names(exprs),
             returns_scalar=False,
             backend_version=self._backend_version,
             version=self._version,
             kwargs={"exprs": exprs},
         )
 
-    def max_horizontal(self: Self, *exprs: IntoSparkLikeExpr) -> SparkLikeExpr:
-        parsed_exprs = parse_into_exprs(*exprs, namespace=self)
-
+    def max_horizontal(self: Self, *exprs: SparkLikeExpr) -> SparkLikeExpr:
         def func(df: SparkLikeLazyFrame) -> list[Column]:
-            cols = [c for _expr in parsed_exprs for c in _expr(df)]
+            cols = [c for _expr in exprs for c in _expr(df)]
             col_name = get_column_name(df, cols[0])
             return [F.greatest(*cols).alias(col_name)]
 
         return SparkLikeExpr(
             call=func,
-            depth=max(x._depth for x in parsed_exprs) + 1,
+            depth=max(x._depth for x in exprs) + 1,
             function_name="max_horizontal",
-            root_names=combine_root_names(parsed_exprs),
-            output_names=reduce_output_names(parsed_exprs),
+            root_names=combine_root_names(exprs),
+            output_names=reduce_output_names(exprs),
             returns_scalar=False,
             backend_version=self._backend_version,
             version=self._version,
             kwargs={"exprs": exprs},
         )
 
-    def min_horizontal(self: Self, *exprs: IntoSparkLikeExpr) -> SparkLikeExpr:
-        parsed_exprs = parse_into_exprs(*exprs, namespace=self)
-
+    def min_horizontal(self: Self, *exprs: SparkLikeExpr) -> SparkLikeExpr:
         def func(df: SparkLikeLazyFrame) -> list[Column]:
-            cols = [c for _expr in parsed_exprs for c in _expr(df)]
+            cols = [c for _expr in exprs for c in _expr(df)]
             col_name = get_column_name(df, cols[0])
             return [F.least(*cols).alias(col_name)]
 
         return SparkLikeExpr(
             call=func,
-            depth=max(x._depth for x in parsed_exprs) + 1,
+            depth=max(x._depth for x in exprs) + 1,
             function_name="min_horizontal",
-            root_names=combine_root_names(parsed_exprs),
-            output_names=reduce_output_names(parsed_exprs),
+            root_names=combine_root_names(exprs),
+            output_names=reduce_output_names(exprs),
             returns_scalar=False,
             backend_version=self._backend_version,
             version=self._version,
@@ -284,22 +269,16 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
 
     def concat_str(
         self: Self,
-        exprs: Iterable[IntoSparkLikeExpr],
-        *more_exprs: IntoSparkLikeExpr,
+        *exprs: SparkLikeExpr,
         separator: str,
         ignore_nulls: bool,
     ) -> SparkLikeExpr:
         from pyspark.sql.types import StringType
 
-        parsed_exprs = [
-            *parse_into_exprs(*exprs, namespace=self),
-            *parse_into_exprs(*more_exprs, namespace=self),
-        ]
-
         def func(df: SparkLikeLazyFrame) -> list[Column]:
-            cols = [s for _expr in parsed_exprs for s in _expr(df)]
+            cols = [s for _expr in exprs for s in _expr(df)]
             cols_casted = [s.cast(StringType()) for s in cols]
-            null_mask = [F.isnull(s) for _expr in parsed_exprs for s in _expr(df)]
+            null_mask = [F.isnull(s) for _expr in exprs for s in _expr(df)]
             first_column_name = get_column_name(df, cols[0])
 
             if not ignore_nulls:
@@ -331,22 +310,21 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
 
         return SparkLikeExpr(
             call=func,
-            depth=max(x._depth for x in parsed_exprs) + 1,
+            depth=max(x._depth for x in exprs) + 1,
             function_name="concat_str",
-            root_names=combine_root_names(parsed_exprs),
-            output_names=reduce_output_names(parsed_exprs),
+            root_names=combine_root_names(exprs),
+            output_names=reduce_output_names(exprs),
             returns_scalar=False,
             backend_version=self._backend_version,
             version=self._version,
             kwargs={
                 "exprs": exprs,
-                "more_exprs": more_exprs,
                 "separator": separator,
                 "ignore_nulls": ignore_nulls,
             },
         )
 
-    def when(self: Self, *predicates: IntoSparkLikeExpr) -> SparkLikeWhen:
+    def when(self: Self, *predicates: SparkLikeExpr) -> SparkLikeWhen:
         plx = self.__class__(backend_version=self._backend_version, version=self._version)
         condition = plx.all_horizontal(*predicates)
         return SparkLikeWhen(
@@ -373,21 +351,20 @@ class SparkLikeWhen:
         self._version = version
 
     def __call__(self: Self, df: SparkLikeLazyFrame) -> list[Column]:
-        plx = df.__narwhals_namespace__()
-        condition = parse_into_expr(self._condition, namespace=plx)(df)[0]
+        condition = self._condition(df)[0]
 
-        try:
-            value_ = parse_into_expr(self._then_value, namespace=plx)(df)[0]
+        if isinstance(self._then_value, SparkLikeExpr):
+            value_ = self._then_value(df)[0]
             col_name = get_column_name(df, value_)
-        except TypeError:
-            # `self._then_value` is a scalar and can't be converted to an expression
+        else:
+            # `self._then_value` is a scalar
             value_ = F.lit(self._then_value)
             col_name = "literal"
 
-        try:
-            other_ = parse_into_expr(self._otherwise_value, namespace=plx)(df)[0]
-        except TypeError:
-            # `self._otherwise_value` is a scalar and can't be converted to an expression
+        if isinstance(self._otherwise_value, SparkLikeExpr):
+            other_ = self._otherwise_value(df)[0]
+        else:
+            # `self._otherwise_value` is a scalar
             other_ = F.lit(self._otherwise_value)
 
         return [
diff --git a/narwhals/_spark_like/typing.py b/narwhals/_spark_like/typing.py
deleted file mode 100644
index fb343044..00000000
--- a/narwhals/_spark_like/typing.py
+++ /dev/null
@@ -1,16 +0,0 @@
-from __future__ import annotations  # pragma: no cover
-
-from typing import TYPE_CHECKING  # pragma: no cover
-from typing import Union  # pragma: no cover
-
-if TYPE_CHECKING:
-    import sys
-
-    if sys.version_info >= (3, 10):
-        from typing import TypeAlias
-    else:
-        from typing_extensions import TypeAlias
-
-    from narwhals._spark_like.expr import SparkLikeExpr
-
-    IntoSparkLikeExpr: TypeAlias = Union[SparkLikeExpr, str]
diff --git a/narwhals/_spark_like/utils.py b/narwhals/_spark_like/utils.py
index dbea4772..37a2426d 100644
--- a/narwhals/_spark_like/utils.py
+++ b/narwhals/_spark_like/utils.py
@@ -6,7 +6,6 @@ from typing import Any
 
 from pyspark.sql import functions as F  # noqa: N812
 
-from narwhals.exceptions import InvalidIntoExprError
 from narwhals.exceptions import UnsupportedDTypeError
 from narwhals.utils import import_dtypes_module
 from narwhals.utils import isinstance_or_issubclass
@@ -16,7 +15,7 @@ if TYPE_CHECKING:
     from pyspark.sql import types as pyspark_types
 
     from narwhals._spark_like.dataframe import SparkLikeLazyFrame
-    from narwhals._spark_like.typing import IntoSparkLikeExpr
+    from narwhals._spark_like.expr import SparkLikeExpr
     from narwhals.dtypes import DType
     from narwhals.utils import Version
 
@@ -114,30 +113,23 @@ def get_column_name(df: SparkLikeLazyFrame, column: Column) -> str:
     return str(df._native_frame.select(column).columns[0])
 
 
-def _columns_from_expr(df: SparkLikeLazyFrame, expr: IntoSparkLikeExpr) -> list[Column]:
-    if isinstance(expr, str):  # pragma: no cover
-        return [F.col(expr)]
-    elif hasattr(expr, "__narwhals_expr__"):
-        col_output_list = expr._call(df)
-        if expr._output_names is not None and (
-            len(col_output_list) != len(expr._output_names)
-        ):  # pragma: no cover
-            msg = "Safety assertion failed, please report a bug to https://github.com/narwhals-dev/narwhals/issues"
-            raise AssertionError(msg)
-        return col_output_list
-    else:
-        raise InvalidIntoExprError.from_invalid_type(type(expr))
+def _columns_from_expr(df: SparkLikeLazyFrame, expr: SparkLikeExpr) -> list[Column]:
+    col_output_list = expr._call(df)
+    if expr._output_names is not None and (
+        len(col_output_list) != len(expr._output_names)
+    ):  # pragma: no cover
+        msg = "Safety assertion failed, please report a bug to https://github.com/narwhals-dev/narwhals/issues"
+        raise AssertionError(msg)
+    return col_output_list
 
 
 def parse_exprs_and_named_exprs(
-    df: SparkLikeLazyFrame, *exprs: IntoSparkLikeExpr, **named_exprs: IntoSparkLikeExpr
+    df: SparkLikeLazyFrame, *exprs: SparkLikeExpr, **named_exprs: SparkLikeExpr
 ) -> dict[str, Column]:
     result_columns: dict[str, list[Column]] = {}
     for expr in exprs:
         column_list = _columns_from_expr(df, expr)
-        if isinstance(expr, str):  # pragma: no cover
-            output_names = [expr]
-        elif expr._output_names is None:
+        if expr._output_names is None:
             output_names = [get_column_name(df, col) for col in column_list]
         else:
             output_names = expr._output_names
diff --git a/narwhals/dataframe.py b/narwhals/dataframe.py
index d9657a61..786fb5c7 100644
--- a/narwhals/dataframe.py
+++ b/narwhals/dataframe.py
@@ -16,6 +16,8 @@ from warnings import warn
 
 from narwhals.dependencies import get_polars
 from narwhals.dependencies import is_numpy_array
+from narwhals.exceptions import ColumnNotFoundError
+from narwhals.exceptions import InvalidIntoExprError
 from narwhals.exceptions import LengthChangingExprError
 from narwhals.exceptions import OrderDependentExprError
 from narwhals.exceptions import ShapeError
@@ -43,6 +45,7 @@ if TYPE_CHECKING:
     from narwhals.group_by import GroupBy
     from narwhals.group_by import LazyGroupBy
     from narwhals.series import Series
+    from narwhals.typing import IntoCompliantExpr
     from narwhals.typing import IntoDataFrame
     from narwhals.typing import IntoExpr
     from narwhals.typing import IntoFrame
@@ -69,15 +72,25 @@ class BaseFrame(Generic[FrameT]):
     def _from_compliant_dataframe(self: Self, df: Any) -> Self:
         # construct, preserving properties
         return self.__class__(  # type: ignore[call-arg]
-            df,
-            level=self._level,
+            df, level=self._level
         )
 
-    def _flatten_and_extract(self: Self, *args: Any, **kwargs: Any) -> Any:
-        """Process `args` and `kwargs`, extracting underlying objects as we go."""
-        args = [self._extract_compliant(v) for v in flatten(args)]  # type: ignore[assignment]
-        kwargs = {k: self._extract_compliant(v) for k, v in kwargs.items()}
-        return args, kwargs
+    def _flatten_and_extract(
+        self, *exprs: IntoExpr | Iterable[IntoExpr], **named_exprs: IntoExpr
+    ) -> tuple[tuple[IntoCompliantExpr[Any]], dict[str, IntoCompliantExpr[Any]]]:
+        """Process `args` and `kwargs`, extracting underlying objects as we go, interpreting strings as column names."""
+        plx = self.__narwhals_namespace__()
+        compliant_exprs = tuple(
+            plx.col(expr) if isinstance(expr, str) else self._extract_compliant(expr)
+            for expr in flatten(exprs)
+        )
+        compliant_named_exprs = {
+            key: plx.col(value)
+            if isinstance(value, str)
+            else self._extract_compliant(value)
+            for key, value in named_exprs.items()
+        }
+        return compliant_exprs, compliant_named_exprs
 
     @abstractmethod
     def _extract_compliant(self: Self, arg: Any) -> Any:
@@ -118,9 +131,11 @@ class BaseFrame(Generic[FrameT]):
     def with_columns(
         self: Self, *exprs: IntoExpr | Iterable[IntoExpr], **named_exprs: IntoExpr
     ) -> Self:
-        exprs, named_exprs = self._flatten_and_extract(*exprs, **named_exprs)
+        compliant_exprs, compliant_named_exprs = self._flatten_and_extract(
+            *exprs, **named_exprs
+        )
         return self._from_compliant_dataframe(
-            self._compliant_frame.with_columns(*exprs, **named_exprs),
+            self._compliant_frame.with_columns(*compliant_exprs, **compliant_named_exprs),
         )
 
     def select(
@@ -128,9 +143,26 @@ class BaseFrame(Generic[FrameT]):
         *exprs: IntoExpr | Iterable[IntoExpr],
         **named_exprs: IntoExpr,
     ) -> Self:
-        exprs, named_exprs = self._flatten_and_extract(*exprs, **named_exprs)
+        flat_exprs = tuple(flatten(exprs))
+        if flat_exprs and all(isinstance(x, str) for x in flat_exprs) and not named_exprs:
+            # fast path!
+            try:
+                return self._from_compliant_dataframe(
+                    self._compliant_frame.simple_select(*flat_exprs),
+                )
+            except Exception as e:
+                # Column not found is the only thing that can realistically be raised here.
+                available_columns = self.columns
+                missing_columns = [x for x in flat_exprs if x not in available_columns]
+                raise ColumnNotFoundError.from_missing_and_available_column_names(
+                    missing_columns, available_columns
+                ) from e
+
+        compliant_exprs, compliant_named_exprs = self._flatten_and_extract(
+            *flat_exprs, **named_exprs
+        )
         return self._from_compliant_dataframe(
-            self._compliant_frame.select(*exprs, **named_exprs),
+            self._compliant_frame.select(*compliant_exprs, **compliant_named_exprs),
         )
 
     def rename(self: Self, mapping: dict[str, str]) -> Self:
@@ -176,10 +208,9 @@ class BaseFrame(Generic[FrameT]):
         descending: bool | Sequence[bool] = False,
         nulls_last: bool = False,
     ) -> Self:
+        by = flatten([*flatten([by]), *more_by])
         return self._from_compliant_dataframe(
-            self._compliant_frame.sort(
-                by, *more_by, descending=descending, nulls_last=nulls_last
-            )
+            self._compliant_frame.sort(*by, descending=descending, nulls_last=nulls_last)
         )
 
     def join(
@@ -376,7 +407,7 @@ class DataFrame(BaseFrame[DataFrameT]):
             return arg._compliant_series
         if isinstance(arg, Expr):
             return arg._to_compliant_expr(self.__narwhals_namespace__())
-        if get_polars() is not None and "polars" in str(type(arg)):
+        if get_polars() is not None and "polars" in str(type(arg)):  # pragma: no cover
             msg = (
                 f"Expected Narwhals object, got: {type(arg)}.\n\n"
                 "Perhaps you:\n"
@@ -384,7 +415,9 @@ class DataFrame(BaseFrame[DataFrameT]):
                 "- Used `pl.col` instead of `nw.col`?"
             )
             raise TypeError(msg)
-        return arg
+        if is_numpy_array(arg):
+            return arg
+        raise InvalidIntoExprError.from_invalid_type(type(arg))
 
     @property
     def _series(self: Self) -> type[Series[Any]]:
@@ -3697,9 +3730,7 @@ class LazyFrame(BaseFrame[FrameT]):
                 "- Used `pl.col` instead of `nw.col`?"
             )
             raise TypeError(msg)
-        # TODO(unassigned): should this line even be reachable? Should we
-        # be raising here?
-        return arg  # pragma: no cover
+        raise InvalidIntoExprError.from_invalid_type(type(arg))  # pragma: no cover
 
     @property
     def _dataframe(self: Self) -> type[DataFrame[Any]]:
diff --git a/narwhals/expr.py b/narwhals/expr.py
index 1ef13790..9e142c6d 100644
--- a/narwhals/expr.py
+++ b/narwhals/expr.py
@@ -264,7 +264,7 @@ class Expr:
     def __eq__(self: Self, other: object) -> Self:  # type: ignore[override]
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).__eq__(
-                extract_compliant(plx, other)
+                extract_compliant(plx, other, parse_column_name_as_expr=False)
             ),
             is_order_dependent=operation_is_order_dependent(self, other),
             changes_length=operation_changes_length(self, other),
@@ -274,7 +274,7 @@ class Expr:
     def __ne__(self: Self, other: object) -> Self:  # type: ignore[override]
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).__ne__(
-                extract_compliant(plx, other)
+                extract_compliant(plx, other, parse_column_name_as_expr=False)
             ),
             is_order_dependent=operation_is_order_dependent(self, other),
             changes_length=operation_changes_length(self, other),
@@ -284,7 +284,7 @@ class Expr:
     def __and__(self: Self, other: Any) -> Self:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).__and__(
-                extract_compliant(plx, other)
+                extract_compliant(plx, other, parse_column_name_as_expr=False)
             ),
             is_order_dependent=operation_is_order_dependent(self, other),
             changes_length=operation_changes_length(self, other),
@@ -293,9 +293,9 @@ class Expr:
 
     def __rand__(self: Self, other: Any) -> Self:
         def func(plx: CompliantNamespace[Any]) -> CompliantExpr[Any]:
-            return plx.lit(extract_compliant(plx, other), dtype=None).__and__(
-                extract_compliant(plx, self)
-            )
+            return plx.lit(
+                extract_compliant(plx, other, parse_column_name_as_expr=False), dtype=None
+            ).__and__(extract_compliant(plx, self, parse_column_name_as_expr=False))
 
         return self.__class__(
             func,
@@ -307,7 +307,7 @@ class Expr:
     def __or__(self: Self, other: Any) -> Self:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).__or__(
-                extract_compliant(plx, other)
+                extract_compliant(plx, other, parse_column_name_as_expr=False)
             ),
             is_order_dependent=operation_is_order_dependent(self, other),
             changes_length=operation_changes_length(self, other),
@@ -316,9 +316,9 @@ class Expr:
 
     def __ror__(self: Self, other: Any) -> Self:
         def func(plx: CompliantNamespace[Any]) -> CompliantExpr[Any]:
-            return plx.lit(extract_compliant(plx, other), dtype=None).__or__(
-                extract_compliant(plx, self)
-            )
+            return plx.lit(
+                extract_compliant(plx, other, parse_column_name_as_expr=False), dtype=None
+            ).__or__(extract_compliant(plx, self, parse_column_name_as_expr=False))
 
         return self.__class__(
             func,
@@ -330,7 +330,7 @@ class Expr:
     def __add__(self: Self, other: Any) -> Self:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).__add__(
-                extract_compliant(plx, other)
+                extract_compliant(plx, other, parse_column_name_as_expr=False)
             ),
             is_order_dependent=operation_is_order_dependent(self, other),
             changes_length=operation_changes_length(self, other),
@@ -339,9 +339,9 @@ class Expr:
 
     def __radd__(self: Self, other: Any) -> Self:
         def func(plx: CompliantNamespace[Any]) -> CompliantExpr[Any]:
-            return plx.lit(extract_compliant(plx, other), dtype=None).__add__(
-                extract_compliant(plx, self)
-            )
+            return plx.lit(
+                extract_compliant(plx, other, parse_column_name_as_expr=False), dtype=None
+            ).__add__(extract_compliant(plx, self, parse_column_name_as_expr=False))
 
         return self.__class__(
             func,
@@ -353,7 +353,7 @@ class Expr:
     def __sub__(self: Self, other: Any) -> Self:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).__sub__(
-                extract_compliant(plx, other)
+                extract_compliant(plx, other, parse_column_name_as_expr=False)
             ),
             is_order_dependent=operation_is_order_dependent(self, other),
             changes_length=operation_changes_length(self, other),
@@ -362,9 +362,9 @@ class Expr:
 
     def __rsub__(self: Self, other: Any) -> Self:
         def func(plx: CompliantNamespace[Any]) -> CompliantExpr[Any]:
-            return plx.lit(extract_compliant(plx, other), dtype=None).__sub__(
-                extract_compliant(plx, self)
-            )
+            return plx.lit(
+                extract_compliant(plx, other, parse_column_name_as_expr=False), dtype=None
+            ).__sub__(extract_compliant(plx, self, parse_column_name_as_expr=False))
 
         return self.__class__(
             func,
@@ -376,7 +376,7 @@ class Expr:
     def __truediv__(self: Self, other: Any) -> Self:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).__truediv__(
-                extract_compliant(plx, other)
+                extract_compliant(plx, other, parse_column_name_as_expr=False)
             ),
             is_order_dependent=operation_is_order_dependent(self, other),
             changes_length=operation_changes_length(self, other),
@@ -385,9 +385,9 @@ class Expr:
 
     def __rtruediv__(self: Self, other: Any) -> Self:
         def func(plx: CompliantNamespace[Any]) -> CompliantExpr[Any]:
-            return plx.lit(extract_compliant(plx, other), dtype=None).__truediv__(
-                extract_compliant(plx, self)
-            )
+            return plx.lit(
+                extract_compliant(plx, other, parse_column_name_as_expr=False), dtype=None
+            ).__truediv__(extract_compliant(plx, self, parse_column_name_as_expr=False))
 
         return self.__class__(
             func,
@@ -399,7 +399,7 @@ class Expr:
     def __mul__(self: Self, other: Any) -> Self:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).__mul__(
-                extract_compliant(plx, other)
+                extract_compliant(plx, other, parse_column_name_as_expr=False)
             ),
             is_order_dependent=operation_is_order_dependent(self, other),
             changes_length=operation_changes_length(self, other),
@@ -408,9 +408,9 @@ class Expr:
 
     def __rmul__(self: Self, other: Any) -> Self:
         def func(plx: CompliantNamespace[Any]) -> CompliantExpr[Any]:
-            return plx.lit(extract_compliant(plx, other), dtype=None).__mul__(
-                extract_compliant(plx, self)
-            )
+            return plx.lit(
+                extract_compliant(plx, other, parse_column_name_as_expr=False), dtype=None
+            ).__mul__(extract_compliant(plx, self, parse_column_name_as_expr=False))
 
         return self.__class__(
             func,
@@ -422,7 +422,7 @@ class Expr:
     def __le__(self: Self, other: Any) -> Self:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).__le__(
-                extract_compliant(plx, other)
+                extract_compliant(plx, other, parse_column_name_as_expr=False)
             ),
             is_order_dependent=operation_is_order_dependent(self, other),
             changes_length=operation_changes_length(self, other),
@@ -432,7 +432,7 @@ class Expr:
     def __lt__(self: Self, other: Any) -> Self:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).__lt__(
-                extract_compliant(plx, other)
+                extract_compliant(plx, other, parse_column_name_as_expr=False)
             ),
             is_order_dependent=operation_is_order_dependent(self, other),
             changes_length=operation_changes_length(self, other),
@@ -442,7 +442,7 @@ class Expr:
     def __gt__(self: Self, other: Any) -> Self:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).__gt__(
-                extract_compliant(plx, other)
+                extract_compliant(plx, other, parse_column_name_as_expr=False)
             ),
             is_order_dependent=operation_is_order_dependent(self, other),
             changes_length=operation_changes_length(self, other),
@@ -452,7 +452,7 @@ class Expr:
     def __ge__(self: Self, other: Any) -> Self:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).__ge__(
-                extract_compliant(plx, other)
+                extract_compliant(plx, other, parse_column_name_as_expr=False)
             ),
             is_order_dependent=operation_is_order_dependent(self, other),
             changes_length=operation_changes_length(self, other),
@@ -462,7 +462,7 @@ class Expr:
     def __pow__(self: Self, other: Any) -> Self:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).__pow__(
-                extract_compliant(plx, other)
+                extract_compliant(plx, other, parse_column_name_as_expr=False)
             ),
             is_order_dependent=operation_is_order_dependent(self, other),
             changes_length=operation_changes_length(self, other),
@@ -471,9 +471,9 @@ class Expr:
 
     def __rpow__(self: Self, other: Any) -> Self:
         def func(plx: CompliantNamespace[Any]) -> CompliantExpr[Any]:
-            return plx.lit(extract_compliant(plx, other), dtype=None).__pow__(
-                extract_compliant(plx, self)
-            )
+            return plx.lit(
+                extract_compliant(plx, other, parse_column_name_as_expr=False), dtype=None
+            ).__pow__(extract_compliant(plx, self, parse_column_name_as_expr=False))
 
         return self.__class__(
             func,
@@ -485,7 +485,7 @@ class Expr:
     def __floordiv__(self: Self, other: Any) -> Self:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).__floordiv__(
-                extract_compliant(plx, other)
+                extract_compliant(plx, other, parse_column_name_as_expr=False)
             ),
             is_order_dependent=operation_is_order_dependent(self, other),
             changes_length=operation_changes_length(self, other),
@@ -494,9 +494,9 @@ class Expr:
 
     def __rfloordiv__(self: Self, other: Any) -> Self:
         def func(plx: CompliantNamespace[Any]) -> CompliantExpr[Any]:
-            return plx.lit(extract_compliant(plx, other), dtype=None).__floordiv__(
-                extract_compliant(plx, self)
-            )
+            return plx.lit(
+                extract_compliant(plx, other, parse_column_name_as_expr=False), dtype=None
+            ).__floordiv__(extract_compliant(plx, self, parse_column_name_as_expr=False))
 
         return self.__class__(
             func,
@@ -508,7 +508,7 @@ class Expr:
     def __mod__(self: Self, other: Any) -> Self:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).__mod__(
-                extract_compliant(plx, other)
+                extract_compliant(plx, other, parse_column_name_as_expr=False)
             ),
             is_order_dependent=operation_is_order_dependent(self, other),
             changes_length=operation_changes_length(self, other),
@@ -517,9 +517,9 @@ class Expr:
 
     def __rmod__(self: Self, other: Any) -> Self:
         def func(plx: CompliantNamespace[Any]) -> CompliantExpr[Any]:
-            return plx.lit(extract_compliant(plx, other), dtype=None).__mod__(
-                extract_compliant(plx, self)
-            )
+            return plx.lit(
+                extract_compliant(plx, other, parse_column_name_as_expr=False), dtype=None
+            ).__mod__(extract_compliant(plx, self, parse_column_name_as_expr=False))
 
         return self.__class__(
             func,
@@ -1977,8 +1977,8 @@ class Expr:
         """Check if this expression is between the given lower and upper bounds.
 
         Arguments:
-            lower_bound: Lower bound value.
-            upper_bound: Upper bound value.
+            lower_bound: Lower bound value. String literals are interpreted as column names.
+            upper_bound: Upper bound value. String literals are interpreted as column names.
             closed: Define which sides of the interval are closed (inclusive).
 
         Returns:
@@ -2035,8 +2035,8 @@ class Expr:
         """
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).is_between(
-                extract_compliant(plx, lower_bound),
-                extract_compliant(plx, upper_bound),
+                extract_compliant(plx, lower_bound, parse_column_name_as_expr=True),
+                extract_compliant(plx, upper_bound, parse_column_name_as_expr=True),
                 closed,
             ),
             is_order_dependent=operation_is_order_dependent(
@@ -2107,7 +2107,7 @@ class Expr:
         if isinstance(other, Iterable) and not isinstance(other, (str, bytes)):
             return self.__class__(
                 lambda plx: self._to_compliant_expr(plx).is_in(
-                    extract_compliant(plx, other)
+                    extract_compliant(plx, other, parse_column_name_as_expr=False)
                 ),
                 self._is_order_dependent,
                 changes_length=self._changes_length,
@@ -2179,7 +2179,10 @@ class Expr:
         flat_predicates = flatten(predicates)
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).filter(
-                *[extract_compliant(plx, pred) for pred in flat_predicates],
+                *[
+                    extract_compliant(plx, pred, parse_column_name_as_expr=True)
+                    for pred in flat_predicates
+                ],
             ),
             is_order_dependent=operation_is_order_dependent(*flat_predicates),
             changes_length=True,
@@ -3321,8 +3324,8 @@ class Expr:
         r"""Clip values in the Series.
 
         Arguments:
-            lower_bound: Lower bound value.
-            upper_bound: Upper bound value.
+            lower_bound: Lower bound value. String literals are treated as column names.
+            upper_bound: Upper bound value. String literals are treated as column names.
 
         Returns:
             A new expression.
@@ -3453,8 +3456,8 @@ class Expr:
         """
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).clip(
-                extract_compliant(plx, lower_bound),
-                extract_compliant(plx, upper_bound),
+                extract_compliant(plx, lower_bound, parse_column_name_as_expr=True),
+                extract_compliant(plx, upper_bound, parse_column_name_as_expr=True),
             ),
             is_order_dependent=operation_is_order_dependent(
                 self, lower_bound, upper_bound
diff --git a/narwhals/functions.py b/narwhals/functions.py
index 090f4b87..7fd648dc 100644
--- a/narwhals/functions.py
+++ b/narwhals/functions.py
@@ -1946,7 +1946,12 @@ def sum_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:
         raise ValueError(msg)
     flat_exprs = flatten(exprs)
     return Expr(
-        lambda plx: plx.sum_horizontal(*[extract_compliant(plx, v) for v in flat_exprs]),
+        lambda plx: plx.sum_horizontal(
+            *(
+                extract_compliant(plx, v, parse_column_name_as_expr=True)
+                for v in flat_exprs
+            )
+        ),
         is_order_dependent=operation_is_order_dependent(*flat_exprs),
         changes_length=operation_changes_length(*flat_exprs),
         aggregates=operation_aggregates(*flat_exprs),
@@ -2018,7 +2023,12 @@ def min_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:
         raise ValueError(msg)
     flat_exprs = flatten(exprs)
     return Expr(
-        lambda plx: plx.min_horizontal(*[extract_compliant(plx, v) for v in flat_exprs]),
+        lambda plx: plx.min_horizontal(
+            *(
+                extract_compliant(plx, v, parse_column_name_as_expr=True)
+                for v in flat_exprs
+            )
+        ),
         is_order_dependent=operation_is_order_dependent(*flat_exprs),
         changes_length=operation_changes_length(*flat_exprs),
         aggregates=operation_aggregates(*flat_exprs),
@@ -2090,7 +2100,12 @@ def max_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:
         raise ValueError(msg)
     flat_exprs = flatten(exprs)
     return Expr(
-        lambda plx: plx.max_horizontal(*[extract_compliant(plx, v) for v in flat_exprs]),
+        lambda plx: plx.max_horizontal(
+            *(
+                extract_compliant(plx, v, parse_column_name_as_expr=True)
+                for v in flat_exprs
+            )
+        ),
         is_order_dependent=operation_is_order_dependent(*flat_exprs),
         changes_length=operation_changes_length(*flat_exprs),
         aggregates=operation_aggregates(*flat_exprs),
@@ -2111,12 +2126,15 @@ class When:
             raise ShapeError(msg)
 
     def _extract_predicates(self: Self, plx: Any) -> Any:
-        return [extract_compliant(plx, v) for v in self._predicates]
+        return [
+            extract_compliant(plx, v, parse_column_name_as_expr=True)
+            for v in self._predicates
+        ]
 
     def then(self: Self, value: IntoExpr | Any) -> Then:
         return Then(
             lambda plx: plx.when(*self._extract_predicates(plx)).then(
-                extract_compliant(plx, value)
+                extract_compliant(plx, value, parse_column_name_as_expr=True)
             ),
             is_order_dependent=operation_is_order_dependent(*self._predicates, value),
             changes_length=operation_changes_length(*self._predicates, value),
@@ -2128,7 +2146,7 @@ class Then(Expr):
     def otherwise(self: Self, value: IntoExpr | Any) -> Expr:
         return Expr(
             lambda plx: self._to_compliant_expr(plx).otherwise(
-                extract_compliant(plx, value)
+                extract_compliant(plx, value, parse_column_name_as_expr=True)
             ),
             is_order_dependent=operation_is_order_dependent(self, value),
             changes_length=operation_changes_length(self, value),
@@ -2284,7 +2302,12 @@ def all_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:
         raise ValueError(msg)
     flat_exprs = flatten(exprs)
     return Expr(
-        lambda plx: plx.all_horizontal(*[extract_compliant(plx, v) for v in flat_exprs]),
+        lambda plx: plx.all_horizontal(
+            *(
+                extract_compliant(plx, v, parse_column_name_as_expr=True)
+                for v in flat_exprs
+            )
+        ),
         is_order_dependent=operation_is_order_dependent(*flat_exprs),
         changes_length=operation_changes_length(*flat_exprs),
         aggregates=operation_aggregates(*flat_exprs),
@@ -2439,7 +2462,12 @@ def any_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:
         raise ValueError(msg)
     flat_exprs = flatten(exprs)
     return Expr(
-        lambda plx: plx.any_horizontal(*[extract_compliant(plx, v) for v in flat_exprs]),
+        lambda plx: plx.any_horizontal(
+            *(
+                extract_compliant(plx, v, parse_column_name_as_expr=True)
+                for v in flat_exprs
+            )
+        ),
         is_order_dependent=operation_is_order_dependent(*flat_exprs),
         changes_length=operation_changes_length(*flat_exprs),
         aggregates=operation_aggregates(*flat_exprs),
@@ -2511,7 +2539,12 @@ def mean_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:
         raise ValueError(msg)
     flat_exprs = flatten(exprs)
     return Expr(
-        lambda plx: plx.mean_horizontal(*[extract_compliant(plx, v) for v in flat_exprs]),
+        lambda plx: plx.mean_horizontal(
+            *(
+                extract_compliant(plx, v, parse_column_name_as_expr=True)
+                for v in flat_exprs
+            )
+        ),
         is_order_dependent=operation_is_order_dependent(*flat_exprs),
         changes_length=operation_changes_length(*flat_exprs),
         aggregates=operation_aggregates(*flat_exprs),
@@ -2596,15 +2629,14 @@ def concat_str(
         ----
         full_sentence: [["2 dogs play","4 cats swim",null]]
     """
-    flat_exprs = flatten([exprs])
+    exprs = flatten([*flatten([exprs]), *more_exprs])
     return Expr(
         lambda plx: plx.concat_str(
-            [extract_compliant(plx, v) for v in flat_exprs],
-            *[extract_compliant(plx, v) for v in more_exprs],
+            *(extract_compliant(plx, v, parse_column_name_as_expr=True) for v in exprs),
             separator=separator,
             ignore_nulls=ignore_nulls,
         ),
-        is_order_dependent=operation_is_order_dependent(*flat_exprs, *more_exprs),
-        changes_length=operation_changes_length(*flat_exprs, *more_exprs),
-        aggregates=operation_aggregates(*flat_exprs, *more_exprs),
+        is_order_dependent=operation_is_order_dependent(*exprs),
+        changes_length=operation_changes_length(*exprs),
+        aggregates=operation_aggregates(*exprs),
     )
diff --git a/narwhals/group_by.py b/narwhals/group_by.py
index 11ad5798..ae818563 100644
--- a/narwhals/group_by.py
+++ b/narwhals/group_by.py
@@ -11,12 +11,13 @@ from typing import cast
 from narwhals.dataframe import DataFrame
 from narwhals.dataframe import LazyFrame
 from narwhals.exceptions import InvalidOperationError
+from narwhals.utils import flatten
 from narwhals.utils import tupleify
 
 if TYPE_CHECKING:
     from typing_extensions import Self
 
-    from narwhals.typing import IntoExpr
+    from narwhals.expr import Expr
 
 DataFrameT = TypeVar("DataFrameT")
 LazyFrameT = TypeVar("LazyFrameT")
@@ -30,9 +31,7 @@ class GroupBy(Generic[DataFrameT]):
             *self._keys, drop_null_keys=drop_null_keys
         )
 
-    def agg(
-        self: Self, *aggs: IntoExpr | Iterable[IntoExpr], **named_aggs: IntoExpr
-    ) -> DataFrameT:
+    def agg(self: Self, *aggs: Expr | Iterable[Expr], **named_aggs: Expr) -> DataFrameT:
         """Compute aggregations for each group of a group by operation.
 
         Arguments:
@@ -112,7 +111,8 @@ class GroupBy(Generic[DataFrameT]):
             │ c   ┆ 3   ┆ 1   │
             └─────┴─────┴─────┘
         """
-        if not all(getattr(x, "_aggregates", True) for x in aggs) and all(
+        flat_aggs = tuple(flatten(aggs))
+        if not all(getattr(x, "_aggregates", True) for x in flat_aggs) and all(
             getattr(x, "_aggregates", True) for x in named_aggs.values()
         ):
             msg = (
@@ -122,9 +122,16 @@ class GroupBy(Generic[DataFrameT]):
                 "but `df.group_by('a').agg(nw.col('b'))` is not."
             )
             raise InvalidOperationError(msg)
-        aggs, named_aggs = self._df._flatten_and_extract(*aggs, **named_aggs)
+        plx = self._df.__narwhals_namespace__()
+        compliant_aggs = (
+            *(x._to_compliant_expr(plx) for x in flat_aggs),
+            *(
+                value._to_compliant_expr(plx).alias(key)
+                for key, value in named_aggs.items()
+            ),
+        )
         return self._df._from_compliant_dataframe(  # type: ignore[return-value]
-            self._grouped.agg(*aggs, **named_aggs),
+            self._grouped.agg(*compliant_aggs),
         )
 
     def __iter__(self: Self) -> Iterator[tuple[Any, DataFrameT]]:
@@ -142,9 +149,7 @@ class LazyGroupBy(Generic[LazyFrameT]):
             *self._keys, drop_null_keys=drop_null_keys
         )
 
-    def agg(
-        self: Self, *aggs: IntoExpr | Iterable[IntoExpr], **named_aggs: IntoExpr
-    ) -> LazyFrameT:
+    def agg(self: Self, *aggs: Expr | Iterable[Expr], **named_aggs: Expr) -> LazyFrameT:
         """Compute aggregations for each group of a group by operation.
 
         If a library does not support lazy execution, then this is a no-op.
@@ -208,7 +213,8 @@ class LazyGroupBy(Generic[LazyFrameT]):
             │ c   ┆ 3   ┆ 1   │
             └─────┴─────┴─────┘
         """
-        if not all(getattr(x, "_aggregates", True) for x in aggs) and all(
+        flat_aggs = tuple(flatten(aggs))
+        if not all(getattr(x, "_aggregates", True) for x in flat_aggs) and all(
             getattr(x, "_aggregates", True) for x in named_aggs.values()
         ):
             msg = (
@@ -218,7 +224,14 @@ class LazyGroupBy(Generic[LazyFrameT]):
                 "but `df.group_by('a').agg(nw.col('b'))` is not."
             )
             raise InvalidOperationError(msg)
-        aggs, named_aggs = self._df._flatten_and_extract(*aggs, **named_aggs)
+        plx = self._df.__narwhals_namespace__()
+        compliant_aggs = (
+            *(x._to_compliant_expr(plx) for x in flat_aggs),
+            *(
+                value._to_compliant_expr(plx).alias(key)
+                for key, value in named_aggs.items()
+            ),
+        )
         return self._df._from_compliant_dataframe(  # type: ignore[return-value]
-            self._grouped.agg(*aggs, **named_aggs),
+            self._grouped.agg(*compliant_aggs),
         )
diff --git a/narwhals/stable/v1/__init__.py b/narwhals/stable/v1/__init__.py
index df443230..abafc412 100644
--- a/narwhals/stable/v1/__init__.py
+++ b/narwhals/stable/v1/__init__.py
@@ -18,6 +18,7 @@ from narwhals import selectors
 from narwhals.dataframe import DataFrame as NwDataFrame
 from narwhals.dataframe import LazyFrame as NwLazyFrame
 from narwhals.dependencies import get_polars
+from narwhals.exceptions import InvalidIntoExprError
 from narwhals.expr import Expr as NwExpr
 from narwhals.functions import Then as NwThen
 from narwhals.functions import When as NwWhen
@@ -265,7 +266,7 @@ class LazyFrame(NwLazyFrame[IntoFrameT]):
                 "- Used `pl.col` instead of `nw.col`?"
             )
             raise TypeError(msg)
-        return arg
+        raise InvalidIntoExprError.from_invalid_type(type(arg))
 
     def collect(self: Self) -> DataFrame[Any]:
         r"""Materialize this LazyFrame into a DataFrame.
diff --git a/narwhals/typing.py b/narwhals/typing.py
index 859e98df..c37b2435 100644
--- a/narwhals/typing.py
+++ b/narwhals/typing.py
@@ -54,6 +54,17 @@ class CompliantSeries(Protocol):
 class CompliantDataFrame(Protocol):
     def __narwhals_dataframe__(self) -> CompliantDataFrame: ...
     def __narwhals_namespace__(self) -> Any: ...
+    def simple_select(
+        self, *column_names: str
+    ) -> CompliantDataFrame: ...  # `select` where all args are column names
+
+
+class CompliantLazyFramme(Protocol):
+    def __narwhals_lazyframe__(self) -> CompliantDataFrame: ...
+    def __narwhals_namespace__(self) -> Any: ...
+    def simple_select(
+        self, *column_names: str
+    ) -> CompliantLazyFrame: ...  # `select` where all args are column names
 
 
 class CompliantLazyFrame(Protocol):
@@ -279,6 +290,14 @@ class DTypes:
     Unknown: type[dtypes.Unknown]
 
 
+if TYPE_CHECKING:
+    # This one needs to be in TYPE_CHECKING to pass on 3.9,
+    # and can only be defined after CompliantExpr has been defined
+    IntoCompliantExpr: TypeAlias = (
+        CompliantExpr[CompliantSeriesT_co] | CompliantSeriesT_co
+    )
+
+
 __all__ = [
     "CompliantDataFrame",
     "CompliantLazyFrame",


Test patch:
diff --git a/tests/expr_and_series/cast_test.py b/tests/expr_and_series/cast_test.py
index c725d8fd..60212363 100644
--- a/tests/expr_and_series/cast_test.py
+++ b/tests/expr_and_series/cast_test.py
@@ -55,6 +55,7 @@ SCHEMA = {
 }
 
 SPARK_INCOMPATIBLE_COLUMNS = {"e", "f", "g", "h", "l", "o", "p"}
+DUCKDB_INCOMPATIBLE_COLUMNS = {"l", "o", "p"}
 
 
 @pytest.mark.filterwarnings("ignore:casting period[M] values to int64:FutureWarning")
@@ -62,8 +63,6 @@ def test_cast(
     constructor: Constructor,
     request: pytest.FixtureRequest,
 ) -> None:
-    if "duckdb" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
     if "pyarrow_table_constructor" in str(constructor) and PYARROW_VERSION <= (
         15,
     ):  # pragma: no cover
@@ -74,6 +73,8 @@ def test_cast(
 
     if "pyspark" in str(constructor):
         incompatible_columns = SPARK_INCOMPATIBLE_COLUMNS  # pragma: no cover
+    elif "duckdb" in str(constructor):
+        incompatible_columns = DUCKDB_INCOMPATIBLE_COLUMNS  # pragma: no cover
     else:
         incompatible_columns = set()
 
diff --git a/tests/expr_and_series/clip_test.py b/tests/expr_and_series/clip_test.py
index 2ae9e043..edb29fe6 100644
--- a/tests/expr_and_series/clip_test.py
+++ b/tests/expr_and_series/clip_test.py
@@ -34,7 +34,7 @@ def test_clip_expr_expressified(
 
     data = {"a": [1, 2, 3, -4, 5], "lb": [3, 2, 1, 1, 1], "ub": [4, 4, 2, 2, 2]}
     df = nw.from_native(constructor(data))
-    result = df.select(nw.col("a").clip(nw.col("lb"), nw.col("ub") + 1))
+    result = df.select(nw.col("a").clip("lb", nw.col("ub") + 1))
     expected_dict = {"a": [3, 2, 3, 1, 3]}
     assert_equal_data(result, expected_dict)
 
diff --git a/tests/expr_and_series/concat_str_test.py b/tests/expr_and_series/concat_str_test.py
index e5273aad..33826955 100644
--- a/tests/expr_and_series/concat_str_test.py
+++ b/tests/expr_and_series/concat_str_test.py
@@ -38,7 +38,7 @@ def test_concat_str(
             nw.concat_str(
                 [
                     nw.col("a") * 2,
-                    nw.col("b"),
+                    "b",
                     nw.col("c"),
                 ],
                 separator=" ",
diff --git a/tests/expr_and_series/is_between_test.py b/tests/expr_and_series/is_between_test.py
index a24277fa..b7bd5e2f 100644
--- a/tests/expr_and_series/is_between_test.py
+++ b/tests/expr_and_series/is_between_test.py
@@ -34,7 +34,7 @@ def test_is_between(
 def test_is_between_expressified(constructor: Constructor) -> None:
     data = {"a": [1, 4, 2, 5], "b": [0, 5, 2, 4], "c": [9, 9, 9, 9]}
     df = nw.from_native(constructor(data))
-    result = df.select(nw.col("a").is_between(nw.col("b") * 0.9, nw.col("c") - 1))
+    result = df.select(nw.col("a").is_between(nw.col("b") * 0.9, "c"))
     expected_dict = {"a": [True, False, True, True]}
     assert_equal_data(result, expected_dict)
 
diff --git a/tests/expr_and_series/std_test.py b/tests/expr_and_series/std_test.py
index 0cc9b672..1f8735c1 100644
--- a/tests/expr_and_series/std_test.py
+++ b/tests/expr_and_series/std_test.py
@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-from contextlib import nullcontext as does_not_raise
-
 import pytest
 
 import narwhals.stable.v1 as nw
@@ -34,19 +32,14 @@ def test_std(constructor: Constructor, input_data: dict[str, list[float | None]]
         "z_ddof_0": [0.816497],
     }
     assert_equal_data(result, expected_results)
-    context = (
-        pytest.raises(NotImplementedError)
-        if "duckdb" in str(constructor)
-        else does_not_raise()
+
+    result = df.select(
+        nw.col("b").std(ddof=2).alias("b_ddof_2"),
     )
-    with context:
-        result = df.select(
-            nw.col("b").std(ddof=2).alias("b_ddof_2"),
-        )
-        expected_results = {
-            "b_ddof_2": [1.632993],
-        }
-        assert_equal_data(result, expected_results)
+    expected_results = {
+        "b_ddof_2": [1.632993],
+    }
+    assert_equal_data(result, expected_results)
 
 
 @pytest.mark.parametrize("input_data", [data, data_with_nulls])
diff --git a/tests/expr_and_series/var_test.py b/tests/expr_and_series/var_test.py
index 0edd8e30..10998540 100644
--- a/tests/expr_and_series/var_test.py
+++ b/tests/expr_and_series/var_test.py
@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-from contextlib import nullcontext as does_not_raise
-
 import pytest
 
 import narwhals.stable.v1 as nw
@@ -34,19 +32,14 @@ def test_var(constructor: Constructor, input_data: dict[str, list[float | None]]
         "z_ddof_0": [0.6666666666666666],
     }
     assert_equal_data(result, expected_results)
-    context = (
-        pytest.raises(NotImplementedError)
-        if "duckdb" in str(constructor)
-        else does_not_raise()
+
+    result = df.select(
+        nw.col("b").var(ddof=2).alias("b_ddof_2"),
     )
-    with context:
-        result = df.select(
-            nw.col("b").var(ddof=2).alias("b_ddof_2"),
-        )
-        expected_results = {
-            "b_ddof_2": [2.666666666666667],
-        }
-        assert_equal_data(result, expected_results)
+    expected_results = {
+        "b_ddof_2": [2.666666666666667],
+    }
+    assert_equal_data(result, expected_results)
 
 
 @pytest.mark.parametrize("input_data", [data, data_with_nulls])
diff --git a/tests/frame/invalid_test.py b/tests/frame/invalid_test.py
index 967481f3..e5eee63b 100644
--- a/tests/frame/invalid_test.py
+++ b/tests/frame/invalid_test.py
@@ -17,7 +17,7 @@ def test_invalid() -> None:
     df = nw.from_native(pd.DataFrame(data))
     with pytest.raises(ValueError, match="Multi-output"):
         df.select(nw.all() + nw.all())
-    with pytest.raises(TypeError, match="Perhaps you:"):
+    with pytest.raises(TypeError, match="Perhaps you"):
         df.select([pl.col("a")])  # type: ignore[list-item]
     with pytest.raises(TypeError, match="Expected Narwhals dtype"):
         df.select([nw.col("a").cast(pl.Int64)])  # type: ignore[arg-type]
diff --git a/tests/frame/select_test.py b/tests/frame/select_test.py
index 3593eed2..36b8fc88 100644
--- a/tests/frame/select_test.py
+++ b/tests/frame/select_test.py
@@ -48,12 +48,7 @@ def test_int_select_pandas() -> None:
 
 
 @pytest.mark.parametrize("invalid_select", [None, 0, Foo()])
-def test_invalid_select(
-    constructor: Constructor, invalid_select: Any, request: pytest.FixtureRequest
-) -> None:
-    if "polars" in str(constructor) and not isinstance(invalid_select, Foo):
-        # https://github.com/narwhals-dev/narwhals/issues/1390
-        request.applymarker(pytest.mark.xfail)
+def test_invalid_select(constructor: Constructor, invalid_select: Any) -> None:
     with pytest.raises(InvalidIntoExprError):
         nw.from_native(constructor({"a": [1, 2, 3]})).select(invalid_select)
 
diff --git a/tests/group_by_test.py b/tests/group_by_test.py
index 3ba35fab..9929d36c 100644
--- a/tests/group_by_test.py
+++ b/tests/group_by_test.py
@@ -137,11 +137,7 @@ def test_group_by_depth_1_agg(
         ("var", 2),
     ],
 )
-def test_group_by_depth_1_std_var(
-    constructor: Constructor, attr: str, ddof: int, request: pytest.FixtureRequest
-) -> None:
-    if "duckdb" in str(constructor) and ddof == 2:
-        request.applymarker(pytest.mark.xfail)
+def test_group_by_depth_1_std_var(constructor: Constructor, attr: str, ddof: int) -> None:
     data = {"a": [1, 1, 1, 2, 2, 2], "b": [4, 5, 6, 0, 5, 5]}
     _pow = 0.5 if attr == "std" else 1
     expected = {
@@ -398,8 +394,6 @@ def test_all_kind_of_aggs(
         # and modin lol https://github.com/modin-project/modin/issues/7414
         # and cudf https://github.com/rapidsai/cudf/issues/17649
         request.applymarker(pytest.mark.xfail)
-    if "duckdb" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
     if "pandas" in str(constructor) and PANDAS_VERSION < (1, 4):
         # Bug in old pandas, can't do DataFrameGroupBy[['b', 'b']]
         request.applymarker(pytest.mark.xfail)