Repository: narwhals-dev/narwhals
Pull request #1880 (2025-01-28 08:37:18)

feat: move robust literal handling for PySpark
<!--
# 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



Patch:
diff --git a/docs/api-reference/selectors.md b/docs/api-reference/selectors.md
index 6480a869..aa990ab1 100644
--- a/docs/api-reference/selectors.md
+++ b/docs/api-reference/selectors.md
@@ -15,6 +15,7 @@ set operations are supported:
         - boolean
         - by_dtype
         - categorical
+        - matches
         - numeric
         - string
       show_root_heading: false
diff --git a/narwhals/_arrow/selectors.py b/narwhals/_arrow/selectors.py
index c83f4a6a..e82a99c0 100644
--- a/narwhals/_arrow/selectors.py
+++ b/narwhals/_arrow/selectors.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import re
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Sequence
@@ -43,6 +44,24 @@ class ArrowSelectorNamespace:
             kwargs={"dtypes": dtypes},
         )
 
+    def matches(self: Self, pattern: str) -> ArrowSelector:
+        def func(df: ArrowDataFrame) -> list[ArrowSeries]:
+            return [df[col] for col in df.columns if re.search(pattern, col)]
+
+        def evalute_output_names(df: ArrowDataFrame) -> Sequence[str]:
+            return [col for col in df.columns if re.search(pattern, col)]
+
+        return ArrowSelector(
+            func,
+            depth=0,
+            function_name="selector",
+            evaluate_output_names=evalute_output_names,
+            alias_output_names=None,
+            backend_version=self._backend_version,
+            version=self._version,
+            kwargs={"pattern": pattern},
+        )
+
     def numeric(self: Self) -> ArrowSelector:
         dtypes = import_dtypes_module(self._version)
         return self.by_dtype(
diff --git a/narwhals/_dask/selectors.py b/narwhals/_dask/selectors.py
index 96f3feb0..8645d77a 100644
--- a/narwhals/_dask/selectors.py
+++ b/narwhals/_dask/selectors.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import re
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Sequence
@@ -48,6 +49,27 @@ class DaskSelectorNamespace:
             kwargs={},
         )
 
+    def matches(self: Self, pattern: str) -> DaskSelector:
+        def func(df: DaskLazyFrame) -> list[dx.Series]:
+            return [
+                df._native_frame[col] for col in df.columns if re.search(pattern, col)
+            ]
+
+        def evalute_output_names(df: DaskLazyFrame) -> Sequence[str]:
+            return [col for col in df.columns if re.search(pattern, col)]
+
+        return DaskSelector(
+            func,
+            depth=0,
+            function_name="selector",
+            evaluate_output_names=evalute_output_names,
+            alias_output_names=None,
+            backend_version=self._backend_version,
+            returns_scalar=False,
+            version=self._version,
+            kwargs={},
+        )
+
     def numeric(self: Self) -> DaskSelector:
         dtypes = import_dtypes_module(self._version)
         return self.by_dtype(
diff --git a/narwhals/_duckdb/selectors.py b/narwhals/_duckdb/selectors.py
index 59848a80..30fd32e3 100644
--- a/narwhals/_duckdb/selectors.py
+++ b/narwhals/_duckdb/selectors.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import re
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Sequence
@@ -45,6 +46,25 @@ class DuckDBSelectorNamespace:
             version=self._version,
         )
 
+    def matches(self: Self, pattern: str) -> DuckDBSelector:
+        def func(df: DuckDBLazyFrame) -> list[duckdb.Expression]:
+            return [
+                ColumnExpression(col) for col in df.columns if re.search(pattern, col)
+            ]
+
+        def evalute_output_names(df: DuckDBLazyFrame) -> Sequence[str]:
+            return [col for col in df.columns if re.search(pattern, col)]
+
+        return DuckDBSelector(
+            func,
+            function_name="selector",
+            evaluate_output_names=evalute_output_names,
+            alias_output_names=None,
+            backend_version=self._backend_version,
+            expr_kind=ExprKind.TRANSFORM,
+            version=self._version,
+        )
+
     def numeric(self: Self) -> DuckDBSelector:
         dtypes = import_dtypes_module(self._version)
         return self.by_dtype(
diff --git a/narwhals/_pandas_like/selectors.py b/narwhals/_pandas_like/selectors.py
index d7e18a2e..2d90adc9 100644
--- a/narwhals/_pandas_like/selectors.py
+++ b/narwhals/_pandas_like/selectors.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import re
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Sequence
@@ -48,6 +49,25 @@ class PandasSelectorNamespace:
             kwargs={"dtypes": dtypes},
         )
 
+    def matches(self: Self, pattern: str) -> PandasSelector:
+        def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
+            return [df[col] for col in df.columns if re.search(pattern, col)]
+
+        def evalute_output_names(df: PandasLikeDataFrame) -> Sequence[str]:
+            return [col for col in df.columns if re.search(pattern, col)]
+
+        return PandasSelector(
+            func,
+            depth=0,
+            function_name="selector",
+            evaluate_output_names=evalute_output_names,
+            alias_output_names=None,
+            implementation=self._implementation,
+            backend_version=self._backend_version,
+            version=self._version,
+            kwargs={"pattern": pattern},
+        )
+
     def numeric(self: Self) -> PandasSelector:
         dtypes = import_dtypes_module(self._version)
         return self.by_dtype(
diff --git a/narwhals/_polars/namespace.py b/narwhals/_polars/namespace.py
index 005af360..4a73c0ed 100644
--- a/narwhals/_polars/namespace.py
+++ b/narwhals/_polars/namespace.py
@@ -217,6 +217,17 @@ class PolarsSelectors:
             backend_version=self._backend_version,
         )
 
+    def matches(self: Self, pattern: str) -> PolarsExpr:
+        import polars as pl
+
+        from narwhals._polars.expr import PolarsExpr
+
+        return PolarsExpr(
+            pl.selectors.matches(pattern=pattern),
+            version=self._version,
+            backend_version=self._backend_version,
+        )
+
     def numeric(self: Self) -> PolarsExpr:
         import polars as pl
 
diff --git a/narwhals/_spark_like/dataframe.py b/narwhals/_spark_like/dataframe.py
index 12e17549..367b66da 100644
--- a/narwhals/_spark_like/dataframe.py
+++ b/narwhals/_spark_like/dataframe.py
@@ -9,6 +9,7 @@ from typing import Sequence
 from pyspark.sql import Window
 from pyspark.sql import functions as F  # noqa: N812
 
+from narwhals._spark_like.utils import ExprKind
 from narwhals._spark_like.utils import native_to_narwhals_dtype
 from narwhals._spark_like.utils import parse_exprs_and_named_exprs
 from narwhals.typing import CompliantLazyFrame
