Repository: narwhals-dev/narwhals
Pull request #2078 (2025-02-23 19:22:30)

chore: add `has_open_windows` to `ExprMetadata` instead of `is_order_dependent`
One step closer to supporting `lf.select(nw.col('a').cum_sum('b', order_by='c'))`. Getting there!

This PR does not (yet) have any user-facing impact. But...we're not that far off from being able to make changes which are!

The gist of the change is to distinguish between:
- transform: preserves length
- window: preserves length and the last node is a window function (e.g. `cum_sum`, `rolling_mean`, `shift`, ...)

Then:
- and open window is a window function not immediately followed by `over` with `order_by` specified
- calling `over` immediately after a window function reduces `n_open_windows` by `1`
- by the time we evaluate an expression, then:
  - in the lazy API, there must be 0 open windows
  - in the eager API, there are no restrictions, as physical row order is always defined

<!--
# 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: ooh nice, looks like the tubular test suite helped find a bug 🙌 

Patch:
diff --git a/narwhals/_arrow/expr.py b/narwhals/_arrow/expr.py
index 669e535d..677a6e9f 100644
--- a/narwhals/_arrow/expr.py
+++ b/narwhals/_arrow/expr.py
@@ -14,6 +14,7 @@ from narwhals._arrow.expr_str import ArrowExprStringNamespace
 from narwhals._arrow.series import ArrowSeries
 from narwhals._expression_parsing import ExprKind
 from narwhals._expression_parsing import evaluate_output_names_and_aliases
+from narwhals._expression_parsing import is_scalar_like
 from narwhals._expression_parsing import reuse_series_implementation
 from narwhals.dependencies import get_numpy
 from narwhals.dependencies import is_numpy_array
@@ -414,10 +415,8 @@ class ArrowExpr(CompliantExpr["ArrowDataFrame", ArrowSeries]):
         )
 
     def over(self: Self, keys: list[str], kind: ExprKind) -> Self:
-        if kind is ExprKind.TRANSFORM:
-            msg = (
-                "Elementwise operations in `over` context are not supported for PyArrow."
-            )
+        if not is_scalar_like(kind):
+            msg = "Only aggregation or literal operations are supported in `over` context for PyArrow."
             raise NotImplementedError(msg)
 
         def func(df: ArrowDataFrame) -> list[ArrowSeries]:
diff --git a/narwhals/_dask/dataframe.py b/narwhals/_dask/dataframe.py
index 935624c4..91a2aead 100644
--- a/narwhals/_dask/dataframe.py
+++ b/narwhals/_dask/dataframe.py
@@ -43,7 +43,7 @@ class DaskLazyFrame(CompliantLazyFrame):
         version: Version,
         validate_column_names: bool,
     ) -> None:
-        self._native_frame = native_dataframe
+        self._native_frame: dd.DataFrame = native_dataframe
         self._backend_version = backend_version
         self._implementation = Implementation.DASK
         self._version = version
@@ -138,7 +138,7 @@ class DaskLazyFrame(CompliantLazyFrame):
 
     @property
     def columns(self: Self) -> list[str]:
-        return self._native_frame.columns.tolist()  # type: ignore[no-any-return]
+        return self._native_frame.columns.tolist()
 
     def filter(self: Self, predicate: DaskExpr) -> Self:
         # `[0]` is safe as the predicate's expression only returns a single column
@@ -426,7 +426,7 @@ class DaskLazyFrame(CompliantLazyFrame):
         return (
             self.with_row_index(row_index_token)
             .filter(
-                (plx.col(row_index_token) >= offset)  # type: ignore[operator]
+                (plx.col(row_index_token) >= offset)
                 & ((plx.col(row_index_token) - offset) % n == 0)
             )
             .drop([row_index_token], strict=False)
diff --git a/narwhals/_dask/expr.py b/narwhals/_dask/expr.py
index 538529e5..22fb068d 100644
--- a/narwhals/_dask/expr.py
+++ b/narwhals/_dask/expr.py
@@ -35,7 +35,7 @@ if TYPE_CHECKING:
     from narwhals.utils import Version
 
 
-class DaskExpr(CompliantExpr["DaskLazyFrame", "dx.Series"]):
+class DaskExpr(CompliantExpr["DaskLazyFrame", "dx.Series"]):  # pyright: ignore[reportInvalidTypeArguments] (#2044)
     _implementation: Implementation = Implementation.DASK
 
     def __init__(
@@ -255,7 +255,7 @@ class DaskExpr(CompliantExpr["DaskLazyFrame", "dx.Series"]):
             lambda _input, other: _input.__ne__(other), "__ne__", other=other
         )
 
-    def __ge__(self: Self, other: DaskExpr) -> Self:
+    def __ge__(self: Self, other: DaskExpr | Any) -> Self:
         return self._from_call(
             lambda _input, other: _input.__ge__(other), "__ge__", other=other
         )
@@ -275,7 +275,7 @@ class DaskExpr(CompliantExpr["DaskLazyFrame", "dx.Series"]):
             lambda _input, other: _input.__lt__(other), "__lt__", other=other
         )
 
-    def __and__(self: Self, other: DaskExpr) -> Self:
+    def __and__(self: Self, other: DaskExpr | Any) -> Self:
         return self._from_call(
             lambda _input, other: _input.__and__(other), "__and__", other=other
         )
@@ -454,7 +454,7 @@ class DaskExpr(CompliantExpr["DaskLazyFrame", "dx.Series"]):
                 _input.dtype, self._version, self._implementation
             )
             if dtype.is_numeric():
-                return _input != _input  # noqa: PLR0124
+                return _input != _input  # pyright: ignore[reportReturnType] # noqa: PLR0124
             msg = f"`.is_nan` only supported for numeric dtypes and not {dtype}, did you mean `.is_null`?"
             raise InvalidOperationError(msg)
 
@@ -487,16 +487,11 @@ class DaskExpr(CompliantExpr["DaskLazyFrame", "dx.Series"]):
         def func(_input: dx.Series) -> dx.Series:
             _name = _input.name
             col_token = generate_temporary_column_name(n_bytes=8, columns=[_name])