@@ -97,9 +98,7 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
         *exprs: SparkLikeExpr,
         **named_exprs: SparkLikeExpr,
     ) -> Self:
-        new_columns, returns_scalar = parse_exprs_and_named_exprs(
-            self, *exprs, **named_exprs
-        )
+        new_columns, expr_kinds = parse_exprs_and_named_exprs(self, *exprs, **named_exprs)
 
         if not new_columns:
             # return empty dataframe, like Polars does
@@ -110,7 +109,7 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
 
             return self._from_native_frame(spark_df)
 
-        if all(returns_scalar):
+        if not any(expr_kind is ExprKind.TRANSFORM for expr_kind in expr_kinds):
             new_columns_list = [
                 col.alias(col_name) for col_name, col in new_columns.items()
             ]
@@ -118,11 +117,9 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
         else:
             new_columns_list = [
                 col.over(Window.partitionBy(F.lit(1))).alias(col_name)
-                if _returns_scalar
+                if expr_kind is ExprKind.AGGREGATION
                 else col.alias(col_name)
-                for (col_name, col), _returns_scalar in zip(
-                    new_columns.items(), returns_scalar
-                )
+                for (col_name, col), expr_kind in zip(new_columns.items(), expr_kinds)
             ]
             return self._from_native_frame(self._native_frame.select(*new_columns_list))
 
@@ -131,15 +128,13 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
         *exprs: SparkLikeExpr,
         **named_exprs: SparkLikeExpr,
     ) -> Self:
-        new_columns, returns_scalar = parse_exprs_and_named_exprs(
-            self, *exprs, **named_exprs
-        )
+        new_columns, expr_kinds = parse_exprs_and_named_exprs(self, *exprs, **named_exprs)
 
         new_columns_map = {
-            col_name: col.over(Window.partitionBy(F.lit(1))) if _returns_scalar else col
-            for (col_name, col), _returns_scalar in zip(
-                new_columns.items(), returns_scalar
-            )
+            col_name: col.over(Window.partitionBy(F.lit(1)))
+            if expr_kind is ExprKind.AGGREGATION
+            else col
+            for (col_name, col), expr_kind in zip(new_columns.items(), expr_kinds)
         }
         return self._from_native_frame(self._native_frame.withColumns(new_columns_map))
 
diff --git a/narwhals/_spark_like/expr.py b/narwhals/_spark_like/expr.py
index 5483f1e7..36c40f2d 100644
--- a/narwhals/_spark_like/expr.py
+++ b/narwhals/_spark_like/expr.py
@@ -12,8 +12,9 @@ from pyspark.sql import functions as F  # noqa: N812
 from narwhals._spark_like.expr_dt import SparkLikeExprDateTimeNamespace
 from narwhals._spark_like.expr_name import SparkLikeExprNameNamespace
 from narwhals._spark_like.expr_str import SparkLikeExprStringNamespace
-from narwhals._spark_like.utils import binary_operation_returns_scalar
+from narwhals._spark_like.utils import ExprKind
 from narwhals._spark_like.utils import maybe_evaluate
+from narwhals._spark_like.utils import n_ary_operation_expr_kind
 from narwhals._spark_like.utils import narwhals_to_native_dtype
 from narwhals.typing import CompliantExpr
 from narwhals.utils import Implementation
@@ -40,9 +41,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
         function_name: str,
         evaluate_output_names: Callable[[SparkLikeLazyFrame], Sequence[str]],
         alias_output_names: Callable[[Sequence[str]], Sequence[str]] | None,
-        # Whether the expression is a length-1 Column resulting from
-        # a reduction, such as `nw.col('a').sum()`
-        returns_scalar: bool,
+        expr_kind: ExprKind,
         backend_version: tuple[int, ...],
         version: Version,
     ) -> None:
@@ -50,7 +49,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
         self._function_name = function_name
         self._evaluate_output_names = evaluate_output_names
         self._alias_output_names = alias_output_names
-        self._returns_scalar = returns_scalar
+        self._expr_kind = expr_kind
         self._backend_version = backend_version
         self._version = version
 
@@ -82,7 +81,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             function_name="col",
             evaluate_output_names=lambda _df: list(column_names),
             alias_output_names=None,
-            returns_scalar=False,
+            expr_kind=ExprKind.TRANSFORM,
             backend_version=backend_version,
             version=version,
         )
@@ -103,7 +102,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             function_name="nth",
             evaluate_output_names=lambda df: [df.columns[i] for i in column_indices],
             alias_output_names=None,
-            returns_scalar=False,
+            expr_kind=ExprKind.TRANSFORM,
             backend_version=backend_version,
             version=version,
         )
@@ -113,13 +112,13 @@ class SparkLikeExpr(CompliantExpr["Column"]):
         call: Callable[..., Column],
         expr_name: str,
         *,
-        returns_scalar: bool,
+        expr_kind: ExprKind,
         **expressifiable_args: Self | Any,
     ) -> Self:
         def func(df: SparkLikeLazyFrame) -> list[Column]:
             native_series_list = self._call(df)
             other_native_series = {
-                key: maybe_evaluate(df, value, returns_scalar=returns_scalar)
+                key: maybe_evaluate(df, value, expr_kind=expr_kind)
                 for key, value in expressifiable_args.items()
             }
             return [
@@ -132,7 +131,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             function_name=f"{self._function_name}->{expr_name}",
             evaluate_output_names=self._evaluate_output_names,
             alias_output_names=self._alias_output_names,
-            returns_scalar=returns_scalar,
+            expr_kind=expr_kind,
             backend_version=self._backend_version,
             version=self._version,
         )
@@ -142,7 +141,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             lambda _input, other: _input.__eq__(other),
             "__eq__",
             other=other,
-            returns_scalar=binary_operation_returns_scalar(self, other),
+            expr_kind=n_ary_operation_expr_kind(self, other),
         )
 
     def __ne__(self: Self, other: SparkLikeExpr) -> Self:  # type: ignore[override]
@@ -150,7 +149,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             lambda _input, other: _input.__ne__(other),
             "__ne__",
             other=other,
-            returns_scalar=binary_operation_returns_scalar(self, other),
+            expr_kind=n_ary_operation_expr_kind(self, other),
         )
 
     def __add__(self: Self, other: SparkLikeExpr) -> Self:
@@ -158,7 +157,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             lambda _input, other: _input.__add__(other),
             "__add__",
             other=other,
-            returns_scalar=binary_operation_returns_scalar(self, other),
+            expr_kind=n_ary_operation_expr_kind(self, other),
         )
 
     def __sub__(self: Self, other: SparkLikeExpr) -> Self:
@@ -166,7 +165,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             lambda _input, other: _input.__sub__(other),
             "__sub__",
             other=other,
-            returns_scalar=binary_operation_returns_scalar(self, other),
+            expr_kind=n_ary_operation_expr_kind(self, other),
         )
 
     def __mul__(self: Self, other: SparkLikeExpr) -> Self:
@@ -174,7 +173,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             lambda _input, other: _input.__mul__(other),
             "__mul__",
             other=other,
-            returns_scalar=binary_operation_returns_scalar(self, other),
+            expr_kind=n_ary_operation_expr_kind(self, other),
         )
 
     def __truediv__(self: Self, other: SparkLikeExpr) -> Self:
@@ -182,7 +181,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             lambda _input, other: _input.__truediv__(other),
             "__truediv__",
             other=other,
-            returns_scalar=binary_operation_returns_scalar(self, other),
+            expr_kind=n_ary_operation_expr_kind(self, other),
         )
 
     def __floordiv__(self: Self, other: SparkLikeExpr) -> Self:
@@ -193,7 +192,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             _floordiv,
             "__floordiv__",
             other=other,
-            returns_scalar=binary_operation_returns_scalar(self, other),
+            expr_kind=n_ary_operation_expr_kind(self, other),
         )
 
     def __pow__(self: Self, other: SparkLikeExpr) -> Self:
@@ -201,7 +200,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             lambda _input, other: _input.__pow__(other),
             "__pow__",
             other=other,
-            returns_scalar=binary_operation_returns_scalar(self, other),
+            expr_kind=n_ary_operation_expr_kind(self, other),
         )
 
     def __mod__(self: Self, other: SparkLikeExpr) -> Self:
@@ -209,7 +208,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             lambda _input, other: _input.__mod__(other),
             "__mod__",
             other=other,
-            returns_scalar=binary_operation_returns_scalar(self, other),
+            expr_kind=n_ary_operation_expr_kind(self, other),
         )
 
     def __ge__(self: Self, other: SparkLikeExpr) -> Self:
@@ -217,7 +216,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             lambda _input, other: _input.__ge__(other),
             "__ge__",
             other=other,
-            returns_scalar=binary_operation_returns_scalar(self, other),
+            expr_kind=n_ary_operation_expr_kind(self, other),
         )
 
     def __gt__(self: Self, other: SparkLikeExpr) -> Self:
@@ -225,7 +224,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             lambda _input, other: _input > other,
             "__gt__",
             other=other,
-            returns_scalar=binary_operation_returns_scalar(self, other),
+            expr_kind=n_ary_operation_expr_kind(self, other),
         )
 
     def __le__(self: Self, other: SparkLikeExpr) -> Self:
@@ -233,7 +232,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             lambda _input, other: _input.__le__(other),
             "__le__",
             other=other,
-            returns_scalar=binary_operation_returns_scalar(self, other),
+            expr_kind=n_ary_operation_expr_kind(self, other),
         )
 
     def __lt__(self: Self, other: SparkLikeExpr) -> Self:
@@ -241,7 +240,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             lambda _input, other: _input.__lt__(other),
             "__lt__",
             other=other,
-            returns_scalar=binary_operation_returns_scalar(self, other),
+            expr_kind=n_ary_operation_expr_kind(self, other),
         )
 
     def __and__(self: Self, other: SparkLikeExpr) -> Self:
@@ -249,7 +248,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             lambda _input, other: _input.__and__(other),
             "__and__",
             other=other,
-            returns_scalar=binary_operation_returns_scalar(self, other),
+            expr_kind=n_ary_operation_expr_kind(self, other),
         )
 
     def __or__(self: Self, other: SparkLikeExpr) -> Self:
@@ -257,18 +256,18 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             lambda _input, other: _input.__or__(other),
             "__or__",
             other=other,
-            returns_scalar=binary_operation_returns_scalar(self, other),
+            expr_kind=n_ary_operation_expr_kind(self, other),
         )
 
     def __invert__(self: Self) -> Self:
         return self._from_call(
             lambda _input: _input.__invert__(),
             "__invert__",
-            returns_scalar=self._returns_scalar,
+            expr_kind=self._expr_kind,
         )
 
     def abs(self: Self) -> Self:
-        return self._from_call(F.abs, "abs", returns_scalar=self._returns_scalar)
+        return self._from_call(F.abs, "abs", expr_kind=self._expr_kind)
 
     def alias(self: Self, name: str) -> Self:
         def alias_output_names(names: Sequence[str]) -> Sequence[str]:
@@ -282,32 +281,32 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             function_name=self._function_name,
             evaluate_output_names=self._evaluate_output_names,
             alias_output_names=alias_output_names,
-            returns_scalar=self._returns_scalar,
+            expr_kind=self._expr_kind,
             backend_version=self._backend_version,
             version=self._version,
         )
 
     def all(self: Self) -> Self:
-        return self._from_call(F.bool_and, "all", returns_scalar=True)
+        return self._from_call(F.bool_and, "all", expr_kind=ExprKind.AGGREGATION)
 
     def any(self: Self) -> Self:
-        return self._from_call(F.bool_or, "any", returns_scalar=True)
+        return self._from_call(F.bool_or, "any", expr_kind=ExprKind.AGGREGATION)
 
     def cast(self: Self, dtype: DType | type[DType]) -> Self:
         def _cast(_input: Column) -> Column:
             spark_dtype = narwhals_to_native_dtype(dtype, self._version)
             return _input.cast(spark_dtype)
 
-        return self._from_call(_cast, "cast", returns_scalar=self._returns_scalar)
+        return self._from_call(_cast, "cast", expr_kind=self._expr_kind)
 
     def count(self: Self) -> Self:
-        return self._from_call(F.count, "count", returns_scalar=True)
+        return self._from_call(F.count, "count", expr_kind=ExprKind.AGGREGATION)
 
     def max(self: Self) -> Self:
-        return self._from_call(F.max, "max", returns_scalar=True)
+        return self._from_call(F.max, "max", expr_kind=ExprKind.AGGREGATION)
 
     def mean(self: Self) -> Self:
-        return self._from_call(F.mean, "mean", returns_scalar=True)
+        return self._from_call(F.mean, "mean", expr_kind=ExprKind.AGGREGATION)
 
     def median(self: Self) -> Self:
         def _median(_input: Column) -> Column:
@@ -319,19 +318,19 @@ class SparkLikeExpr(CompliantExpr["Column"]):
 
             return F.median(_input)
 
-        return self._from_call(_median, "median", returns_scalar=True)
+        return self._from_call(_median, "median", expr_kind=ExprKind.AGGREGATION)
 
     def min(self: Self) -> Self:
-        return self._from_call(F.min, "min", returns_scalar=True)
+        return self._from_call(F.min, "min", expr_kind=ExprKind.AGGREGATION)
 
     def null_count(self: Self) -> Self:
         def _null_count(_input: Column) -> Column:
             return F.count_if(F.isnull(_input))
 