-            _input = add_row_index(
-                _input.to_frame(),
-                col_token,
-                backend_version=self._backend_version,
-                implementation=self._implementation,
+            frame = add_row_index(
+                _input.to_frame(), col_token, self._backend_version, self._implementation
             )
-            first_distinct_index = _input.groupby(_name).agg({col_token: "min"})[
-                col_token
-            ]
-            return _input[col_token].isin(first_distinct_index)
+            first_distinct_index = frame.groupby(_name).agg({col_token: "min"})[col_token]
+            return frame[col_token].isin(first_distinct_index)
 
         return self._from_call(func, "is_first_distinct")
 
@@ -504,14 +499,11 @@ class DaskExpr(CompliantExpr["DaskLazyFrame", "dx.Series"]):
         def func(_input: dx.Series) -> dx.Series:
             _name = _input.name
             col_token = generate_temporary_column_name(n_bytes=8, columns=[_name])
-            _input = add_row_index(
-                _input.to_frame(),
-                col_token,
-                backend_version=self._backend_version,
-                implementation=self._implementation,
+            frame = add_row_index(
+                _input.to_frame(), col_token, self._backend_version, self._implementation
             )
-            last_distinct_index = _input.groupby(_name).agg({col_token: "max"})[col_token]
-            return _input[col_token].isin(last_distinct_index)
+            last_distinct_index = frame.groupby(_name).agg({col_token: "max"})[col_token]
+            return frame[col_token].isin(last_distinct_index)
 
         return self._from_call(func, "is_last_distinct")
 
diff --git a/narwhals/_dask/expr_dt.py b/narwhals/_dask/expr_dt.py
index c569d7dc..b7355dad 100644
--- a/narwhals/_dask/expr_dt.py
+++ b/narwhals/_dask/expr_dt.py
@@ -96,9 +96,9 @@ class DaskExprDateTimeNamespace:
                 s.dtype, self._compliant_expr._version, Implementation.DASK
             )
             if dtype.time_zone is None:  # type: ignore[attr-defined]
-                return s.dt.tz_localize("UTC").dt.tz_convert(time_zone)
+                return s.dt.tz_localize("UTC").dt.tz_convert(time_zone)  # pyright: ignore[reportAttributeAccessIssue]
             else:
-                return s.dt.tz_convert(time_zone)
+                return s.dt.tz_convert(time_zone)  # pyright: ignore[reportAttributeAccessIssue]
 
         return self._compliant_expr._from_call(func, "tz_convert", time_zone=time_zone)
 
@@ -125,7 +125,7 @@ class DaskExprDateTimeNamespace:
             else:
                 msg = "Input should be either of Date or Datetime type"
                 raise TypeError(msg)
-            return result.where(~mask_na)
+            return result.where(~mask_na)  # pyright: ignore[reportReturnType]
 
         return self._compliant_expr._from_call(func, "datetime", time_unit=time_unit)
 
diff --git a/narwhals/_dask/group_by.py b/narwhals/_dask/group_by.py
index c0322679..6a972c6a 100644
--- a/narwhals/_dask/group_by.py
+++ b/narwhals/_dask/group_by.py
@@ -26,7 +26,6 @@ if TYPE_CHECKING:
 
     from narwhals._dask.dataframe import DaskLazyFrame
     from narwhals._dask.expr import DaskExpr
-    from narwhals.typing import CompliantExpr
 
     PandasSeriesGroupBy: TypeAlias = _PandasSeriesGroupBy[Any, Any]
     _AggFn: TypeAlias = Callable[..., Any]
@@ -73,7 +72,7 @@ class DaskLazyGroupBy:
     def __init__(
         self: Self, df: DaskLazyFrame, keys: list[str], *, drop_null_keys: bool
     ) -> None:
-        self._df = df
+        self._df: DaskLazyFrame = df
         self._keys = keys
         self._grouped = self._df._native_frame.groupby(
             list(self._keys),
@@ -93,11 +92,11 @@ class DaskLazyGroupBy:
             self._from_native_frame,
         )
 
-    def _from_native_frame(self: Self, df: DaskLazyFrame) -> DaskLazyFrame:
+    def _from_native_frame(self: Self, df: dd.DataFrame) -> DaskLazyFrame:
         from narwhals._dask.dataframe import DaskLazyFrame
 
         return DaskLazyFrame(
-            df,  # pyright: ignore[reportArgumentType]
+            df,
             backend_version=self._df._backend_version,
             version=self._df._version,
             validate_column_names=True,
@@ -107,7 +106,7 @@ class DaskLazyGroupBy:
 def agg_dask(
     df: DaskLazyFrame,
     grouped: Any,
-    exprs: Sequence[CompliantExpr[DaskLazyFrame, dx.Series]],
+    exprs: Sequence[DaskExpr],
     keys: list[str],
     from_dataframe: Callable[[Any], DaskLazyFrame],
 ) -> DaskLazyFrame:
@@ -148,7 +147,7 @@ def agg_dask(
             agg_function = POLARS_TO_DASK_AGGREGATIONS.get(function_name, function_name)
             # deal with n_unique case in a "lazy" mode to not depend on dask globally
             agg_function = (
-                agg_function(**expr._call_kwargs)  # type: ignore[attr-defined]
+                agg_function(**expr._call_kwargs)
                 if callable(agg_function)
                 else agg_function
             )
diff --git a/narwhals/_dask/namespace.py b/narwhals/_dask/namespace.py
index 4232ae21..8522dc86 100644
--- a/narwhals/_dask/namespace.py
+++ b/narwhals/_dask/namespace.py
@@ -8,7 +8,6 @@ from typing import Callable
 from typing import Iterable
 from typing import Literal
 from typing import Sequence
-from typing import cast
 
 import dask.dataframe as dd
 import pandas as pd
@@ -24,7 +23,6 @@ from narwhals._dask.utils import validate_comparand
 from narwhals._expression_parsing import combine_alias_output_names
 from narwhals._expression_parsing import combine_evaluate_output_names
 from narwhals.typing import CompliantNamespace
-from narwhals.utils import is_compliant_expr
 
 if TYPE_CHECKING:
     from typing_extensions import Self
@@ -38,7 +36,7 @@ if TYPE_CHECKING:
         import dask_expr as dx
 
 
-class DaskNamespace(CompliantNamespace[DaskLazyFrame, "dx.Series"]):
+class DaskNamespace(CompliantNamespace[DaskLazyFrame, "dx.Series"]):  # pyright: ignore[reportInvalidTypeArguments] (#2044)
     @property
     def selectors(self: Self) -> DaskSelectorNamespace:
         return DaskSelectorNamespace(self)
@@ -347,17 +345,16 @@ class DaskWhen:
         version: Version,
     ) -> None:
         self._backend_version = backend_version
-        self._condition = condition
-        self._then_value = then_value
-        self._otherwise_value = otherwise_value
+        self._condition: DaskExpr = condition
+        self._then_value: DaskExpr | Any = then_value
+        self._otherwise_value: DaskExpr | Any = otherwise_value
         self._version = version
 
     def __call__(self: Self, df: DaskLazyFrame) -> Sequence[dx.Series]:
         condition = self._condition(df)[0]
-        condition = cast("dx.Series", condition)
 
-        if is_compliant_expr(self._then_value):
-            then_value: dx.Series | object = self._then_value(df)[0]
+        if isinstance(self._then_value, DaskExpr):
+            then_value = self._then_value(df)[0]
         else:
             then_value = self._then_value
         (then_series,) = align_series_full_broadcast(df, then_value)
@@ -366,13 +363,13 @@ class DaskWhen:
         if self._otherwise_value is None:
             return [then_series.where(condition)]
 
-        if is_compliant_expr(self._otherwise_value):
-            otherwise_value: dx.Series | object = self._otherwise_value(df)[0]
+        if isinstance(self._otherwise_value, DaskExpr):
+            otherwise_value = self._otherwise_value(df)[0]
         else:
             otherwise_value = self._otherwise_value
         (otherwise_series,) = align_series_full_broadcast(df, otherwise_value)
         validate_comparand(condition, otherwise_series)
-        return [then_series.where(condition, otherwise_series)]
+        return [then_series.where(condition, otherwise_series)]  # pyright: ignore[reportArgumentType]
 
     def then(self: Self, value: DaskExpr | Any) -> DaskThen:
         self._then_value = value
@@ -405,7 +402,7 @@ class DaskThen(DaskExpr):
     ) -> None:
         self._backend_version = backend_version
         self._version = version
-        self._call = call
+        self._call: DaskWhen = call
         self._depth = depth
         self._function_name = function_name
         self._evaluate_output_names = evaluate_output_names
@@ -413,9 +410,6 @@ class DaskThen(DaskExpr):
         self._call_kwargs = call_kwargs or {}
 
     def otherwise(self: Self, value: DaskExpr | Any) -> DaskExpr:
-        # type ignore because we are setting the `_call` attribute to a
-        # callable object of type `DaskWhen`, base class has the attribute as
-        # only a `Callable`
-        self._call._otherwise_value = value  # type: ignore[attr-defined]
+        self._call._otherwise_value = value
         self._function_name = "whenotherwise"
         return self
diff --git a/narwhals/_dask/utils.py b/narwhals/_dask/utils.py
index 5c1c4c07..a98241ec 100644
--- a/narwhals/_dask/utils.py
+++ b/narwhals/_dask/utils.py
@@ -62,7 +62,7 @@ def align_series_full_broadcast(
     return [
         s if isinstance(s, dx.Series) else df._native_frame.assign(_tmp=s)["_tmp"]
         for s in series
-    ]
+    ]  # pyright: ignore[reportReturnType]
 
 
 def add_row_index(
@@ -155,8 +155,8 @@ def narwhals_to_native_dtype(dtype: DType | type[DType], version: Version) -> An
 
 
 def name_preserving_sum(s1: dx.Series, s2: dx.Series) -> dx.Series:
-    return (s1 + s2).rename(s1.name)
+    return (s1 + s2).rename(s1.name)  # pyright: ignore[reportOperatorIssue]
 
 
 def name_preserving_div(s1: dx.Series, s2: dx.Series) -> dx.Series:
-    return (s1 / s2).rename(s1.name)
+    return (s1 / s2).rename(s1.name)  # pyright: ignore[reportOperatorIssue]
diff --git a/narwhals/_expression_parsing.py b/narwhals/_expression_parsing.py
index 148259d8..b567af23 100644
--- a/narwhals/_expression_parsing.py
+++ b/narwhals/_expression_parsing.py
@@ -9,6 +9,7 @@ from itertools import chain
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Callable
+from typing import Literal
 from typing import Sequence
 from typing import TypeVar
 from typing import overload
@@ -331,8 +332,8 @@ class ExprKind(Enum):
     Commutative composition rules are:
     - LITERAL vs LITERAL -> LITERAL
     - CHANGES_LENGTH vs (LITERAL | AGGREGATION) -> CHANGES_LENGTH
-    - CHANGES_LENGTH vs (CHANGES_LENGTH | TRANSFORM) -> raise
-    - TRANSFORM vs (LITERAL | AGGREGATION) -> TRANSFORM
+    - CHANGES_LENGTH vs (CHANGES_LENGTH | TRANSFORM | WINDOW) -> raise
+    - (TRANSFORM | WINDOW) vs (LITERAL | AGGREGATION) -> TRANSFORM
     - AGGREGATION vs (LITERAL | AGGREGATION) -> AGGREGATION
     """
 
@@ -343,18 +344,49 @@ class ExprKind(Enum):
     """e.g. `nw.col('a').mean()`"""
 
     TRANSFORM = auto()
-    """length-preserving, e.g. `nw.col('a').round()`"""
+    """preserves length, e.g. `nw.col('a').round()`"""
+
+    WINDOW = auto()
+    """transform in which last node is order-dependent
+
+    examples:
+    - `nw.col('a').cum_sum()`
+    - `(nw.col('a')+1).cum_sum()`
+
+    non-examples:
+    - `nw.col('a').cum_sum()+1`
+    - `nw.col('a').cum_sum().mean()`
+    """
 
     CHANGES_LENGTH = auto()
     """e.g. `nw.col('a').drop_nulls()`"""
 
+    def preserves_length(self) -> bool:
+        return self in {ExprKind.TRANSFORM, ExprKind.WINDOW}
+
+    def is_window(self) -> bool:
+        return self is ExprKind.WINDOW
+
+    def is_changes_length(self) -> bool:
+        return self is ExprKind.CHANGES_LENGTH
+
+    def is_scalar_like(self) -> bool:
+        return is_scalar_like(self)
+
+
+def is_scalar_like(
+    kind: ExprKind,
+) -> TypeIs[Literal[ExprKind.AGGREGATION, ExprKind.LITERAL]]:
+    # Like ExprKind.is_scalar_like, but uses TypeIs for better type checking.
+    return kind in {ExprKind.AGGREGATION, ExprKind.LITERAL}
+
 
 class ExprMetadata:
-    __slots__ = ("_kind", "_order_dependent")
+    __slots__ = ("_kind", "_n_open_windows")
 
-    def __init__(self, kind: ExprKind, /, *, order_dependent: bool) -> None:
+    def __init__(self, kind: ExprKind, /, *, n_open_windows: int) -> None:
         self._kind: ExprKind = kind
-        self._order_dependent: bool = order_dependent
+        self._n_open_windows = n_open_windows
 
     def __init_subclass__(cls, /, *args: Any, **kwds: Any) -> Never:  # pragma: no cover
         msg = f"Cannot subclass {cls.__name__!r}"
@@ -364,50 +396,42 @@ class ExprMetadata:
     def kind(self) -> ExprKind:
         return self._kind
 
-    def is_order_dependent(self) -> bool:
-        return self._order_dependent
-
-    def is_transform(self) -> bool:
-        return self.kind is ExprKind.TRANSFORM
-
-    def is_aggregation_or_literal(self) -> bool:
-        return self.kind in {ExprKind.AGGREGATION, ExprKind.LITERAL}
-
-    def is_changes_length(self) -> bool:
-        return self.kind is ExprKind.CHANGES_LENGTH
+    @property
+    def n_open_windows(self) -> int:
+        return self._n_open_windows
 
     def with_kind(self, kind: ExprKind, /) -> ExprMetadata:
         """Change metadata kind, leaving all other attributes the same."""
-        return ExprMetadata(kind, order_dependent=self.is_order_dependent())
+        return ExprMetadata(kind, n_open_windows=self._n_open_windows)
 
-    def with_order_dependence(self) -> ExprMetadata:
-        """Set `order_dependent` to True, leaving all other attributes the same."""
-        return ExprMetadata(self.kind, order_dependent=True)
+    def with_extra_open_window(self) -> ExprMetadata:
+        """Increment `n_open_windows` leaving other attributes the same."""
+        return ExprMetadata(self.kind, n_open_windows=self._n_open_windows + 1)
 
-    def with_kind_and_order_dependence(self, kind: ExprKind, /) -> ExprMetadata:
-        """Change kind and set `order_dependent` to True."""
-        return ExprMetadata(kind, order_dependent=True)
+    def with_kind_and_extra_open_window(self, kind: ExprKind, /) -> ExprMetadata:
+        """Change metadata kind and increment `n_open_windows`."""
+        return ExprMetadata(kind, n_open_windows=self._n_open_windows + 1)
 
     @staticmethod
     def selector() -> ExprMetadata:
-        return ExprMetadata(ExprKind.TRANSFORM, order_dependent=False)
+        return ExprMetadata(ExprKind.TRANSFORM, n_open_windows=0)
 
 
 def combine_metadata(*args: IntoExpr | object | None, str_as_lit: bool) -> ExprMetadata:
     # Combine metadata from `args`.
 
     n_changes_length = 0
-    has_transforms = False
+    has_transforms_or_windows = False
     has_aggregations = False
     has_literals = False
-    result_is_order_dependent = False
+    result_n_open_windows = 0
 
     for arg in args:
         if isinstance(arg, str) and not str_as_lit:
-            has_transforms = True
+            has_transforms_or_windows = True
         elif is_expr(arg):
-            if arg._metadata.is_order_dependent():
-                result_is_order_dependent = True
+            if arg._metadata.n_open_windows:
+                result_n_open_windows += 1
             kind = arg._metadata.kind
             if kind is ExprKind.AGGREGATION:
                 has_aggregations = True
@@ -415,54 +439,55 @@ def combine_metadata(*args: IntoExpr | object | None, str_as_lit: bool) -> ExprM
                 has_literals = True
             elif kind is ExprKind.CHANGES_LENGTH:
                 n_changes_length += 1
-            elif kind is ExprKind.TRANSFORM:
-                has_transforms = True
+            elif kind.preserves_length():
+                has_transforms_or_windows = True
             else:  # pragma: no cover
                 msg = "unreachable code"
                 raise AssertionError(msg)
     if (
         has_literals
         and not has_aggregations
-        and not has_transforms
+        and not has_transforms_or_windows
         and not n_changes_length
     ):
         result_kind = ExprKind.LITERAL
     elif n_changes_length > 1:
         msg = "Length-changing expressions can only be used in isolation, or followed by an aggregation"
         raise LengthChangingExprError(msg)
-    elif n_changes_length and has_transforms:
+    elif n_changes_length and has_transforms_or_windows:
         msg = "Cannot combine length-changing expressions with length-preserving ones or aggregations"
         raise ShapeError(msg)
     elif n_changes_length:
         result_kind = ExprKind.CHANGES_LENGTH
-    elif has_transforms:
+    elif has_transforms_or_windows:
         result_kind = ExprKind.TRANSFORM
     else:
         result_kind = ExprKind.AGGREGATION
 
-    return ExprMetadata(result_kind, order_dependent=result_is_order_dependent)
+    return ExprMetadata(result_kind, n_open_windows=result_n_open_windows)
 
 
-def check_expressions_transform(*args: IntoExpr, function_name: str) -> None:
+def check_expressions_preserve_length(*args: IntoExpr, function_name: str) -> None:
     # Raise if any argument in `args` isn't length-preserving.
     # For Series input, we don't raise (yet), we let such checks happen later,
     # as this function works lazily and so can't evaluate lengths.
     from narwhals.series import Series
 
     if not all(
-        (is_expr(x) and x._metadata.is_transform()) or isinstance(x, (str, Series))
+        (is_expr(x) and x._metadata.kind.preserves_length())
+        or isinstance(x, (str, Series))
         for x in args
     ):
         msg = f"Expressions which aggregate or change length cannot be passed to '{function_name}'."
         raise ShapeError(msg)
 
 
-def all_exprs_are_aggs_or_literals(*args: IntoExpr, **kwargs: IntoExpr) -> bool:
+def all_exprs_are_scalar_like(*args: IntoExpr, **kwargs: IntoExpr) -> bool:
     # Raise if any argument in `args` isn't an aggregation or literal.
     # For Series input, we don't raise (yet), we let such checks happen later,
     # as this function works lazily and so can't evaluate lengths.
     exprs = chain(args, kwargs.values())
-    return all(is_expr(x) and x._metadata.is_aggregation_or_literal() for x in exprs)
+    return all(is_expr(x) and x._metadata.kind.is_scalar_like() for x in exprs)
 
 
 def infer_kind(obj: IntoExpr | _1DArray | object, *, str_as_lit: bool) -> ExprKind:
@@ -489,7 +514,7 @@ def apply_n_ary_operation(
     )
     kinds = [infer_kind(comparand, str_as_lit=str_as_lit) for comparand in comparands]
 
-    broadcast = any(kind is ExprKind.TRANSFORM for kind in kinds)
+    broadcast = any(kind.preserves_length() for kind in kinds)
     compliant_exprs = (
         compliant_expr.broadcast(kind)
         if broadcast
diff --git a/narwhals/_pandas_like/expr.py b/narwhals/_pandas_like/expr.py
index ce98c382..d2269b91 100644
--- a/narwhals/_pandas_like/expr.py
+++ b/narwhals/_pandas_like/expr.py
@@ -9,6 +9,7 @@ from typing import Sequence
 
 from narwhals._expression_parsing import ExprKind
 from narwhals._expression_parsing import evaluate_output_names_and_aliases
+from narwhals._expression_parsing import is_scalar_like
 from narwhals._expression_parsing import is_simple_aggregation
 from narwhals._expression_parsing import reuse_series_implementation
 from narwhals._pandas_like.expr_cat import PandasLikeExprCatNamespace
@@ -475,9 +476,9 @@ class PandasLikeExpr(CompliantExpr["PandasLikeDataFrame", PandasLikeSeries]):
                     )
                 )
                 return [result_frame[name] for name in aliases]
-        elif kind is ExprKind.TRANSFORM:
+        elif not is_scalar_like(kind):
             msg = (
-                "Elementwise operations are only supported in `over` context "
+                "Length-preserving operations are only supported in `over` context "
                 "for pandas if they are elementary "
                 "(e.g. `nw.col('a').cum_sum().over('b'))`)."
             )
diff --git a/narwhals/_spark_like/dataframe.py b/narwhals/_spark_like/dataframe.py
index 968ffaf1..6ee7a86b 100644
--- a/narwhals/_spark_like/dataframe.py
+++ b/narwhals/_spark_like/dataframe.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import warnings
+from importlib import import_module
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Literal
@@ -13,6 +14,7 @@ from narwhals.typing import CompliantDataFrame
 from narwhals.typing import CompliantLazyFrame
 from narwhals.utils import Implementation
 from narwhals.utils import check_column_exists
+from narwhals.utils import check_column_names_are_unique
 from narwhals.utils import find_stacklevel
 from narwhals.utils import import_dtypes_module
 from narwhals.utils import parse_columns_to_drop
@@ -23,6 +25,7 @@ if TYPE_CHECKING:
     from types import ModuleType
 
     import pyarrow as pa
+    from pyspark.sql import Column
     from pyspark.sql import DataFrame
     from typing_extensions import Self
 
@@ -41,7 +44,10 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
         backend_version: tuple[int, ...],
         version: Version,
         implementation: Implementation,
+        validate_column_names: bool,
     ) -> None:
+        if validate_column_names:
+            check_column_names_are_unique(native_dataframe.columns)
         self._native_frame = native_dataframe
         self._backend_version = backend_version
         self._implementation = implementation
@@ -51,9 +57,12 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
     @property
     def _F(self: Self) -> Any:  # noqa: N802
         if self._implementation is Implementation.SQLFRAME:
-            from sqlframe.duckdb import functions
+            from sqlframe.base.session import _BaseSession
+
+            return import_module(
+                f"sqlframe.{_BaseSession().execution_dialect_name}.functions"
+            )
 
-            return functions
         from pyspark.sql import functions
 
         return functions
@@ -61,9 +70,12 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
     @property
     def _native_dtypes(self: Self) -> Any:
         if self._implementation is Implementation.SQLFRAME:
-            from sqlframe.duckdb import types
+            from sqlframe.base.session import _BaseSession
+
+            return import_module(
+                f"sqlframe.{_BaseSession().execution_dialect_name}.types"
+            )
 
-            return types
         from pyspark.sql import types
 
         return types
@@ -71,13 +83,24 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
     @property
     def _Window(self: Self) -> Any:  # noqa: N802
         if self._implementation is Implementation.SQLFRAME:
-            from sqlframe.duckdb import Window
+            from sqlframe.base.session import _BaseSession
+
+            _window = import_module(
+                f"sqlframe.{_BaseSession().execution_dialect_name}.window"
+            )
+            return _window.Window
 
-            return Window
         from pyspark.sql import Window
 
         return Window
 
+    @property
+    def _session(self: Self) -> Any:
+        if self._implementation is Implementation.SQLFRAME:
+            return self._native_frame.session
+
+        return self._native_frame.sparkSession
+
     def __native_namespace__(self: Self) -> ModuleType:  # pragma: no cover
         return self._implementation.to_native_namespace()
 
@@ -99,14 +122,18 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
             backend_version=self._backend_version,
             version=version,
             implementation=self._implementation,
+            validate_column_names=False,
         )
 
-    def _from_native_frame(self: Self, df: DataFrame) -> Self:
+    def _from_native_frame(
+        self: Self, df: DataFrame, *, validate_column_names: bool = True
+    ) -> Self:
         return self.__class__(
             df,
             backend_version=self._backend_version,
             version=self._version,
             implementation=self._implementation,
+            validate_column_names=validate_column_names,
         )
 
     def _collect_to_arrow(self) -> pa.Table:
@@ -205,7 +232,9 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
         raise ValueError(msg)  # pragma: no cover
 
     def simple_select(self: Self, *column_names: str) -> Self:
-        return self._from_native_frame(self._native_frame.select(*column_names))
+        return self._from_native_frame(
+            self._native_frame.select(*column_names), validate_column_names=False
+        )
 
     def aggregate(
         self: Self,
@@ -214,7 +243,9 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
         new_columns = parse_exprs(self, *exprs)
 
         new_columns_list = [col.alias(col_name) for col_name, col in new_columns.items()]
-        return self._from_native_frame(self._native_frame.agg(*new_columns_list))
+        return self._from_native_frame(
+            self._native_frame.agg(*new_columns_list), validate_column_names=False
+        )
 
     def select(
         self: Self,
@@ -224,17 +255,18 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
 
         if not new_columns:
             # return empty dataframe, like Polars does
-            spark_session = self._native_frame.sparkSession
-            spark_df = spark_session.createDataFrame(
+            spark_df = self._session.createDataFrame(
                 [], self._native_dtypes.StructType([])
             )
 
-            return self._from_native_frame(spark_df)
+            return self._from_native_frame(spark_df, validate_column_names=False)
 
         new_columns_list = [
             col.alias(col_name) for (col_name, col) in new_columns.items()
         ]
-        return self._from_native_frame(self._native_frame.select(*new_columns_list))
+        return self._from_native_frame(
+            self._native_frame.select(*new_columns_list), validate_column_names=False
+        )
 
     def with_columns(self: Self, *exprs: SparkLikeExpr) -> Self:
         new_columns = parse_exprs(self, *exprs)
@@ -244,7 +276,7 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
         # `[0]` is safe as the predicate's expression only returns a single column
         condition = predicate._call(self)[0]
         spark_df = self._native_frame.where(condition)
-        return self._from_native_frame(spark_df)
+        return self._from_native_frame(spark_df, validate_column_names=False)
 
     @property
     def schema(self: Self) -> dict[str, DType]:
@@ -264,13 +296,13 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
         columns_to_drop = parse_columns_to_drop(
             compliant_frame=self, columns=columns, strict=strict
         )
-        return self._from_native_frame(self._native_frame.drop(*columns_to_drop))
+        return self._from_native_frame(
+            self._native_frame.drop(*columns_to_drop), validate_column_names=False
+        )
 
     def head(self: Self, n: int) -> Self:
-        spark_session = self._native_frame.sparkSession
-
         return self._from_native_frame(
-            spark_session.createDataFrame(self._native_frame.take(num=n))
+            self._native_frame.limit(num=n), validate_column_names=False
         )
 
     def group_by(self: Self, *keys: str, drop_null_keys: bool) -> SparkLikeLazyGroupBy:
@@ -301,10 +333,14 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
             )
 
         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))
+        return self._from_native_frame(
+            self._native_frame.sort(*sort_cols), validate_column_names=False
+        )
 
     def drop_nulls(self: Self, subset: list[str] | None) -> Self:
-        return self._from_native_frame(self._native_frame.dropna(subset=subset))
+        return self._from_native_frame(
+            self._native_frame.dropna(subset=subset), validate_column_names=False
+        )
 
     def rename(self: Self, mapping: dict[str, str]) -> Self:
         rename_mapping = {
@@ -326,7 +362,9 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
             msg = "`LazyFrame.unique` with PySpark backend only supports `keep='any'`."
             raise ValueError(msg)
         check_column_exists(self.columns, subset)
-        return self._from_native_frame(self._native_frame.dropDuplicates(subset=subset))
+        return self._from_native_frame(
+            self._native_frame.dropDuplicates(subset=subset), validate_column_names=False
+        )
 
     def join(
         self: Self,
@@ -357,7 +395,7 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
                 for colname in list(set(right_columns).difference(set(right_on or [])))
             },
         }
-        other = other_native.select(
+        other_native = other_native.select(
             [self._F.col(old).alias(new) for old, new in rename_mapping.items()]
         )
 
@@ -375,7 +413,7 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
                 ]
             )
         return self._from_native_frame(
-            self_native.join(other, on=left_on, how=how).select(col_order)
+            self_native.join(other_native, on=left_on, how=how).select(col_order)
         )
 
     def explode(self: Self, columns: list[str]) -> Self:
@@ -402,16 +440,51 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
             )
             raise NotImplementedError(msg)
 
-        return self._from_native_frame(
-            native_frame.select(
-                *[
-                    self._F.col(col_name).alias(col_name)
-                    if col_name != columns[0]
-                    else self._F.explode_outer(col_name).alias(col_name)
-                    for col_name in column_names
-                ]
+        if self._implementation.is_pyspark():
+            return self._from_native_frame(
+                native_frame.select(
+                    *[
+                        self._F.col(col_name).alias(col_name)
+                        if col_name != columns[0]
+                        else self._F.explode_outer(col_name).alias(col_name)
+                        for col_name in column_names
+                    ]
+                ),
+                validate_column_names=False,
             )
-        )
+        elif self._implementation.is_sqlframe():
+            # Not every sqlframe dialect supports `explode_outer` function
+            # (see https://github.com/eakmanrq/sqlframe/blob/3cb899c515b101ff4c197d84b34fae490d0ed257/sqlframe/base/functions.py#L2288-L2289)
+            # therefore we simply explode the array column which will ignore nulls and
+            # zero sized arrays, and append these specific condition with nulls (to
+            # match polars behavior).
+
+            def null_condition(col_name: str) -> Column:
+                return self._F.isnull(col_name) | (self._F.array_size(col_name) == 0)
+
+            return self._from_native_frame(
+                native_frame.select(
+                    *[
+                        self._F.col(col_name).alias(col_name)
+                        if col_name != columns[0]
+                        else self._F.explode(col_name).alias(col_name)
+                        for col_name in column_names
+                    ]
+                ).union(
+                    native_frame.filter(null_condition(columns[0])).select(
+                        *[
+                            self._F.col(col_name).alias(col_name)
+                            if col_name != columns[0]
+                            else self._F.lit(None).alias(col_name)
+                            for col_name in column_names
+                        ]
+                    )
+                ),
+                validate_column_names=False,
+            )
+        else:  # pragma: no cover
+            msg = "Unreachable code, please report an issue at https://github.com/narwhals-dev/narwhals/issues"
+            raise AssertionError(msg)
 
     def unpivot(
         self: Self,
@@ -420,6 +493,15 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
         variable_name: str,
         value_name: str,
     ) -> Self:
+        if self._implementation.is_sqlframe():
+            if variable_name == "":
+                msg = "`variable_name` cannot be empty string for sqlframe backend."
+                raise NotImplementedError(msg)
+
+            if value_name == "":
+                msg = "`value_name` cannot be empty string for sqlframe backend."
+                raise NotImplementedError(msg)
+
         ids = tuple(self.columns) if index is None else tuple(index)
         values = (
             tuple(set(self.columns).difference(set(ids))) if on is None else tuple(on)
diff --git a/narwhals/_spark_like/expr.py b/narwhals/_spark_like/expr.py
index dbc833d3..0bc57253 100644
--- a/narwhals/_spark_like/expr.py
+++ b/narwhals/_spark_like/expr.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import operator
+from importlib import import_module
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Callable
@@ -15,6 +16,7 @@ from narwhals._spark_like.expr_name import SparkLikeExprNameNamespace
 from narwhals._spark_like.expr_str import SparkLikeExprStringNamespace
 from narwhals._spark_like.utils import maybe_evaluate_expr
 from narwhals._spark_like.utils import narwhals_to_native_dtype
+from narwhals.dependencies import get_pyspark
 from narwhals.typing import CompliantExpr
 from narwhals.utils import Implementation
 from narwhals.utils import parse_version
@@ -75,31 +77,41 @@ class SparkLikeExpr(CompliantExpr["SparkLikeLazyFrame", "Column"]):
         )
 
     @property
-    def _F(self) -> Any:  # noqa: N802
+    def _F(self: Self) -> Any:  # noqa: N802
         if self._implementation is Implementation.SQLFRAME:
-            from sqlframe.duckdb import functions
+            from sqlframe.base.session import _BaseSession
+
+            return import_module(
+                f"sqlframe.{_BaseSession().execution_dialect_name}.functions"
+            )
 
-            return functions
         from pyspark.sql import functions
 
         return functions
 
     @property
-    def _native_types(self) -> Any:
+    def _native_dtypes(self: Self) -> Any:
         if self._implementation is Implementation.SQLFRAME:
-            from sqlframe.duckdb import types
+            from sqlframe.base.session import _BaseSession
+
+            return import_module(
+                f"sqlframe.{_BaseSession().execution_dialect_name}.types"
+            )
 
-            return types
         from pyspark.sql import types
 
         return types
 
     @property
-    def _Window(self) -> Any:  # noqa: N802
+    def _Window(self: Self) -> Any:  # noqa: N802
         if self._implementation is Implementation.SQLFRAME:
-            from sqlframe.duckdb import Window
+            from sqlframe.base.session import _BaseSession
+
+            _window = import_module(
+                f"sqlframe.{_BaseSession().execution_dialect_name}.window"
+            )
+            return _window.Window
 
-            return Window
         from pyspark.sql import Window
 
         return Window
@@ -321,7 +333,7 @@ class SparkLikeExpr(CompliantExpr["SparkLikeLazyFrame", "Column"]):
     def cast(self: Self, dtype: DType | type[DType]) -> Self:
         def _cast(_input: Column) -> Column:
             spark_dtype = narwhals_to_native_dtype(
-                dtype, self._version, self._native_types
+                dtype, self._version, self._native_dtypes
             )
             return _input.cast(spark_dtype)
 
@@ -338,9 +350,11 @@ class SparkLikeExpr(CompliantExpr["SparkLikeLazyFrame", "Column"]):
 
     def median(self: Self) -> Self:
         def _median(_input: Column) -> Column:
-            import pyspark  # ignore-banned-import
-
-            if parse_version(pyspark) < (3, 4):
+            if (
+                self._implementation.is_pyspark()
+                and (pyspark := get_pyspark()) is not None
+                and parse_version(pyspark) < (3, 4)
+            ):
                 # Use percentile_approx with default accuracy parameter (10000)
                 return self._F.percentile_approx(_input.cast("double"), 0.5)
 
@@ -422,7 +436,7 @@ class SparkLikeExpr(CompliantExpr["SparkLikeLazyFrame", "Column"]):
 
     def is_in(self: Self, values: Sequence[Any]) -> Self:
         def _is_in(_input: Column) -> Column:
-            return _input.isin(values)
+            return _input.isin(values) if values else self._F.lit(False)  # noqa: FBT003
 
         return self._from_call(_is_in, "is_in")
 
@@ -452,7 +466,7 @@ class SparkLikeExpr(CompliantExpr["SparkLikeLazyFrame", "Column"]):
     def n_unique(self: Self) -> Self:
         def _n_unique(_input: Column) -> Column:
             return self._F.count_distinct(_input) + self._F.max(
-                self._F.isnull(_input).cast(self._native_types.IntegerType())
+                self._F.isnull(_input).cast(self._native_dtypes.IntegerType())
             )
 
         return self._from_call(_n_unique, "n_unique")
diff --git a/narwhals/_spark_like/namespace.py b/narwhals/_spark_like/namespace.py
index 1b68d073..d64401e0 100644
--- a/narwhals/_spark_like/namespace.py
+++ b/narwhals/_spark_like/namespace.py
@@ -246,6 +246,7 @@ class SparkLikeNamespace(CompliantNamespace["SparkLikeLazyFrame", "Column"]):
                 backend_version=self._backend_version,
                 version=self._version,
                 implementation=self._implementation,
+                validate_column_names=False,
             )
 
         if how == "diagonal":
@@ -256,6 +257,7 @@ class SparkLikeNamespace(CompliantNamespace["SparkLikeLazyFrame", "Column"]):
                 backend_version=self._backend_version,
                 version=self._version,
                 implementation=self._implementation,
+                validate_column_names=False,
             )
         raise NotImplementedError
 
diff --git a/narwhals/dataframe.py b/narwhals/dataframe.py
index 0de0c3c3..a461a0e5 100644
--- a/narwhals/dataframe.py
+++ b/narwhals/dataframe.py
@@ -16,9 +16,10 @@ from typing import overload
 from warnings import warn
 
 from narwhals._expression_parsing import ExprKind
-from narwhals._expression_parsing import all_exprs_are_aggs_or_literals
-from narwhals._expression_parsing import check_expressions_transform
+from narwhals._expression_parsing import all_exprs_are_scalar_like
+from narwhals._expression_parsing import check_expressions_preserve_length
 from narwhals._expression_parsing import infer_kind
+from narwhals._expression_parsing import is_scalar_like
 from narwhals.dependencies import get_polars
 from narwhals.dependencies import is_numpy_array
 from narwhals.dependencies import is_numpy_array_1d
@@ -137,9 +138,7 @@ class BaseFrame(Generic[_FrameT]):
     ) -> Self:
         compliant_exprs, kinds = self._flatten_and_extract(*exprs, **named_exprs)
         compliant_exprs = [
-            compliant_expr.broadcast(kind)
-            if (kind is ExprKind.LITERAL or kind is ExprKind.AGGREGATION)
-            else compliant_expr
+            compliant_expr.broadcast(kind) if is_scalar_like(kind) else compliant_expr
             for compliant_expr, kind in zip(compliant_exprs, kinds)
         ]
         return self._from_compliant_dataframe(
@@ -166,14 +165,12 @@ class BaseFrame(Generic[_FrameT]):
                     missing_columns, available_columns
                 ) from e
         compliant_exprs, kinds = self._flatten_and_extract(*flat_exprs, **named_exprs)
-        if compliant_exprs and all_exprs_are_aggs_or_literals(*flat_exprs, **named_exprs):
+        if compliant_exprs and all_exprs_are_scalar_like(*flat_exprs, **named_exprs):
             return self._from_compliant_dataframe(
                 self._compliant_frame.aggregate(*compliant_exprs),
             )
         compliant_exprs = [
-            compliant_expr.broadcast(kind)
-            if (kind is ExprKind.LITERAL or kind is ExprKind.AGGREGATION)
-            else compliant_expr
+            compliant_expr.broadcast(kind) if is_scalar_like(kind) else compliant_expr
             for compliant_expr, kind in zip(compliant_exprs, kinds)
         ]
         return self._from_compliant_dataframe(
@@ -205,7 +202,7 @@ class BaseFrame(Generic[_FrameT]):
             and all(isinstance(x, bool) for x in predicates[0])
         ):
             flat_predicates = flatten(predicates)
-            check_expressions_transform(*flat_predicates, function_name="filter")
+            check_expressions_preserve_length(*flat_predicates, function_name="filter")
             compliant_predicates, _kinds = self._flatten_and_extract(*flat_predicates)
             plx = self.__narwhals_namespace__()
             predicate = plx.all_horizontal(
@@ -2204,18 +2201,18 @@ class LazyFrame(BaseFrame[FrameT]):
             plx = self.__narwhals_namespace__()
             return plx.col(arg)
         if isinstance(arg, Expr):
-            if arg._metadata.is_order_dependent():
+            if arg._metadata.n_open_windows > 0:
                 msg = (
                     "Order-dependent expressions are not supported for use in LazyFrame.\n\n"
                     "Hints:\n"
                     "- Instead of `lf.select(nw.col('a').sort())`, use `lf.select('a').sort()\n"
                     "- Instead of `lf.select(nw.col('a').head())`, use `lf.select('a').head()\n"
                     "- `Expr.cum_sum`, and other such expressions, are not currently supported.\n"
-                    "  In a future version of Narwhals, a `order_by` argument will be added and \n"
-                    "  they will be supported."
+                    "  In a future version of Narwhals, a `order_by` argument will be added to\n"
+                    "  `over` and they will be supported."
                 )
                 raise OrderDependentExprError(msg)
-            if arg._metadata.is_changes_length():
+            if arg._metadata.kind.is_changes_length():
                 msg = (
                     "Length-changing expressions are not supported for use in LazyFrame, unless\n"
                     "followed by an aggregation.\n\n"
diff --git a/narwhals/expr.py b/narwhals/expr.py
index 962d2450..36dc6cda 100644
--- a/narwhals/expr.py
+++ b/narwhals/expr.py
@@ -649,7 +649,7 @@ class Expr:
                 function=function, return_dtype=return_dtype
             ),
             # safest assumptions
-            self._metadata.with_kind_and_order_dependence(ExprKind.CHANGES_LENGTH),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.CHANGES_LENGTH),
         )
 
     def skew(self: Self) -> Self:
@@ -773,7 +773,7 @@ class Expr:
         """
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).arg_min(),
-            ExprMetadata(ExprKind.AGGREGATION, order_dependent=True),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.AGGREGATION),
         )
 
     def arg_max(self: Self) -> Self:
@@ -797,7 +797,7 @@ class Expr:
         """
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).arg_max(),
-            ExprMetadata(ExprKind.AGGREGATION, order_dependent=True),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.AGGREGATION),
         )
 
     def count(self: Self) -> Self:
@@ -922,7 +922,7 @@ class Expr:
         """
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).cum_sum(reverse=reverse),
-            self._metadata.with_order_dependence(),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.WINDOW),
         )
 
     def diff(self: Self) -> Self:
@@ -965,7 +965,7 @@ class Expr:
         """
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).diff(),
-            self._metadata.with_order_dependence(),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.WINDOW),
         )
 
     def shift(self: Self, n: int) -> Self:
@@ -1011,7 +1011,7 @@ class Expr:
         """
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).shift(n),
-            self._metadata.with_order_dependence(),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.WINDOW),
         )
 
     def replace_strict(
@@ -1101,7 +1101,7 @@ class Expr:
             lambda plx: self._to_compliant_expr(plx).sort(
                 descending=descending, nulls_last=nulls_last
             ),
-            self._metadata.with_order_dependence(),
+            self._metadata.with_extra_open_window(),
         )
 
     # --- transform ---
@@ -1321,7 +1321,7 @@ class Expr:
         issue_deprecation_warning(msg, _version="1.23.0")
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).arg_true(),
-            self._metadata.with_kind_and_order_dependence(ExprKind.CHANGES_LENGTH),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.CHANGES_LENGTH),
         )
 
     def fill_null(
@@ -1493,13 +1493,16 @@ class Expr:
             self._metadata.with_kind(ExprKind.CHANGES_LENGTH),
         )
 
-    def over(self: Self, *keys: str | Iterable[str]) -> Self:
+    def over(
+        self: Self, *keys: str | Iterable[str], _order_by: str | None = None
+    ) -> Self:
         """Compute expressions over the given groups.
 
         Arguments:
             keys: Names of columns to compute window expression over.
                   Must be names of columns, as opposed to expressions -
                   so, this is a bit less flexible than Polars' `Expr.over`.
+            _order_by: Unused, but this is building up to something.
 
         Returns:
             A new expression.
@@ -1532,15 +1535,20 @@ class Expr:
             |2  4  y                    4|
             └────────────────────────────┘
         """
-        if self._metadata.is_changes_length():
+        if self._metadata.kind.is_changes_length():
             msg = "`.over()` can not be used for expressions which change length."
             raise LengthChangingExprError(msg)
+        kind = ExprKind.TRANSFORM
+        n_open_windows = self._metadata.n_open_windows
+        if _order_by is not None and self._metadata.kind.is_window():
+            n_open_windows -= 1
+        metadata = ExprMetadata(kind, n_open_windows=n_open_windows)
         flattened = flatten(keys)
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).over(
                 flattened, kind=self._metadata.kind
             ),
-            self._metadata.with_kind(ExprKind.TRANSFORM),
+            metadata,
         )
 
     def is_duplicated(self: Self) -> Self:
@@ -1648,7 +1656,7 @@ class Expr:
         """
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).is_first_distinct(),
-            self._metadata.with_order_dependence(),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.WINDOW),
         )
 
     def is_last_distinct(self: Self) -> Self:
@@ -1677,7 +1685,7 @@ class Expr:
         """
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).is_last_distinct(),
-            self._metadata.with_order_dependence(),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.WINDOW),
         )
 
     def quantile(
@@ -1746,7 +1754,7 @@ class Expr:
         issue_deprecation_warning(msg, _version="1.22.0")
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).head(n),
-            self._metadata.with_kind_and_order_dependence(ExprKind.CHANGES_LENGTH),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.CHANGES_LENGTH),
         )
 
     def tail(self: Self, n: int = 10) -> Self:
@@ -1774,7 +1782,7 @@ class Expr:
         issue_deprecation_warning(msg, _version="1.22.0")
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).tail(n),
-            self._metadata.with_kind_and_order_dependence(ExprKind.CHANGES_LENGTH),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.CHANGES_LENGTH),
         )
 
     def round(self: Self, decimals: int = 0) -> Self:
@@ -1869,7 +1877,7 @@ class Expr:
         issue_deprecation_warning(msg, _version="1.22.0")
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).gather_every(n=n, offset=offset),
-            self._metadata.with_kind_and_order_dependence(ExprKind.CHANGES_LENGTH),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.CHANGES_LENGTH),
         )
 
     # need to allow numeric typing