-        return self._from_call(_null_count, "null_count", returns_scalar=True)
+        return self._from_call(_null_count, "null_count", expr_kind=ExprKind.AGGREGATION)
 
     def sum(self: Self) -> Self:
-        return self._from_call(F.sum, "sum", returns_scalar=True)
+        return self._from_call(F.sum, "sum", expr_kind=ExprKind.AGGREGATION)
 
     def std(self: Self, ddof: int) -> Self:
         from functools import partial
@@ -342,7 +341,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
 
         func = partial(_std, ddof=ddof, np_version=parse_version(np.__version__))
 
-        return self._from_call(func, "std", returns_scalar=True)
+        return self._from_call(func, "std", expr_kind=ExprKind.AGGREGATION)
 
     def var(self: Self, ddof: int) -> Self:
         from functools import partial
@@ -353,7 +352,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
 
         func = partial(_var, ddof=ddof, np_version=parse_version(np.__version__))
 
-        return self._from_call(func, "var", returns_scalar=True)
+        return self._from_call(func, "var", expr_kind=ExprKind.AGGREGATION)
 
     def clip(
         self: Self,
@@ -379,7 +378,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             "clip",
             lower_bound=lower_bound,
             upper_bound=upper_bound,
-            returns_scalar=self._returns_scalar,
+            expr_kind=self._expr_kind,
         )
 
     def is_between(
@@ -402,7 +401,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             "is_between",
             lower_bound=lower_bound,
             upper_bound=upper_bound,
-            returns_scalar=self._returns_scalar,
+            expr_kind=self._expr_kind,
         )
 
     def is_duplicated(self: Self) -> Self:
@@ -410,9 +409,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             # Create a window spec that treats each value separately.
             return F.count("*").over(Window.partitionBy(_input)) > 1
 
-        return self._from_call(
-            _is_duplicated, "is_duplicated", returns_scalar=self._returns_scalar
-        )
+        return self._from_call(_is_duplicated, "is_duplicated", expr_kind=self._expr_kind)
 
     def is_finite(self: Self) -> Self:
         def _is_finite(_input: Column) -> Column:
@@ -423,9 +420,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             )
             return F.when(~F.isnull(_input), is_finite_condition).otherwise(None)
 
-        return self._from_call(
-            _is_finite, "is_finite", returns_scalar=self._returns_scalar
-        )
+        return self._from_call(_is_finite, "is_finite", expr_kind=self._expr_kind)
 
     def is_in(self: Self, values: Sequence[Any]) -> Self:
         def _is_in(_input: Column) -> Column:
@@ -434,7 +429,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
         return self._from_call(
             _is_in,
             "is_in",
-            returns_scalar=self._returns_scalar,
+            expr_kind=self._expr_kind,
         )
 
     def is_unique(self: Self) -> Self:
@@ -442,16 +437,14 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             # Create a window spec that treats each value separately
             return F.count("*").over(Window.partitionBy(_input)) == 1
 
-        return self._from_call(
-            _is_unique, "is_unique", returns_scalar=self._returns_scalar
-        )
+        return self._from_call(_is_unique, "is_unique", expr_kind=self._expr_kind)
 
     def len(self: Self) -> Self:
         def _len(_input: Column) -> Column:
             # Use count(*) to count all rows including nulls
             return F.count("*")
 
-        return self._from_call(_len, "len", returns_scalar=True)
+        return self._from_call(_len, "len", expr_kind=ExprKind.AGGREGATION)
 
     def round(self: Self, decimals: int) -> Self:
         def _round(_input: Column) -> Column:
@@ -460,11 +453,11 @@ class SparkLikeExpr(CompliantExpr["Column"]):
         return self._from_call(
             _round,
             "round",
-            returns_scalar=self._returns_scalar,
+            expr_kind=self._expr_kind,
         )
 
     def skew(self: Self) -> Self:
-        return self._from_call(F.skewness, "skew", returns_scalar=True)
+        return self._from_call(F.skewness, "skew", expr_kind=ExprKind.AGGREGATION)
 
     def n_unique(self: Self) -> Self:
         from pyspark.sql.types import IntegerType
@@ -472,7 +465,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
         def _n_unique(_input: Column) -> Column:
             return F.count_distinct(_input) + F.max(F.isnull(_input).cast(IntegerType()))
 
-        return self._from_call(_n_unique, "n_unique", returns_scalar=True)
+        return self._from_call(_n_unique, "n_unique", expr_kind=ExprKind.AGGREGATION)
 
     def over(self: Self, keys: list[str]) -> Self:
         def func(df: SparkLikeLazyFrame) -> list[Column]:
@@ -485,17 +478,17 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             alias_output_names=self._alias_output_names,
             backend_version=self._backend_version,
             version=self._version,
-            returns_scalar=False,
+            expr_kind=ExprKind.TRANSFORM,
         )
 
     def is_null(self: Self) -> Self:
-        return self._from_call(F.isnull, "is_null", returns_scalar=self._returns_scalar)
+        return self._from_call(F.isnull, "is_null", expr_kind=self._expr_kind)
 
     def is_nan(self: Self) -> Self:
         def _is_nan(_input: Column) -> Column:
             return F.when(F.isnull(_input), None).otherwise(F.isnan(_input))
 
-        return self._from_call(_is_nan, "is_nan", returns_scalar=self._returns_scalar)
+        return self._from_call(_is_nan, "is_nan", expr_kind=self._expr_kind)
 
     @property
     def str(self: Self) -> SparkLikeExprStringNamespace:
diff --git a/narwhals/_spark_like/expr_dt.py b/narwhals/_spark_like/expr_dt.py
index ad5d58b8..d0229b38 100644
--- a/narwhals/_spark_like/expr_dt.py
+++ b/narwhals/_spark_like/expr_dt.py
@@ -19,49 +19,49 @@ class SparkLikeExprDateTimeNamespace:
         return self._compliant_expr._from_call(
             F.to_date,
             "date",
-            returns_scalar=self._compliant_expr._returns_scalar,
+            expr_kind=self._compliant_expr._expr_kind,
         )
 
     def year(self: Self) -> SparkLikeExpr:
         return self._compliant_expr._from_call(
             F.year,
             "year",
-            returns_scalar=self._compliant_expr._returns_scalar,
+            expr_kind=self._compliant_expr._expr_kind,
         )
 
     def month(self: Self) -> SparkLikeExpr:
         return self._compliant_expr._from_call(
             F.month,
             "month",
-            returns_scalar=self._compliant_expr._returns_scalar,
+            expr_kind=self._compliant_expr._expr_kind,
         )
 
     def day(self: Self) -> SparkLikeExpr:
         return self._compliant_expr._from_call(
             F.day,
             "day",
-            returns_scalar=self._compliant_expr._returns_scalar,
+            expr_kind=self._compliant_expr._expr_kind,
         )
 
     def hour(self: Self) -> SparkLikeExpr:
         return self._compliant_expr._from_call(
             F.hour,
             "hour",
-            returns_scalar=self._compliant_expr._returns_scalar,
+            expr_kind=self._compliant_expr._expr_kind,
         )
 
     def minute(self: Self) -> SparkLikeExpr:
         return self._compliant_expr._from_call(
             F.minute,
             "minute",
-            returns_scalar=self._compliant_expr._returns_scalar,
+            expr_kind=self._compliant_expr._expr_kind,
         )
 
     def second(self: Self) -> SparkLikeExpr:
         return self._compliant_expr._from_call(
             F.second,
             "second",
-            returns_scalar=self._compliant_expr._returns_scalar,
+            expr_kind=self._compliant_expr._expr_kind,
         )
 
     def millisecond(self: Self) -> SparkLikeExpr:
@@ -71,7 +71,7 @@ class SparkLikeExprDateTimeNamespace:
         return self._compliant_expr._from_call(
             _millisecond,
             "millisecond",
-            returns_scalar=self._compliant_expr._returns_scalar,
+            expr_kind=self._compliant_expr._expr_kind,
         )
 
     def microsecond(self: Self) -> SparkLikeExpr:
@@ -81,7 +81,7 @@ class SparkLikeExprDateTimeNamespace:
         return self._compliant_expr._from_call(
             _microsecond,
             "microsecond",
-            returns_scalar=self._compliant_expr._returns_scalar,
+            expr_kind=self._compliant_expr._expr_kind,
         )
 
     def nanosecond(self: Self) -> SparkLikeExpr:
@@ -91,14 +91,14 @@ class SparkLikeExprDateTimeNamespace:
         return self._compliant_expr._from_call(
             _nanosecond,
             "nanosecond",
-            returns_scalar=self._compliant_expr._returns_scalar,
+            expr_kind=self._compliant_expr._expr_kind,
         )
 
     def ordinal_day(self: Self) -> SparkLikeExpr:
         return self._compliant_expr._from_call(
             F.dayofyear,
             "ordinal_day",
-            returns_scalar=self._compliant_expr._returns_scalar,
+            expr_kind=self._compliant_expr._expr_kind,
         )
 
     def weekday(self: Self) -> SparkLikeExpr:
@@ -109,5 +109,5 @@ class SparkLikeExprDateTimeNamespace:
         return self._compliant_expr._from_call(
             _weekday,
             "weekday",
-            returns_scalar=self._compliant_expr._returns_scalar,
+            expr_kind=self._compliant_expr._expr_kind,
         )
diff --git a/narwhals/_spark_like/expr_name.py b/narwhals/_spark_like/expr_name.py
index c1febdd7..bd3a6021 100644
--- a/narwhals/_spark_like/expr_name.py
+++ b/narwhals/_spark_like/expr_name.py
@@ -61,7 +61,7 @@ class SparkLikeExprNameNamespace:
             function_name=self._compliant_expr._function_name,
             evaluate_output_names=self._compliant_expr._evaluate_output_names,
             alias_output_names=alias_output_names,
-            returns_scalar=self._compliant_expr._returns_scalar,
+            expr_kind=self._compliant_expr._expr_kind,
             backend_version=self._compliant_expr._backend_version,
             version=self._compliant_expr._version,
         )
diff --git a/narwhals/_spark_like/expr_str.py b/narwhals/_spark_like/expr_str.py
index ff2223bd..e67fb14a 100644
--- a/narwhals/_spark_like/expr_str.py
+++ b/narwhals/_spark_like/expr_str.py
@@ -20,7 +20,7 @@ class SparkLikeExprStringNamespace:
         return self._compliant_expr._from_call(
             F.char_length,
             "len",
-            returns_scalar=self._compliant_expr._returns_scalar,
+            expr_kind=self._compliant_expr._expr_kind,
         )
 
     def replace_all(
@@ -33,7 +33,7 @@ class SparkLikeExprStringNamespace:
         return self._compliant_expr._from_call(
             func,
             "replace",
-            returns_scalar=self._compliant_expr._returns_scalar,
+            expr_kind=self._compliant_expr._expr_kind,
         )
 
     def strip_chars(self: Self, characters: str | None) -> SparkLikeExpr:
@@ -46,21 +46,21 @@ class SparkLikeExprStringNamespace:
         return self._compliant_expr._from_call(
             func,
             "strip",
-            returns_scalar=self._compliant_expr._returns_scalar,
+            expr_kind=self._compliant_expr._expr_kind,
         )
 
     def starts_with(self: Self, prefix: str) -> SparkLikeExpr:
         return self._compliant_expr._from_call(
             lambda _input: F.startswith(_input, F.lit(prefix)),
             "starts_with",
-            returns_scalar=self._compliant_expr._returns_scalar,
+            expr_kind=self._compliant_expr._expr_kind,
         )
 
     def ends_with(self: Self, suffix: str) -> SparkLikeExpr:
         return self._compliant_expr._from_call(
             lambda _input: F.endswith(_input, F.lit(suffix)),
             "ends_with",
-            returns_scalar=self._compliant_expr._returns_scalar,
+            expr_kind=self._compliant_expr._expr_kind,
         )
 
     def contains(self: Self, pattern: str, *, literal: bool) -> SparkLikeExpr:
@@ -71,7 +71,7 @@ class SparkLikeExprStringNamespace:
         return self._compliant_expr._from_call(
             func,
             "contains",
-            returns_scalar=self._compliant_expr._returns_scalar,
+            expr_kind=self._compliant_expr._expr_kind,
         )
 
     def slice(self: Self, offset: int, length: int | None) -> SparkLikeExpr:
@@ -87,21 +87,21 @@ class SparkLikeExprStringNamespace:
         return self._compliant_expr._from_call(
             func,
             "slice",
-            returns_scalar=self._compliant_expr._returns_scalar,
+            expr_kind=self._compliant_expr._expr_kind,
         )
 
     def to_uppercase(self: Self) -> SparkLikeExpr:
         return self._compliant_expr._from_call(
             F.upper,
             "to_uppercase",
-            returns_scalar=self._compliant_expr._returns_scalar,
+            expr_kind=self._compliant_expr._expr_kind,
         )
 
     def to_lowercase(self: Self) -> SparkLikeExpr:
         return self._compliant_expr._from_call(
             F.lower,
             "to_lowercase",
-            returns_scalar=self._compliant_expr._returns_scalar,
+            expr_kind=self._compliant_expr._expr_kind,
         )
 
     def to_datetime(self: Self, format: str | None) -> SparkLikeExpr:  # noqa: A002
@@ -111,7 +111,7 @@ class SparkLikeExprStringNamespace:
                 format=strptime_to_pyspark_format(format),
             ),
             "to_datetime",