@@ -2012,7 +2020,7 @@ class Expr:
         """
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).cum_count(reverse=reverse),
-            self._metadata.with_order_dependence(),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.WINDOW),
         )
 
     def cum_min(self: Self, *, reverse: bool = False) -> Self:
@@ -2045,7 +2053,7 @@ class Expr:
         """
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).cum_min(reverse=reverse),
-            self._metadata.with_order_dependence(),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.WINDOW),
         )
 
     def cum_max(self: Self, *, reverse: bool = False) -> Self:
@@ -2078,7 +2086,7 @@ class Expr:
         """
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).cum_max(reverse=reverse),
-            self._metadata.with_order_dependence(),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.WINDOW),
         )
 
     def cum_prod(self: Self, *, reverse: bool = False) -> Self:
@@ -2111,7 +2119,7 @@ class Expr:
         """
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).cum_prod(reverse=reverse),
-            self._metadata.with_order_dependence(),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.WINDOW),
         )
 
     def rolling_sum(
@@ -2173,7 +2181,7 @@ class Expr:
                 min_samples=min_samples,
                 center=center,
             ),
-            self._metadata.with_order_dependence(),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.WINDOW),
         )
 
     def rolling_mean(
@@ -2235,7 +2243,7 @@ class Expr:
                 min_samples=min_samples,
                 center=center,
             ),
-            self._metadata.with_order_dependence(),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.WINDOW),
         )
 
     def rolling_var(
@@ -2297,7 +2305,7 @@ class Expr:
             lambda plx: self._to_compliant_expr(plx).rolling_var(
                 window_size=window_size, min_samples=min_samples, center=center, ddof=ddof
             ),
-            self._metadata.with_order_dependence(),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.WINDOW),
         )
 
     def rolling_std(
@@ -2362,7 +2370,7 @@ class Expr:
                 center=center,
                 ddof=ddof,
             ),
-            self._metadata.with_order_dependence(),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.WINDOW),
         )
 
     def rank(
@@ -2428,7 +2436,7 @@ class Expr:
             lambda plx: self._to_compliant_expr(plx).rank(
                 method=method, descending=descending
             ),
-            self._metadata.with_order_dependence(),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.WINDOW),
         )
 
     @property
diff --git a/narwhals/functions.py b/narwhals/functions.py
index 17e69a2f..9f64dba2 100644
--- a/narwhals/functions.py
+++ b/narwhals/functions.py
@@ -16,10 +16,11 @@ from typing import overload
 from narwhals._expression_parsing import ExprKind
 from narwhals._expression_parsing import ExprMetadata
 from narwhals._expression_parsing import apply_n_ary_operation
-from narwhals._expression_parsing import check_expressions_transform
+from narwhals._expression_parsing import check_expressions_preserve_length
 from narwhals._expression_parsing import combine_metadata
 from narwhals._expression_parsing import extract_compliant
 from narwhals._expression_parsing import infer_kind
+from narwhals._expression_parsing import is_scalar_like
 from narwhals.dataframe import DataFrame
 from narwhals.dataframe import LazyFrame
 from narwhals.dependencies import is_numpy_array
@@ -1138,7 +1139,7 @@ def len_() -> Expr:
     def func(plx: Any) -> Any:
         return plx.len()
 
-    return Expr(func, ExprMetadata(ExprKind.AGGREGATION, order_dependent=False))
+    return Expr(func, ExprMetadata(ExprKind.AGGREGATION, n_open_windows=0))
 
 
 def sum(*columns: str) -> Expr:
@@ -1436,7 +1437,7 @@ def max_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:
 class When:
     def __init__(self: Self, *predicates: IntoExpr | Iterable[IntoExpr]) -> None:
         self._predicate = all_horizontal(*flatten(predicates))
-        check_expressions_transform(self._predicate, function_name="when")
+        check_expressions_preserve_length(self._predicate, function_name="when")
 
     def then(self: Self, value: IntoExpr | Any) -> Then:
         return Then(
@@ -1458,9 +1459,7 @@ class Then(Expr):
         def func(plx: CompliantNamespace[Any, Any]) -> CompliantExpr[Any, Any]:
             compliant_expr = self._to_compliant_expr(plx)
             compliant_value = extract_compliant(plx, value, str_as_lit=False)
-            if (
-                kind is ExprKind.AGGREGATION or kind is ExprKind.LITERAL
-            ) and is_compliant_expr(compliant_value):
+            if is_scalar_like(kind) and is_compliant_expr(compliant_value):
                 compliant_value = compliant_value.broadcast(kind)
             return compliant_expr.otherwise(compliant_value)  # type: ignore[no-any-return]
 
@@ -1596,7 +1595,7 @@ def lit(value: Any, dtype: DType | type[DType] | None = None) -> Expr:
 
     return Expr(
         lambda plx: plx.lit(value, dtype),
-        ExprMetadata(ExprKind.LITERAL, order_dependent=False),
+        ExprMetadata(ExprKind.LITERAL, n_open_windows=0),
     )
 
 
diff --git a/narwhals/group_by.py b/narwhals/group_by.py
index 7e13ef02..e299d17d 100644
--- a/narwhals/group_by.py
+++ b/narwhals/group_by.py
@@ -8,7 +8,9 @@ from typing import Iterator
 from typing import TypeVar
 from typing import cast
 
-from narwhals._expression_parsing import all_exprs_are_aggs_or_literals
+from narwhals._expression_parsing import all_exprs_are_scalar_like
+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
@@ -113,7 +115,7 @@ class GroupBy(Generic[DataFrameT]):
             └─────┴─────┴─────┘
         """
         flat_aggs = tuple(flatten(aggs))
-        if not all_exprs_are_aggs_or_literals(*flat_aggs, **named_aggs):
+        if not all_exprs_are_scalar_like(*flat_aggs, **named_aggs):
             msg = (
                 "Found expression which does not aggregate.\n\n"
                 "All expressions passed to GroupBy.agg must aggregate.\n"
@@ -213,7 +215,7 @@ class LazyGroupBy(Generic[LazyFrameT]):
             └─────┴─────┴─────┘
         """
         flat_aggs = tuple(flatten(aggs))
-        if not all_exprs_are_aggs_or_literals(*flat_aggs, **named_aggs):
+        if not all_exprs_are_scalar_like(*flat_aggs, **named_aggs):
             msg = (
                 "Found expression which does not aggregate.\n\n"
                 "All expressions passed to GroupBy.agg must aggregate.\n"
diff --git a/narwhals/stable/v1/__init__.py b/narwhals/stable/v1/__init__.py
index 1001a7a7..3d001af3 100644
--- a/narwhals/stable/v1/__init__.py
+++ b/narwhals/stable/v1/__init__.py
@@ -968,7 +968,7 @@ class Expr(NwExpr):
         """
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).head(n),
-            self._metadata.with_kind_and_order_dependence(ExprKind.CHANGES_LENGTH),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.CHANGES_LENGTH),
         )
 
     def tail(self: Self, n: int = 10) -> Self:
@@ -982,7 +982,7 @@ class Expr(NwExpr):
         """
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).tail(n),
-            self._metadata.with_kind_and_order_dependence(ExprKind.CHANGES_LENGTH),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.CHANGES_LENGTH),
         )
 
     def gather_every(self: Self, n: int, offset: int = 0) -> Self:
@@ -997,7 +997,7 @@ class Expr(NwExpr):
         """
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).gather_every(n=n, offset=offset),
-            self._metadata.with_kind_and_order_dependence(ExprKind.CHANGES_LENGTH),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.CHANGES_LENGTH),
         )
 
     def unique(self: Self, *, maintain_order: bool | None = None) -> Self:
@@ -1036,7 +1036,7 @@ class Expr(NwExpr):
             lambda plx: self._to_compliant_expr(plx).sort(
                 descending=descending, nulls_last=nulls_last
             ),
-            self._metadata.with_order_dependence(),
+            self._metadata.with_extra_open_window(),
         )
 
     def arg_true(self: Self) -> Self:
@@ -1047,7 +1047,7 @@ class Expr(NwExpr):
         """
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).arg_true(),
-            self._metadata.with_kind_and_order_dependence(ExprKind.CHANGES_LENGTH),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.CHANGES_LENGTH),
         )
 
     def sample(
@@ -1081,7 +1081,7 @@ class Expr(NwExpr):
             lambda plx: self._to_compliant_expr(plx).sample(
                 n, fraction=fraction, with_replacement=with_replacement, seed=seed
             ),
-            self._metadata.with_kind_and_order_dependence(ExprKind.CHANGES_LENGTH),
+            self._metadata.with_kind(ExprKind.CHANGES_LENGTH),
         )
 
 
diff --git a/narwhals/translate.py b/narwhals/translate.py
index 5dd7c839..e545c272 100644
--- a/narwhals/translate.py
+++ b/narwhals/translate.py
@@ -738,6 +738,7 @@ def _from_native_impl(  # noqa: PLR0915
                 backend_version=parse_version(get_pyspark()),
                 version=version,
                 implementation=Implementation.PYSPARK,
+                validate_column_names=True,
             ),
             level="lazy",
         )