-            returns_scalar=self._compliant_expr._returns_scalar,
+            expr_kind=self._compliant_expr._expr_kind,
         )
 
 
diff --git a/narwhals/_spark_like/namespace.py b/narwhals/_spark_like/namespace.py
index 5b0ce713..f39cd29d 100644
--- a/narwhals/_spark_like/namespace.py
+++ b/narwhals/_spark_like/namespace.py
@@ -17,6 +17,8 @@ from narwhals._expression_parsing import combine_evaluate_output_names
 from narwhals._spark_like.dataframe import SparkLikeLazyFrame
 from narwhals._spark_like.expr import SparkLikeExpr
 from narwhals._spark_like.selectors import SparkLikeSelectorNamespace
+from narwhals._spark_like.utils import ExprKind
+from narwhals._spark_like.utils import n_ary_operation_expr_kind
 from narwhals.typing import CompliantNamespace
 
 if TYPE_CHECKING:
@@ -50,7 +52,7 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
             function_name="all",
             evaluate_output_names=lambda df: df.columns,
             alias_output_names=None,
-            returns_scalar=False,
+            expr_kind=ExprKind.TRANSFORM,
             backend_version=self._backend_version,
             version=self._version,
         )
@@ -80,7 +82,7 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
             function_name="lit",
             evaluate_output_names=lambda _df: ["literal"],
             alias_output_names=None,
-            returns_scalar=True,
+            expr_kind=ExprKind.LITERAL,
             backend_version=self._backend_version,
             version=self._version,
         )
@@ -94,7 +96,7 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
             function_name="len",
             evaluate_output_names=lambda _df: ["len"],
             alias_output_names=None,
-            returns_scalar=True,
+            expr_kind=ExprKind.AGGREGATION,
             backend_version=self._backend_version,
             version=self._version,
         )
@@ -109,7 +111,7 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
             function_name="all_horizontal",
             evaluate_output_names=combine_evaluate_output_names(*exprs),
             alias_output_names=combine_alias_output_names(*exprs),
-            returns_scalar=False,
+            expr_kind=n_ary_operation_expr_kind(*exprs),
             backend_version=self._backend_version,
             version=self._version,
         )
@@ -124,7 +126,7 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
             function_name="any_horizontal",
             evaluate_output_names=combine_evaluate_output_names(*exprs),
             alias_output_names=combine_alias_output_names(*exprs),
-            returns_scalar=False,
+            expr_kind=n_ary_operation_expr_kind(*exprs),
             backend_version=self._backend_version,
             version=self._version,
         )
@@ -144,7 +146,7 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
             function_name="sum_horizontal",
             evaluate_output_names=combine_evaluate_output_names(*exprs),
             alias_output_names=combine_alias_output_names(*exprs),
-            returns_scalar=False,
+            expr_kind=n_ary_operation_expr_kind(*exprs),
             backend_version=self._backend_version,
             version=self._version,
         )
@@ -167,7 +169,7 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
             function_name="mean_horizontal",
             evaluate_output_names=combine_evaluate_output_names(*exprs),
             alias_output_names=combine_alias_output_names(*exprs),
-            returns_scalar=False,
+            expr_kind=n_ary_operation_expr_kind(*exprs),
             backend_version=self._backend_version,
             version=self._version,
         )
@@ -182,7 +184,7 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
             function_name="max_horizontal",
             evaluate_output_names=combine_evaluate_output_names(*exprs),
             alias_output_names=combine_alias_output_names(*exprs),
-            returns_scalar=False,
+            expr_kind=n_ary_operation_expr_kind(*exprs),
             backend_version=self._backend_version,
             version=self._version,
         )
@@ -197,7 +199,7 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
             function_name="min_horizontal",
             evaluate_output_names=combine_evaluate_output_names(*exprs),
             alias_output_names=combine_alias_output_names(*exprs),
-            returns_scalar=False,
+            expr_kind=n_ary_operation_expr_kind(*exprs),
             backend_version=self._backend_version,
             version=self._version,
         )
@@ -289,7 +291,7 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
             function_name="concat_str",
             evaluate_output_names=combine_evaluate_output_names(*exprs),
             alias_output_names=combine_alias_output_names(*exprs),
-            returns_scalar=False,
+            expr_kind=n_ary_operation_expr_kind(*exprs),
             backend_version=self._backend_version,
             version=self._version,
         )
@@ -298,7 +300,10 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
         plx = self.__class__(backend_version=self._backend_version, version=self._version)
         condition = plx.all_horizontal(*predicates)
         return SparkLikeWhen(
-            condition, self._backend_version, returns_scalar=False, version=self._version
+            condition,
+            self._backend_version,
+            expr_kind=ExprKind.TRANSFORM,
+            version=self._version,
         )
 
 
@@ -310,14 +315,14 @@ class SparkLikeWhen:
         then_value: Any | None = None,
         otherwise_value: Any | None = None,
         *,
-        returns_scalar: bool,
+        expr_kind: ExprKind,
         version: Version,
     ) -> None:
         self._backend_version = backend_version
         self._condition = condition
         self._then_value = then_value
         self._otherwise_value = otherwise_value
-        self._returns_scalar = returns_scalar
+        self._expr_kind = expr_kind
         self._version = version
 
     def __call__(self: Self, df: SparkLikeLazyFrame) -> list[Column]:
@@ -347,7 +352,7 @@ class SparkLikeWhen:
                 value, "_evaluate_output_names", lambda _df: ["literal"]
             ),
             alias_output_names=getattr(value, "_alias_output_names", None),
-            returns_scalar=self._returns_scalar,
+            expr_kind=self._expr_kind,
             backend_version=self._backend_version,
             version=self._version,
         )
@@ -361,7 +366,7 @@ class SparkLikeThen(SparkLikeExpr):
         function_name: str,
         evaluate_output_names: Callable[[SparkLikeLazyFrame], Sequence[str]],
         alias_output_names: Callable[[Sequence[str]], Sequence[str]] | None,
-        returns_scalar: bool,
+        expr_kind: ExprKind,
         backend_version: tuple[int, ...],
         version: Version,
     ) -> None:
@@ -371,7 +376,7 @@ class SparkLikeThen(SparkLikeExpr):
         self._function_name = function_name
         self._evaluate_output_names = evaluate_output_names
         self._alias_output_names = alias_output_names
-        self._returns_scalar = returns_scalar
+        self._expr_kind = expr_kind
 
     def otherwise(self: Self, value: SparkLikeExpr | Any) -> SparkLikeExpr:
         # type ignore because we are setting the `_call` attribute to a