@@ -760,6 +761,7 @@ def _from_native_impl(  # noqa: PLR0915
                 backend_version=backend_version,
                 version=version,
                 implementation=Implementation.SQLFRAME,
+                validate_column_names=True,
             ),
             level="lazy",
         )
diff --git a/narwhals/utils.py b/narwhals/utils.py
index 172fa5d9..da8eadb8 100644
--- a/narwhals/utils.py
+++ b/narwhals/utils.py
@@ -26,6 +26,7 @@ from narwhals.dependencies import get_pandas
 from narwhals.dependencies import get_polars
 from narwhals.dependencies import get_pyarrow
 from narwhals.dependencies import get_pyspark_sql
+from narwhals.dependencies import get_sqlframe
 from narwhals.dependencies import is_cudf_series
 from narwhals.dependencies import is_modin_series
 from narwhals.dependencies import is_pandas_dataframe
@@ -157,6 +158,7 @@ class Implementation(Enum):
             get_dask_dataframe(): Implementation.DASK,
             get_duckdb(): Implementation.DUCKDB,
             get_ibis(): Implementation.IBIS,
+            get_sqlframe(): Implementation.SQLFRAME,
         }
         return mapping.get(native_namespace, Implementation.UNKNOWN)
 
@@ -182,6 +184,7 @@ class Implementation(Enum):
             "dask": Implementation.DASK,
             "duckdb": Implementation.DUCKDB,
             "ibis": Implementation.IBIS,
+            "sqlframe": Implementation.SQLFRAME,
         }
         return mapping.get(backend_name, Implementation.UNKNOWN)
 
@@ -244,6 +247,12 @@ class Implementation(Enum):
             import duckdb  # ignore-banned-import
 
             return duckdb
+
+        if self is Implementation.SQLFRAME:
+            import sqlframe  # ignore-banned-import
+
+            return sqlframe
+
         msg = "Not supported Implementation"  # pragma: no cover
         raise AssertionError(msg)
 
@@ -283,6 +292,22 @@ class Implementation(Enum):
             Implementation.CUDF,
         }
 
+    def is_spark_like(self: Self) -> bool:
+        """Return whether implementation is pyspark or sqlframe.
+
+        Returns:
+            Boolean.
+
+        Examples:
+            >>> import pandas as pd
+            >>> import narwhals as nw
+            >>> df_native = pd.DataFrame({"a": [1, 2, 3]})
+            >>> df = nw.from_native(df_native)
+            >>> df.implementation.is_spark_like()
+            False
+        """
+        return self in {Implementation.PYSPARK, Implementation.SQLFRAME}
+
     def is_polars(self: Self) -> bool:
         """Return whether implementation is Polars.
 
@@ -411,6 +436,22 @@ class Implementation(Enum):
         """
         return self is Implementation.IBIS  # pragma: no cover
 