diff --git a/narwhals/_spark_like/selectors.py b/narwhals/_spark_like/selectors.py
index b9771bcd..2e33b7bc 100644
--- a/narwhals/_spark_like/selectors.py
+++ b/narwhals/_spark_like/selectors.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import re
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Sequence
@@ -7,6 +8,7 @@ from typing import Sequence
 from pyspark.sql import functions as F  # noqa: N812
 
 from narwhals._spark_like.expr import SparkLikeExpr
+from narwhals._spark_like.utils import ExprKind
 from narwhals.utils import import_dtypes_module
 
 if TYPE_CHECKING:
@@ -32,6 +34,23 @@ class SparkLikeSelectorNamespace:
         def evalute_output_names(df: SparkLikeLazyFrame) -> Sequence[str]:
             return [col for col in df.columns if df.schema[col] in dtypes]
 
+        return SparkLikeSelector(
+            func,
+            function_name="selector",
+            evaluate_output_names=evalute_output_names,
+            alias_output_names=None,
+            backend_version=self._backend_version,
+            expr_kind=ExprKind.TRANSFORM,
+            version=self._version,
+        )
+
+    def matches(self: Self, pattern: str) -> SparkLikeSelector:
+        def func(df: SparkLikeLazyFrame) -> list[Column]:
+            return [F.col(col) for col in df.columns if re.search(pattern, col)]
+
+        def evalute_output_names(df: SparkLikeLazyFrame) -> Sequence[str]:
+            return [col for col in df.columns if re.search(pattern, col)]
+
         return SparkLikeSelector(
             func,
             function_name="selector",
@@ -83,7 +102,7 @@ class SparkLikeSelectorNamespace:
             evaluate_output_names=lambda df: df.columns,
             alias_output_names=None,
             backend_version=self._backend_version,
-            returns_scalar=False,
+            expr_kind=ExprKind.TRANSFORM,
             version=self._version,
         )
 
@@ -99,7 +118,7 @@ class SparkLikeSelector(SparkLikeExpr):
             evaluate_output_names=self._evaluate_output_names,
             alias_output_names=self._alias_output_names,
             backend_version=self._backend_version,
-            returns_scalar=self._returns_scalar,
+            expr_kind=self._expr_kind,
             version=self._version,
         )
 
@@ -123,7 +142,7 @@ class SparkLikeSelector(SparkLikeExpr):
                 evaluate_output_names=evaluate_output_names,
                 alias_output_names=None,
                 backend_version=self._backend_version,
-                returns_scalar=self._returns_scalar,
+                expr_kind=self._expr_kind,
                 version=self._version,
             )
         else:
@@ -153,7 +172,7 @@ class SparkLikeSelector(SparkLikeExpr):
                 evaluate_output_names=evaluate_output_names,
                 alias_output_names=None,
                 backend_version=self._backend_version,
-                returns_scalar=self._returns_scalar,
+                expr_kind=self._expr_kind,
                 version=self._version,
             )
         else:
@@ -179,7 +198,7 @@ class SparkLikeSelector(SparkLikeExpr):
                 evaluate_output_names=evaluate_output_names,
                 alias_output_names=None,
                 backend_version=self._backend_version,
-                returns_scalar=self._returns_scalar,
+                expr_kind=self._expr_kind,
                 version=self._version,
             )
         else:
diff --git a/narwhals/_spark_like/utils.py b/narwhals/_spark_like/utils.py
index 5eb9b304..a29d96ad 100644
--- a/narwhals/_spark_like/utils.py
+++ b/narwhals/_spark_like/utils.py
@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from enum import Enum
+from enum import auto
 from functools import lru_cache
 from typing import TYPE_CHECKING
 from typing import Any
@@ -22,6 +24,21 @@ if TYPE_CHECKING:
     from narwhals.utils import Version
 
 
+class ExprKind(Enum):
+    """Describe which kind of expression we are dealing with.
+
+    Composition rule is:
+    - LITERAL vs LITERAL -> LITERAL
+    - TRANSFORM vs anything -> TRANSFORM
+    - anything vs TRANSFORM -> TRANSFORM
+    - all remaining cases -> AGGREGATION
+    """
+
+    LITERAL = auto()  # e.g. nw.lit(1)
+    AGGREGATION = auto()  # e.g. nw.col('a').mean()
+    TRANSFORM = auto()  # e.g. nw.col('a').round()
+
+
 @lru_cache(maxsize=16)
 def native_to_narwhals_dtype(
     dtype: pyspark_types.DataType,
@@ -109,13 +126,10 @@ def narwhals_to_native_dtype(
 
 def parse_exprs_and_named_exprs(
     df: SparkLikeLazyFrame, /, *exprs: SparkLikeExpr, **named_exprs: SparkLikeExpr
-) -> tuple[dict[str, Column], list[bool]]:
+) -> tuple[dict[str, Column], list[ExprKind]]:
     native_results: dict[str, list[Column]] = {}
 
-    # `returns_scalar` keeps track if an expression returns a scalar and is not lit.
-    # Notice that lit is quite special case, since it gets broadcasted by pyspark
-    # without the need of adding `.over(Window.partitionBy(F.lit(1)))`
-    returns_scalar: list[bool] = []
+    expr_kinds: list[ExprKind] = []
     for expr in exprs:
         native_series_list = expr._call(df)
         output_names = expr._evaluate_output_names(df)
@@ -125,28 +139,19 @@ def parse_exprs_and_named_exprs(
             msg = f"Internal error: got output names {output_names}, but only got {len(native_series_list)} results"
             raise AssertionError(msg)
         native_results.update(zip(output_names, native_series_list))
-        returns_scalar.extend(
-            [
-                expr._returns_scalar
-                and expr._function_name.split("->", maxsplit=1)[0] != "lit"
-            ]
-            * len(output_names)
-        )
+        expr_kinds.extend([expr._expr_kind] * len(output_names))
     for col_alias, expr in named_exprs.items():
         native_series_list = expr._call(df)
         if len(native_series_list) != 1:  # pragma: no cover
             msg = "Named expressions must return a single column"
             raise ValueError(msg)
         native_results[col_alias] = native_series_list[0]
-        returns_scalar.append(
-            expr._returns_scalar
-            and expr._function_name.split("->", maxsplit=1)[0] != "lit"
-        )
+        expr_kinds.append(expr._expr_kind)
 
-    return native_results, returns_scalar
+    return native_results, expr_kinds
 
 
-def maybe_evaluate(df: SparkLikeLazyFrame, obj: Any, *, returns_scalar: bool) -> Column:
+def maybe_evaluate(df: SparkLikeLazyFrame, obj: Any, *, expr_kind: ExprKind) -> Column:
     from narwhals._spark_like.expr import SparkLikeExpr
 
     if isinstance(obj, SparkLikeExpr):
@@ -155,11 +160,7 @@ def maybe_evaluate(df: SparkLikeLazyFrame, obj: Any, *, returns_scalar: bool) ->
             msg = "Multi-output expressions (e.g. `nw.all()` or `nw.col('a', 'b')`) not supported in this context"
             raise NotImplementedError(msg)
         column_result = column_results[0]
-        if (
-            obj._returns_scalar
-            and obj._function_name.split("->", maxsplit=1)[0] != "lit"
-            and not returns_scalar
-        ):
+        if obj._expr_kind is ExprKind.AGGREGATION and expr_kind is ExprKind.TRANSFORM:
             # Returns scalar, but overall expression doesn't.
             # Let PySpark do its broadcasting
             return column_result.over(Window.partitionBy(F.lit(1)))
@@ -195,8 +196,13 @@ def _var(_input: Column | str, ddof: int, np_version: tuple[int, ...]) -> Column
     return var(input_col, ddof=ddof)
 
 
-def binary_operation_returns_scalar(lhs: SparkLikeExpr, rhs: SparkLikeExpr | Any) -> bool:
-    # If `rhs` is a SparkLikeExpr, we look at `_returns_scalar`. If it isn't,
-    # it means that it was a scalar (e.g. nw.col('a') + 1), and so we default
-    # to `True`.
-    return lhs._returns_scalar and getattr(rhs, "_returns_scalar", True)
+def n_ary_operation_expr_kind(*args: SparkLikeExpr | Any) -> ExprKind:
+    if all(
+        getattr(arg, "_expr_kind", ExprKind.LITERAL) is ExprKind.LITERAL for arg in args
+    ):
+        return ExprKind.LITERAL
+    if any(
+        getattr(arg, "_expr_kind", ExprKind.LITERAL) is ExprKind.TRANSFORM for arg in args
+    ):
+        return ExprKind.TRANSFORM
+    return ExprKind.AGGREGATION
diff --git a/narwhals/selectors.py b/narwhals/selectors.py
index dabf6f83..6228030b 100644
--- a/narwhals/selectors.py
+++ b/narwhals/selectors.py
@@ -87,6 +87,62 @@ def by_dtype(*dtypes: Any) -> Expr:
     )
 
 
+def matches(pattern: str) -> Expr:
+    """Select all columns that match the given regex pattern.
+
+    Arguments:
+        pattern: A valid regular expression pattern.
+
+    Returns:
+        A new expression.
+
+    Examples:
+        >>> import narwhals as nw
+        >>> import narwhals.selectors as ncs
+        >>> import pandas as pd
+        >>> import polars as pl
+        >>>
+        >>> data = {
+        ...     "foo": ["x", "y"],
+        ...     "bar": [123, 456],
+        ...     "baz": [2.0, 5.5],
+        ...     "zap": [0, 1],
+        ... }
+        >>> df_pd = pd.DataFrame(data)
+        >>> df_pl = pl.DataFrame(data)
+
+        Let's define a dataframe-agnostic function to select column names
+        containing an 'a', preceded by a character that is not 'z':
+
+        >>> @nw.narwhalify
+        ... def func(df):
+        ...     return df.select(ncs.matches("[^z]a"))
+
+        We can then pass either pandas or Polars dataframes:
+
+        >>> func(df_pd)
+           bar  baz
+        0  123  2.0
+        1  456  5.5
+        >>> func(df_pl)
+        shape: (2, 2)
+        ┌─────┬─────┐
+        │ bar ┆ baz │
+        │ --- ┆ --- │
+        │ i64 ┆ f64 │
+        ╞═════╪═════╡
+        │ 123 ┆ 2.0 │
+        │ 456 ┆ 5.5 │
+        └─────┴─────┘
+    """
+    return Selector(
+        lambda plx: plx.selectors.matches(pattern),
+        is_order_dependent=False,
+        changes_length=False,
+        aggregates=False,
+    )
+
+
 def numeric() -> Expr:
     """Select numeric columns.
 
diff --git a/narwhals/stable/v1/selectors.py b/narwhals/stable/v1/selectors.py
index 0d82484e..f938dfa3 100644
--- a/narwhals/stable/v1/selectors.py
+++ b/narwhals/stable/v1/selectors.py
@@ -4,6 +4,7 @@ from narwhals.selectors import all
 from narwhals.selectors import boolean
 from narwhals.selectors import by_dtype
 from narwhals.selectors import categorical
+from narwhals.selectors import matches
 from narwhals.selectors import numeric
 from narwhals.selectors import string
 
@@ -12,6 +13,7 @@ __all__ = [
     "boolean",
     "by_dtype",
     "categorical",
+    "matches",
     "numeric",
     "string",
 ]
diff --git a/pyproject.toml b/pyproject.toml
index ffca1096..fd2860b7 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -18,6 +18,7 @@ classifiers = [
   "License :: OSI Approved :: MIT License",
   "Programming Language :: Python",
   "Operating System :: OS Independent",
+  "Typing :: Typed",
 ]
 
 [project.optional-dependencies]


Test patch:
diff --git a/tests/selectors_test.py b/tests/selectors_test.py
index 9e16da5a..20a75d22 100644
--- a/tests/selectors_test.py
+++ b/tests/selectors_test.py
@@ -10,6 +10,7 @@ from narwhals.stable.v1.selectors import all
 from narwhals.stable.v1.selectors import boolean
 from narwhals.stable.v1.selectors import by_dtype
 from narwhals.stable.v1.selectors import categorical
+from narwhals.stable.v1.selectors import matches
 from narwhals.stable.v1.selectors import numeric
 from narwhals.stable.v1.selectors import string
 from tests.utils import POLARS_VERSION
@@ -25,6 +26,13 @@ data = {
     "d": [True, False, True],
 }
 
+data_regex = {
+    "foo": ["x", "y"],
+    "bar": [123, 456],
+    "baz": [2.0, 5.5],
+    "zap": [0, 1],
+}
+
 
 def test_selectors(constructor: Constructor) -> None:
     df = nw.from_native(constructor(data))
@@ -33,6 +41,16 @@ def test_selectors(constructor: Constructor) -> None:
     assert_equal_data(result, expected)
 
 
+def test_matches(constructor: Constructor) -> None:
+    df = nw.from_native(constructor(data_regex))
+    result = df.select(matches("[^z]a") + 1)
+    expected = {
+        "bar": [124, 457],
+        "baz": [3.0, 6.5],
+    }
+    assert_equal_data(result, expected)
+
+
 def test_numeric(constructor: Constructor) -> None:
     df = nw.from_native(constructor(data))
     result = df.select(numeric() + 1)