+    def is_sqlframe(self: Self) -> bool:
+        """Return whether implementation is SQLFrame.
+
+        Returns:
+            Boolean.
+
+        Examples:
+            >>> import polars as pl
+            >>> import narwhals as nw
+            >>> df_native = pl.DataFrame({"a": [1, 2, 3]})
+            >>> df = nw.from_native(df_native)
+            >>> df.implementation.is_sqlframe()
+            False
+        """
+        return self is Implementation.SQLFRAME  # pragma: no cover
+
 
 MIN_VERSIONS: dict[Implementation, tuple[int, ...]] = {
     Implementation.PANDAS: (0, 25, 3),
@@ -422,7 +463,7 @@ MIN_VERSIONS: dict[Implementation, tuple[int, ...]] = {
     Implementation.DASK: (2024, 8),
     Implementation.DUCKDB: (1,),
     Implementation.IBIS: (6,),
-    Implementation.SQLFRAME: (3, 14, 2),
+    Implementation.SQLFRAME: (3, 22, 0),
 }
 
 
diff --git a/pyproject.toml b/pyproject.toml
index 44984eb9..8a5dc190 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -37,6 +37,7 @@ polars = ["polars>=0.20.3"]
 dask = ["dask[dataframe]>=2024.8"]
 duckdb = ["duckdb>=1.0"]
 ibis = ["ibis-framework>=6.0.0", "rich", "packaging", "pyarrow_hotfix"]
+sqlframe = ["sqlframe>=3.22.0"]
 tests = [
   "covdefaults",
   "pytest",
@@ -201,7 +202,8 @@ filterwarnings = [
   'ignore:.*is_datetime64tz_dtype is deprecated and will be removed in a future version.*:DeprecationWarning:pyspark',
   # Warning raised by PyArrow nightly just by importing pandas
   'ignore:.*Python binding for RankQuantileOptions not exposed:RuntimeWarning:pyarrow',
-  'ignore:.*pandas only supports SQLAlchemy:UserWarning'
+  'ignore:.*pandas only supports SQLAlchemy:UserWarning',
+  'ignore:.*numpy.core is deprecated and has been renamed to numpy._core.*:DeprecationWarning:sqlframe',
 ]
 xfail_strict = true
 markers = ["slow: marks tests as slow (deselect with '-m \"not slow\"')"]
@@ -267,6 +269,7 @@ module = [
   "*._pandas_like.*",
   "*._ibis.*",
   "*._arrow.*",
+  "*._dask.*",
 ]
 warn_return_any = false
 


Test patch:
diff --git a/.github/workflows/pytest.yml b/.github/workflows/pytest.yml
index 3042f630..6bb570e5 100644
--- a/.github/workflows/pytest.yml
+++ b/.github/workflows/pytest.yml
@@ -53,12 +53,12 @@ jobs:
           cache-dependency-glob: "pyproject.toml"
       - name: install-reqs
         # we are not testing pyspark on Windows here because it is very slow
-        run: uv pip install -e ".[tests, core, extra, dask, modin]" --system
+        run: uv pip install -e ".[tests, core, extra, dask, modin, sqlframe]" --system
       - name: show-deps
         run: uv pip freeze
       - name: Run pytest
         run: |
-          pytest tests --cov=narwhals --cov=tests --runslow --cov-fail-under=95 --constructors=pandas,pandas[nullable],pandas[pyarrow],pyarrow,modin[pyarrow],polars[eager],polars[lazy],dask,duckdb
+          pytest tests --cov=narwhals --cov=tests --runslow --cov-fail-under=95 --constructors=pandas,pandas[nullable],pandas[pyarrow],pyarrow,modin[pyarrow],polars[eager],polars[lazy],dask,duckdb,sqlframe
 
   pytest-full-coverage:
     strategy:
@@ -83,7 +83,7 @@ jobs:
           cache-suffix: ${{ matrix.python-version }}
           cache-dependency-glob: "pyproject.toml"
       - name: install-reqs
-        run: uv pip install -e ".[tests, core, extra, modin, dask]" --system
+        run: uv pip install -e ".[tests, core, extra, modin, dask, sqlframe]" --system
       - name: install pyspark
         run: uv pip install -e ".[pyspark]" --system
         # PySpark is not yet available on Python3.12+
diff --git a/tests/conftest.py b/tests/conftest.py
index 9ee92bcb..a0cde090 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -184,7 +184,7 @@ def pyspark_lazy_constructor() -> Callable[[Any], IntoFrame]:  # pragma: no cove
 
 def sqlframe_pyspark_lazy_constructor(
     obj: dict[str, Any],
-) -> Callable[[Any], IntoFrame]:  # pragma: no cover
+) -> IntoFrame:  # pragma: no cover
     from sqlframe.duckdb import DuckDBSession
 
     session = DuckDBSession()
@@ -208,9 +208,7 @@ LAZY_CONSTRUCTORS: dict[str, Callable[[Any], IntoFrame]] = {
     "polars[lazy]": polars_lazy_constructor,
     "duckdb": duckdb_lazy_constructor,
     "pyspark": pyspark_lazy_constructor,  # type: ignore[dict-item]
-    # We've reported several bugs to sqlframe - once they address
-    # them, we can start testing them as part of our CI.
-    # "sqlframe": sqlframe_pyspark_lazy_constructor,  # noqa: ERA001
+    "sqlframe": sqlframe_pyspark_lazy_constructor,
 }
 GPU_CONSTRUCTORS: dict[str, Callable[[Any], IntoFrame]] = {"cudf": cudf_constructor}
 
diff --git a/tests/expr_and_series/cast_test.py b/tests/expr_and_series/cast_test.py
index 1fa603bd..1f512cde 100644
--- a/tests/expr_and_series/cast_test.py
+++ b/tests/expr_and_series/cast_test.py
@@ -54,7 +54,7 @@ SCHEMA = {
     "p": nw.Int64,
 }
 
-SPARK_INCOMPATIBLE_COLUMNS = {"e", "f", "g", "h", "l", "o", "p"}
+SPARK_LIKE_INCOMPATIBLE_COLUMNS = {"e", "f", "g", "h", "l", "o", "p"}
 DUCKDB_INCOMPATIBLE_COLUMNS = {"l", "o", "p"}
 
 
@@ -72,7 +72,7 @@ def test_cast(
         request.applymarker(pytest.mark.xfail)
 
     if "pyspark" in str(constructor):
-        incompatible_columns = SPARK_INCOMPATIBLE_COLUMNS  # pragma: no cover
+        incompatible_columns = SPARK_LIKE_INCOMPATIBLE_COLUMNS  # pragma: no cover
     elif "duckdb" in str(constructor):
         incompatible_columns = DUCKDB_INCOMPATIBLE_COLUMNS  # pragma: no cover
     else:
@@ -185,7 +185,7 @@ def test_cast_raises_for_unknown_dtype(
         request.applymarker(pytest.mark.xfail)
 
     if "pyspark" in str(constructor):
-        incompatible_columns = SPARK_INCOMPATIBLE_COLUMNS  # pragma: no cover
+        incompatible_columns = SPARK_LIKE_INCOMPATIBLE_COLUMNS  # pragma: no cover
     else:
         incompatible_columns = set()
 
@@ -236,7 +236,9 @@ def test_cast_datetime_tz_aware(
 
 
 def test_cast_struct(request: pytest.FixtureRequest, constructor: Constructor) -> None:
-    if any(backend in str(constructor) for backend in ("dask", "modin", "cudf")):
+    if any(
+        backend in str(constructor) for backend in ("dask", "modin", "cudf", "sqlframe")
+    ):
         request.applymarker(pytest.mark.xfail)
 
     if "pandas" in str(constructor) and PANDAS_VERSION < (2, 2):
@@ -254,8 +256,9 @@ def test_cast_struct(request: pytest.FixtureRequest, constructor: Constructor) -
     if "spark" in str(constructor):  # pragma: no cover
         # Special handling for pyspark as it natively maps the input to
         # a column of type MAP<STRING, STRING>
-        import pyspark.sql.functions as F  # noqa: N812
-        import pyspark.sql.types as T  # noqa: N812
+        _tmp_nw_compliant_frame = nw.from_native(native_df)._compliant_frame
+        F = _tmp_nw_compliant_frame._F  # noqa: N806
+        T = _tmp_nw_compliant_frame._native_dtypes  # noqa: N806
 
         native_df = native_df.withColumn(  # type: ignore[union-attr]
             "a",
diff --git a/tests/expr_and_series/over_test.py b/tests/expr_and_series/over_test.py
index 65dba7f6..65d4f60e 100644
--- a/tests/expr_and_series/over_test.py
+++ b/tests/expr_and_series/over_test.py
@@ -268,5 +268,5 @@ def test_unsupported_over() -> None:
     with pytest.raises(NotImplementedError, match="elementary"):
         nw.from_native(df).select(nw.col("a").shift(1).cum_sum().over("b"))
     tbl = pa.table(data)  # type: ignore[arg-type]
-    with pytest.raises(NotImplementedError, match="Elementwise"):
+    with pytest.raises(NotImplementedError, match="aggregation or literal"):
         nw.from_native(tbl).select(nw.col("a").shift(1).cum_sum().over("b"))
diff --git a/tests/expr_and_series/str/to_uppercase_to_lowercase_test.py b/tests/expr_and_series/str/to_uppercase_to_lowercase_test.py
index c70b5f84..28aaa713 100644
--- a/tests/expr_and_series/str/to_uppercase_to_lowercase_test.py
+++ b/tests/expr_and_series/str/to_uppercase_to_lowercase_test.py
@@ -37,6 +37,7 @@ def test_str_to_uppercase(
             "pyarrow_table_constructor",
             "modin_pyarrow_constructor",
             "duckdb_lazy_constructor",
+            "sqlframe_pyspark_lazy_constructor",
         }
         or ("dask" in str(constructor) and PYARROW_VERSION >= (12,))
     ):
@@ -44,7 +45,6 @@ def test_str_to_uppercase(
         # since the pyarrow backend will convert
         # smaller cap 'ß' to upper cap 'ẞ' instead of 'SS'
         request.applymarker(pytest.mark.xfail)
-
     df = nw.from_native(constructor(data))
     result_frame = df.select(nw.col("a").str.to_uppercase())
 
diff --git a/tests/expr_and_series/unary_test.py b/tests/expr_and_series/unary_test.py
index 280ace82..40c9a49e 100644
--- a/tests/expr_and_series/unary_test.py
+++ b/tests/expr_and_series/unary_test.py
@@ -77,7 +77,11 @@ def test_unary_series(constructor_eager: ConstructorEager) -> None:
     assert_equal_data(result, expected)
 
 
-def test_unary_two_elements(constructor: Constructor) -> None:
+def test_unary_two_elements(
+    constructor: Constructor, request: pytest.FixtureRequest
+) -> None:
+    if "sqlframe" in str(constructor):
+        request.applymarker(pytest.mark.xfail)
     data = {"a": [1, 2], "b": [2, 10], "c": [2.0, None]}
     result = nw.from_native(constructor(data)).select(
         a_nunique=nw.col("a").n_unique(),
@@ -123,7 +127,7 @@ def test_unary_two_elements_series(constructor_eager: ConstructorEager) -> None:
 def test_unary_one_element(
     constructor: Constructor, request: pytest.FixtureRequest
 ) -> None:
-    if "pyspark" in str(constructor):
+    if "pyspark" in str(constructor) and "sqlframe" not in str(constructor):
         request.applymarker(pytest.mark.xfail)
     data = {"a": [1], "b": [2], "c": [None]}
     # Dask runs into a divide by zero RuntimeWarning for 1 element skew.
diff --git a/tests/expression_parsing_test.py b/tests/expression_parsing_test.py
new file mode 100644
index 00000000..b540a6fa
--- /dev/null
+++ b/tests/expression_parsing_test.py
@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+import pytest
+
+import narwhals as nw
+
+
+@pytest.mark.parametrize(
+    ("expr", "expected"),
+    [
+        (nw.col("a"), 0),
+        (nw.col("a").mean(), 0),
+        (nw.col("a").cum_sum(), 1),
+        (nw.col("a").cum_sum().over(_order_by="id"), 0),
+        ((nw.col("a").cum_sum() + 1).over(_order_by="id"), 1),
+        (nw.col("a").cum_sum().cum_sum().over(_order_by="id"), 1),
+        (nw.col("a").cum_sum().cum_sum(), 2),
+        (nw.sum_horizontal(nw.col("a"), nw.col("a").cum_sum()), 1),
+        (nw.sum_horizontal(nw.col("a"), nw.col("a").cum_sum()).over("a"), 1),
+    ],
+)
+def test_has_open_windows(expr: nw.Expr, expected: int) -> None:
+    assert expr._metadata.n_open_windows == expected
diff --git a/tests/frame/join_test.py b/tests/frame/join_test.py
index 71058638..41a9823d 100644
--- a/tests/frame/join_test.py
+++ b/tests/frame/join_test.py
@@ -669,14 +669,14 @@ def test_joinasof_by_exceptions(constructor: Constructor) -> None:
 def test_join_duplicate_column_names(
     constructor: Constructor, request: pytest.FixtureRequest
 ) -> None:
-    if "polars" in str(constructor):
-        # https://github.com/pola-rs/polars/issues/21048
-        request.applymarker(pytest.mark.xfail)
-    if "cudf" in str(constructor):
+    if (
+        "polars" in str(constructor)  # https://github.com/pola-rs/polars/issues/21048
+        or "cudf" in str(constructor)
         # TODO(unassigned): cudf doesn't raise here for some reason,
         # need to investigate.
+    ):
         request.applymarker(pytest.mark.xfail)
-    if "pyspark" in str(constructor):
+    if "pyspark" in str(constructor) and "sqlframe" not in str(constructor):
         from pyspark.errors import AnalysisException
 
         exception = AnalysisException
diff --git a/tests/frame/select_test.py b/tests/frame/select_test.py
index ee46a4a6..3ba2110a 100644
--- a/tests/frame/select_test.py
+++ b/tests/frame/select_test.py
@@ -30,7 +30,7 @@ def test_select(constructor: Constructor) -> None:
 
 
 def test_empty_select(constructor: Constructor, request: pytest.FixtureRequest) -> None:
-    if "duckdb" in str(constructor):
+    if "duckdb" in str(constructor) or "sqlframe" in str(constructor):
         request.applymarker(pytest.mark.xfail)
     result = nw.from_native(constructor({"a": [1, 2, 3]})).lazy().select()
     assert result.collect().shape == (0, 0)
@@ -151,3 +151,10 @@ def test_changes_length_vs_aggregation(constructor_eager: ConstructorEager) -> N
     result = df.select(nw.sum_horizontal(nw.col("a").drop_nulls(), nw.col("a").mean()))
     expected = {"a": [3.0, 5.0]}
     assert_equal_data(result, expected)
+
+
+def test_binary_window_aggregation(constructor_eager: ConstructorEager) -> None:
+    df = nw.from_native(constructor_eager({"a": [1, 1, 2]}))
+    result = df.select(nw.col("a").cum_sum() + nw.col("a").sum())
+    expected = {"a": [5, 6, 8]}
+    assert_equal_data(result, expected)
diff --git a/tests/frame/unique_test.py b/tests/frame/unique_test.py
index 8e1b71de..61d64200 100644
--- a/tests/frame/unique_test.py
+++ b/tests/frame/unique_test.py
@@ -39,9 +39,7 @@ def test_unique(
         "last",
     }:
         context: Any = pytest.raises(ValueError, match="row order")
-    elif (
-        keep == "none" and df.implementation is nw.Implementation.PYSPARK
-    ):  # pragma: no cover
+    elif keep == "none" and df.implementation.is_spark_like():  # pragma: no cover
         context = pytest.raises(
             ValueError,
             match="`LazyFrame.unique` with PySpark backend only supports `keep='any'`.",
diff --git a/tests/frame/unpivot_test.py b/tests/frame/unpivot_test.py
index e1573c92..706defa9 100644
--- a/tests/frame/unpivot_test.py
+++ b/tests/frame/unpivot_test.py
@@ -90,7 +90,15 @@ def test_unpivot_var_value_names(
 ) -> None:
     context = (
         pytest.raises(NotImplementedError)
-        if ("duckdb" in str(constructor) and any([variable_name == "", value_name == ""]))
+        if (
+            any([variable_name == "", value_name == ""])
+            and (
+                "duckdb" in str(constructor)
+                # This might depend from the dialect we use in sqlframe.
+                # Since for now we use only duckdb, we need to xfail it
+                or "sqlframe" in str(constructor)
+            )
+        )
         else does_not_raise()
     )
 
diff --git a/tests/selectors_test.py b/tests/selectors_test.py
index c7703c5d..b9e06f7e 100644
--- a/tests/selectors_test.py
+++ b/tests/selectors_test.py
@@ -221,7 +221,7 @@ def test_set_ops(
     expected: list[str],
     request: pytest.FixtureRequest,
 ) -> None:
-    if "duckdb" in str(constructor) and not expected:
+    if ("duckdb" in str(constructor) or "sqlframe" in str(constructor)) and not expected:
         request.applymarker(pytest.mark.xfail)
     df = nw.from_native(constructor(data))
     result = df.select(selector).collect_schema().names()