Repository: narwhals-dev/narwhals
Pull request #2093 (2025-02-25 13:17:46)

perf: use cached property for schema/columns lazy frames
thanks @dangotbanned for highlighting this area of improvement!

<!--
# 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: ah same error...ok, gonna remove the `lru_cache` for now, and do the manual `cached_property`
MarcoGorelli: pyarrow is completely eager so computing schema there should really be practically free

Patch:
diff --git a/narwhals/_arrow/dataframe.py b/narwhals/_arrow/dataframe.py
index 6b97c230..d08e6c71 100644
--- a/narwhals/_arrow/dataframe.py
+++ b/narwhals/_arrow/dataframe.py
@@ -356,7 +356,7 @@ class ArrowDataFrame(CompliantDataFrame, CompliantLazyFrame):
         names = [s.name for s in new_series]
         new_series = align_series_full_broadcast(*new_series)
         df = pa.Table.from_arrays([s._native_series for s in new_series], names=names)
-        return self._from_native_frame(df, validate_column_names=False)
+        return self._from_native_frame(df, validate_column_names=True)
 
     def with_columns(self: Self, *exprs: ArrowExpr) -> Self:
         native_frame = self._native_frame
diff --git a/narwhals/_dask/dataframe.py b/narwhals/_dask/dataframe.py
index 91a2aead..2409d731 100644
--- a/narwhals/_dask/dataframe.py
+++ b/narwhals/_dask/dataframe.py
@@ -10,7 +10,6 @@ import pandas as pd
 
 from narwhals._dask.utils import add_row_index
 from narwhals._dask.utils import evaluate_exprs
-from narwhals._pandas_like.utils import check_column_names_are_unique
 from narwhals._pandas_like.utils import native_to_narwhals_dtype
 from narwhals._pandas_like.utils import select_columns_by_name
 from narwhals.typing import CompliantDataFrame
@@ -41,15 +40,15 @@ class DaskLazyFrame(CompliantLazyFrame):
         *,
         backend_version: tuple[int, ...],
         version: Version,
-        validate_column_names: bool,
+        # Unused, just for compatibility. We only validate when collecting.
+        validate_column_names: bool = False,
     ) -> None:
         self._native_frame: dd.DataFrame = native_dataframe
         self._backend_version = backend_version
         self._implementation = Implementation.DASK
         self._version = version
+        self._cached_schema: dict[str, DType] | None = None
         validate_backend_version(self._implementation, self._backend_version)
-        if validate_column_names:
-            check_column_names_are_unique(native_dataframe.columns)
 
     def __native_namespace__(self: Self) -> ModuleType:
         if self._implementation is Implementation.DASK:
@@ -71,23 +70,19 @@ class DaskLazyFrame(CompliantLazyFrame):
             self._native_frame,
             backend_version=self._backend_version,
             version=version,
-            validate_column_names=False,
         )
 
-    def _from_native_frame(
-        self: Self, df: Any, *, validate_column_names: bool = True
-    ) -> Self:
+    def _from_native_frame(self: Self, df: Any) -> Self:
         return self.__class__(
             df,
             backend_version=self._backend_version,
             version=self._version,
-            validate_column_names=validate_column_names,
         )
 
     def with_columns(self: Self, *exprs: DaskExpr) -> Self:
         df = self._native_frame
         new_series = evaluate_exprs(self, *exprs)
-        df = df.assign(**new_series)
+        df = df.assign(**dict(new_series))
         return self._from_native_frame(df)
 
     def collect(
@@ -107,7 +102,7 @@ class DaskLazyFrame(CompliantLazyFrame):
                 implementation=Implementation.PANDAS,
                 backend_version=parse_version(pd),
                 version=self._version,
-                validate_column_names=False,
+                validate_column_names=True,
             )
 
         if backend is Implementation.POLARS:
@@ -130,7 +125,7 @@ class DaskLazyFrame(CompliantLazyFrame):
                 pa.Table.from_pandas(result),
                 backend_version=parse_version(pa),
                 version=self._version,
-                validate_column_names=False,
+                validate_column_names=True,
             )
 
         msg = f"Unsupported `backend` value: {backend}"  # pragma: no cover
@@ -138,15 +133,13 @@ class DaskLazyFrame(CompliantLazyFrame):
 
     @property
     def columns(self: Self) -> list[str]:
-        return self._native_frame.columns.tolist()
+        return list(self.schema)
 
     def filter(self: Self, predicate: DaskExpr) -> Self:
         # `[0]` is safe as the predicate's expression only returns a single column
         mask = predicate._call(self)[0]
 
-        return self._from_native_frame(
-            self._native_frame.loc[mask], validate_column_names=False
-        )
+        return self._from_native_frame(self._native_frame.loc[mask])
 
     def simple_select(self: Self, *column_names: str) -> Self:
         return self._from_native_frame(
@@ -156,13 +149,12 @@ class DaskLazyFrame(CompliantLazyFrame):
                 self._backend_version,
                 self._implementation,
             ),
-            validate_column_names=False,
         )
 
     def aggregate(self: Self, *exprs: DaskExpr) -> Self:
         new_series = evaluate_exprs(self, *exprs)
-        df = dd.concat([val.rename(name) for name, val in new_series.items()], axis=1)
-        return self._from_native_frame(df, validate_column_names=False)
+        df = dd.concat([val.rename(name) for name, val in new_series], axis=1)
+        return self._from_native_frame(df)
 
     def select(self: Self, *exprs: DaskExpr) -> Self:
         new_series = evaluate_exprs(self, *exprs)
@@ -173,34 +165,33 @@ class DaskLazyFrame(CompliantLazyFrame):
                 dd.from_pandas(
                     pd.DataFrame(), npartitions=self._native_frame.npartitions
                 ),
-                validate_column_names=False,
             )
 
         df = select_columns_by_name(
-            self._native_frame.assign(**new_series),
-            list(new_series.keys()),
+            self._native_frame.assign(**dict(new_series)),
+            [s[0] for s in new_series],
             self._backend_version,
             self._implementation,
         )
-        return self._from_native_frame(df, validate_column_names=False)
+        return self._from_native_frame(df)
 
     def drop_nulls(self: Self, subset: list[str] | None) -> Self:
         if subset is None:
-            return self._from_native_frame(
-                self._native_frame.dropna(), validate_column_names=False
-            )
+            return self._from_native_frame(self._native_frame.dropna())
         plx = self.__narwhals_namespace__()
         return self.filter(~plx.any_horizontal(plx.col(*subset).is_null()))
 
     @property
     def schema(self: Self) -> dict[str, DType]:
-        native_dtypes = self._native_frame.dtypes
-        return {
-            col: native_to_narwhals_dtype(
-                native_dtypes[col], self._version, self._implementation
-            )
-            for col in self._native_frame.columns
-        }
+        if self._cached_schema is None:
+            native_dtypes = self._native_frame.dtypes
+            self._cached_schema = {
+                col: native_to_narwhals_dtype(
+                    native_dtypes[col], self._version, self._implementation
+                )
+                for col in self._native_frame.columns
+            }
+        return self._cached_schema
 
     def collect_schema(self: Self) -> dict[str, DType]:
         return self.schema
@@ -210,9 +201,7 @@ class DaskLazyFrame(CompliantLazyFrame):
             compliant_frame=self, columns=columns, strict=strict
         )
 
-        return self._from_native_frame(
-            self._native_frame.drop(columns=to_drop), validate_column_names=False
-        )
+        return self._from_native_frame(self._native_frame.drop(columns=to_drop))
 
     def with_row_index(self: Self, name: str) -> Self:
         # Implementation is based on the following StackOverflow reply:
@@ -228,8 +217,7 @@ class DaskLazyFrame(CompliantLazyFrame):
 
     def head(self: Self, n: int) -> Self:
         return self._from_native_frame(
-            self._native_frame.head(n=n, compute=False, npartitions=-1),
-            validate_column_names=False,
+            self._native_frame.head(n=n, compute=False, npartitions=-1)
         )
 
     def unique(
@@ -250,7 +238,7 @@ class DaskLazyFrame(CompliantLazyFrame):
         else:
             mapped_keep = {"any": "first"}.get(keep, keep)
             result = native_frame.drop_duplicates(subset=subset, keep=mapped_keep)
-        return self._from_native_frame(result, validate_column_names=False)
+        return self._from_native_frame(result)
 
     def sort(
         self: Self,
@@ -265,8 +253,7 @@ class DaskLazyFrame(CompliantLazyFrame):
             ascending = [not d for d in descending]
         na_position = "last" if nulls_last else "first"
         return self._from_native_frame(
-            df.sort_values(list(by), ascending=ascending, na_position=na_position),
-            validate_column_names=False,
+            df.sort_values(list(by), ascending=ascending, na_position=na_position)
         )
 
     def join(
@@ -413,9 +400,7 @@ class DaskLazyFrame(CompliantLazyFrame):
         n_partitions = native_frame.npartitions
 
         if n_partitions == 1:
-            return self._from_native_frame(
-                self._native_frame.tail(n=n, compute=False), validate_column_names=False
-            )
+            return self._from_native_frame(self._native_frame.tail(n=n, compute=False))
         else:
             msg = "`LazyFrame.tail` is not supported for Dask backend with multiple partitions."
             raise NotImplementedError(msg)
diff --git a/narwhals/_dask/group_by.py b/narwhals/_dask/group_by.py
index 6a972c6a..cc678051 100644
--- a/narwhals/_dask/group_by.py
+++ b/narwhals/_dask/group_by.py
@@ -99,7 +99,6 @@ class DaskLazyGroupBy:
             df,
             backend_version=self._df._backend_version,
             version=self._df._version,
-            validate_column_names=True,
         )
 
 
diff --git a/narwhals/_dask/namespace.py b/narwhals/_dask/namespace.py
index 8522dc86..106ae2b7 100644
--- a/narwhals/_dask/namespace.py
+++ b/narwhals/_dask/namespace.py
@@ -192,7 +192,6 @@ class DaskNamespace(CompliantNamespace[DaskLazyFrame, "dx.Series"]):  # pyright:
                 dd.concat(dfs, axis=0, join="inner"),
                 backend_version=self._backend_version,
                 version=self._version,
-                validate_column_names=True,
             )
         if how == "horizontal":
             all_column_names: list[str] = [
@@ -211,14 +210,12 @@ class DaskNamespace(CompliantNamespace[DaskLazyFrame, "dx.Series"]):  # pyright:
                 dd.concat(dfs, axis=1, join="outer"),
                 backend_version=self._backend_version,
                 version=self._version,
-                validate_column_names=True,
             )
         if how == "diagonal":
             return DaskLazyFrame(
                 dd.concat(dfs, axis=0, join="outer"),
                 backend_version=self._backend_version,
                 version=self._version,
-                validate_column_names=True,
             )
 
         raise NotImplementedError
diff --git a/narwhals/_dask/utils.py b/narwhals/_dask/utils.py
index a98241ec..3bacc328 100644
--- a/narwhals/_dask/utils.py
+++ b/narwhals/_dask/utils.py
@@ -39,20 +39,15 @@ def maybe_evaluate_expr(df: DaskLazyFrame, obj: DaskExpr | object) -> dx.Series
     return obj
 
 
-def evaluate_exprs(df: DaskLazyFrame, /, *exprs: DaskExpr) -> dict[str, dx.Series]:
-    native_results: dict[str, dx.Series] = {}
+def evaluate_exprs(df: DaskLazyFrame, /, *exprs: DaskExpr) -> list[tuple[str, dx.Series]]:
+    native_results: list[tuple[str, dx.Series]] = []
     for expr in exprs:
         native_series_list = expr._call(df)
         _, aliases = evaluate_output_names_and_aliases(expr, df, [])
         if len(aliases) != len(native_series_list):  # pragma: no cover
             msg = f"Internal error: got aliases {aliases}, but only got {len(native_series_list)} results"
             raise AssertionError(msg)
-        native_results.update(
-            {
-                alias: native_series
-                for native_series, alias in zip(native_series_list, aliases)
-            }
-        )
+        native_results.extend(zip(aliases, native_series_list))
     return native_results
 
 
diff --git a/narwhals/_duckdb/dataframe.py b/narwhals/_duckdb/dataframe.py
index b997d364..e194edd4 100644
--- a/narwhals/_duckdb/dataframe.py
+++ b/narwhals/_duckdb/dataframe.py
@@ -9,16 +9,15 @@ import duckdb
 from duckdb import ColumnExpression
 from duckdb import FunctionExpression
 
+from narwhals._duckdb.utils import evaluate_exprs
 from narwhals._duckdb.utils import lit
 from narwhals._duckdb.utils import native_to_narwhals_dtype
-from narwhals._duckdb.utils import parse_exprs
 from narwhals.dependencies import get_duckdb
 from narwhals.exceptions import ColumnNotFoundError
 from narwhals.exceptions import InvalidOperationError
 from narwhals.typing import CompliantDataFrame
 from narwhals.utils import Implementation
 from narwhals.utils import Version
-from narwhals.utils import check_column_names_are_unique
 from narwhals.utils import generate_temporary_column_name
 from narwhals.utils import import_dtypes_module
 from narwhals.utils import parse_columns_to_drop
@@ -50,13 +49,13 @@ class DuckDBLazyFrame(CompliantLazyFrame):
         *,
         backend_version: tuple[int, ...],
         version: Version,
-        validate_column_names: bool,
+        # Unused, just for compatibility. We only validate when collecting.
+        validate_column_names: bool = False,
     ) -> None:
-        if validate_column_names:
-            check_column_names_are_unique(df.columns)
         self._native_frame: duckdb.DuckDBPyRelation = df
         self._version = version
         self._backend_version = backend_version
+        self._cached_schema: dict[str, DType] | None = None
         validate_backend_version(self._implementation, self._backend_version)
 
     def __narwhals_dataframe__(self: Self) -> Self:  # pragma: no cover
@@ -100,7 +99,7 @@ class DuckDBLazyFrame(CompliantLazyFrame):
                 native_dataframe=self._native_frame.arrow(),
                 backend_version=parse_version(pa),
                 version=self._version,
-                validate_column_names=False,
+                validate_column_names=True,
             )
 
         if backend is Implementation.PANDAS:
@@ -113,7 +112,7 @@ class DuckDBLazyFrame(CompliantLazyFrame):
                 implementation=Implementation.PANDAS,
                 backend_version=parse_version(pd),
                 version=self._version,
-                validate_column_names=False,
+                validate_column_names=True,
             )
 
         if backend is Implementation.POLARS:
@@ -131,39 +130,29 @@ class DuckDBLazyFrame(CompliantLazyFrame):
         raise ValueError(msg)  # pragma: no cover
 
     def head(self: Self, n: int) -> Self:
-        return self._from_native_frame(
-            self._native_frame.limit(n), validate_column_names=False
-        )
+        return self._from_native_frame(self._native_frame.limit(n))
 
     def simple_select(self, *column_names: str) -> Self:
-        return self._from_native_frame(
-            self._native_frame.select(*column_names), validate_column_names=False
-        )
+        return self._from_native_frame(self._native_frame.select(*column_names))
 
     def aggregate(self: Self, *exprs: DuckDBExpr) -> Self:
-        new_columns_map = parse_exprs(self, *exprs)
+        new_columns_map = evaluate_exprs(self, *exprs)
         return self._from_native_frame(
             self._native_frame.aggregate(
-                [val.alias(col) for col, val in new_columns_map.items()]  # type: ignore[arg-type]
+                [val.alias(col) for col, val in new_columns_map]  # type: ignore[arg-type]
             ),
-            validate_column_names=False,
         )
 
     def select(
         self: Self,
         *exprs: DuckDBExpr,
     ) -> Self:
-        new_columns_map = parse_exprs(self, *exprs)
+        new_columns_map = evaluate_exprs(self, *exprs)
         if not new_columns_map:
             # TODO(marco): return empty relation with 0 columns?
-            return self._from_native_frame(
-                self._native_frame.limit(0), validate_column_names=False
-            )
+            return self._from_native_frame(self._native_frame.limit(0))
         return self._from_native_frame(
-            self._native_frame.select(
-                *(val.alias(col) for col, val in new_columns_map.items())
-            ),
-            validate_column_names=False,
+            self._native_frame.select(*(val.alias(col) for col, val in new_columns_map)),
         )
 
     def drop(self: Self, columns: list[str], strict: bool) -> Self:  # noqa: FBT001
@@ -171,9 +160,7 @@ class DuckDBLazyFrame(CompliantLazyFrame):
             compliant_frame=self, columns=columns, strict=strict
         )
         selection = (col for col in self.columns if col not in columns_to_drop)
-        return self._from_native_frame(
-            self._native_frame.select(*selection), validate_column_names=False
-        )
+        return self._from_native_frame(self._native_frame.select(*selection))
 
     def lazy(self: Self, *, backend: Implementation | None = None) -> Self:
         # The `backend`` argument has no effect but we keep it here for
@@ -186,7 +173,7 @@ class DuckDBLazyFrame(CompliantLazyFrame):
         return self
 
     def with_columns(self: Self, *exprs: DuckDBExpr) -> Self:
-        new_columns_map = parse_exprs(self, *exprs)
+        new_columns_map = dict(evaluate_exprs(self, *exprs))
         result = [
             new_columns_map.pop(col).alias(col)
             if col in new_columns_map
@@ -194,29 +181,29 @@ class DuckDBLazyFrame(CompliantLazyFrame):
             for col in self._native_frame.columns
         ]
         result.extend(value.alias(col) for col, value in new_columns_map.items())
-        return self._from_native_frame(
-            self._native_frame.select(*result), validate_column_names=False
-        )
+        return self._from_native_frame(self._native_frame.select(*result))
 
     def filter(self: Self, predicate: DuckDBExpr) -> Self:
         # `[0]` is safe as the predicate's expression only returns a single column
         mask = predicate._call(self)[0]
-        return self._from_native_frame(
-            self._native_frame.filter(mask), validate_column_names=False
-        )
+        return self._from_native_frame(self._native_frame.filter(mask))
 
     @property
     def schema(self: Self) -> dict[str, DType]:
-        return {
-            column_name: native_to_narwhals_dtype(str(duckdb_dtype), self._version)
-            for column_name, duckdb_dtype in zip(
-                self._native_frame.columns, self._native_frame.types
-            )
-        }
+        if self._cached_schema is None:
+            # Note: prefer `self._cached_schema` over `functools.cached_property`
+            # due to Python3.13 failures.
+            self._cached_schema = {
+                column_name: native_to_narwhals_dtype(str(duckdb_dtype), self._version)
+                for column_name, duckdb_dtype in zip(
+                    self._native_frame.columns, self._native_frame.types
+                )
+            }
+        return self._cached_schema
 
     @property
     def columns(self: Self) -> list[str]:
-        return self._native_frame.columns
+        return list(self.schema)
 
     def to_pandas(self: Self) -> pd.DataFrame:
         # only if version is v1, keep around for backcompat
@@ -237,17 +224,13 @@ class DuckDBLazyFrame(CompliantLazyFrame):
             self._native_frame,
             version=version,
             backend_version=self._backend_version,
-            validate_column_names=False,
         )
 
-    def _from_native_frame(
-        self: Self, df: duckdb.DuckDBPyRelation, *, validate_column_names: bool = True
-    ) -> Self:
+    def _from_native_frame(self: Self, df: duckdb.DuckDBPyRelation) -> Self:
         return self.__class__(
             df,
             backend_version=self._backend_version,
             version=self._version,
-            validate_column_names=validate_column_names,
         )
 
     def group_by(self: Self, *keys: str, drop_null_keys: bool) -> DuckDBGroupBy:
@@ -387,10 +370,9 @@ class DuckDBLazyFrame(CompliantLazyFrame):
                 )
                 select * exclude ({idx_name}, {count_name}) from cte {keep_condition}
                 """  # noqa: S608
-            return self._from_native_frame(duckdb.sql(query), validate_column_names=False)
+            return self._from_native_frame(duckdb.sql(query))
         return self._from_native_frame(
             self._native_frame.unique(", ".join(self.columns)),
-            validate_column_names=False,
         )
 
     def sort(
@@ -413,14 +395,14 @@ class DuckDBLazyFrame(CompliantLazyFrame):
                 )
             )
         )
-        return self._from_native_frame(result, validate_column_names=False)
+        return self._from_native_frame(result)
 
     def drop_nulls(self: Self, subset: list[str] | None) -> Self:
         rel = self._native_frame
         subset_ = subset if subset is not None else rel.columns
         keep_condition = " and ".join(f'"{col}" is not null' for col in subset_)
         query = f"select * from rel where {keep_condition}"  # noqa: S608
-        return self._from_native_frame(duckdb.sql(query), validate_column_names=False)
+        return self._from_native_frame(duckdb.sql(query))
 
     def explode(self: Self, columns: list[str]) -> Self:
         dtypes = import_dtypes_module(self._version)
@@ -462,9 +444,7 @@ class DuckDBLazyFrame(CompliantLazyFrame):
             *(lit(None).alias(col) if col in columns else col for col in original_columns)
         )
 
-        return self._from_native_frame(
-            non_null_rel.union(null_rel), validate_column_names=False
-        )
+        return self._from_native_frame(non_null_rel.union(null_rel))
 
     def unpivot(
         self: Self,
diff --git a/narwhals/_duckdb/utils.py b/narwhals/_duckdb/utils.py
index 9fbf8084..5fa20527 100644
--- a/narwhals/_duckdb/utils.py
+++ b/narwhals/_duckdb/utils.py
@@ -34,10 +34,10 @@ def maybe_evaluate_expr(
     return duckdb.ConstantExpression(obj)
 
 
-def parse_exprs(
+def evaluate_exprs(
     df: DuckDBLazyFrame, /, *exprs: DuckDBExpr
-) -> dict[str, duckdb.Expression]:
-    native_results: dict[str, duckdb.Expression] = {}
+) -> list[tuple[str, duckdb.Expression]]:
+    native_results: list[tuple[str, duckdb.Expression]] = []
     for expr in exprs:
         native_series_list = expr._call(df)
         output_names = expr._evaluate_output_names(df)
@@ -46,7 +46,7 @@ def parse_exprs(
         if len(output_names) != len(native_series_list):  # pragma: no cover
             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))
+        native_results.extend(zip(output_names, native_series_list))
     return native_results
 
 
diff --git a/narwhals/_expression_parsing.py b/narwhals/_expression_parsing.py
index b567af23..c3323a5f 100644
--- a/narwhals/_expression_parsing.py
+++ b/narwhals/_expression_parsing.py
@@ -331,8 +331,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 | WINDOW) -> raise
+    - FILTRATION vs (LITERAL | AGGREGATION) -> FILTRATION
+    - FILTRATION vs (FILTRATION | TRANSFORM | WINDOW) -> raise
     - (TRANSFORM | WINDOW) vs (LITERAL | AGGREGATION) -> TRANSFORM
     - AGGREGATION vs (LITERAL | AGGREGATION) -> AGGREGATION
     """
@@ -358,7 +358,7 @@ class ExprKind(Enum):
     - `nw.col('a').cum_sum().mean()`
     """
 
-    CHANGES_LENGTH = auto()
+    FILTRATION = auto()
     """e.g. `nw.col('a').drop_nulls()`"""
 
     def preserves_length(self) -> bool:
@@ -367,8 +367,8 @@ class ExprKind(Enum):
     def is_window(self) -> bool:
         return self is ExprKind.WINDOW
 
-    def is_changes_length(self) -> bool:
-        return self is ExprKind.CHANGES_LENGTH
+    def is_filtration(self) -> bool:
+        return self is ExprKind.FILTRATION
 
     def is_scalar_like(self) -> bool:
         return is_scalar_like(self)
@@ -420,7 +420,7 @@ class ExprMetadata:
 def combine_metadata(*args: IntoExpr | object | None, str_as_lit: bool) -> ExprMetadata:
     # Combine metadata from `args`.
 
-    n_changes_length = 0
+    n_filtrations = 0
     has_transforms_or_windows = False
     has_aggregations = False
     has_literals = False
@@ -437,8 +437,8 @@ def combine_metadata(*args: IntoExpr | object | None, str_as_lit: bool) -> ExprM
                 has_aggregations = True
             elif kind is ExprKind.LITERAL:
                 has_literals = True
-            elif kind is ExprKind.CHANGES_LENGTH:
-                n_changes_length += 1
+            elif kind is ExprKind.FILTRATION:
+                n_filtrations += 1
             elif kind.preserves_length():
                 has_transforms_or_windows = True
             else:  # pragma: no cover
@@ -448,17 +448,17 @@ def combine_metadata(*args: IntoExpr | object | None, str_as_lit: bool) -> ExprM
         has_literals
         and not has_aggregations
         and not has_transforms_or_windows
-        and not n_changes_length
+        and not n_filtrations
     ):
         result_kind = ExprKind.LITERAL
-    elif n_changes_length > 1:
+    elif n_filtrations > 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_or_windows:
+    elif n_filtrations 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 n_filtrations:
+        result_kind = ExprKind.FILTRATION
     elif has_transforms_or_windows:
         result_kind = ExprKind.TRANSFORM
     else:
diff --git a/narwhals/_pandas_like/dataframe.py b/narwhals/_pandas_like/dataframe.py
index d716d20d..8d4dc035 100644
--- a/narwhals/_pandas_like/dataframe.py
+++ b/narwhals/_pandas_like/dataframe.py
@@ -403,7 +403,7 @@ class PandasLikeDataFrame(CompliantDataFrame, CompliantLazyFrame):
             implementation=self._implementation,
             backend_version=self._backend_version,
         )
-        return self._from_native_frame(df, validate_column_names=False)
+        return self._from_native_frame(df, validate_column_names=True)
 
     def drop_nulls(self: Self, subset: list[str] | None) -> Self:
         if subset is None:
diff --git a/narwhals/_spark_like/dataframe.py b/narwhals/_spark_like/dataframe.py
index 6ee7a86b..6198835d 100644
--- a/narwhals/_spark_like/dataframe.py
+++ b/narwhals/_spark_like/dataframe.py
@@ -7,14 +7,13 @@ from typing import Any
 from typing import Literal
 from typing import Sequence
 
+from narwhals._spark_like.utils import evaluate_exprs
 from narwhals._spark_like.utils import native_to_narwhals_dtype
-from narwhals._spark_like.utils import parse_exprs
 from narwhals.exceptions import InvalidOperationError
 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
@@ -44,14 +43,14 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
         backend_version: tuple[int, ...],
         version: Version,
         implementation: Implementation,
-        validate_column_names: bool,
+        # Unused, just for compatibility. We only validate when collecting.
+        validate_column_names: bool = False,
     ) -> 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
         self._version = version
+        self._cached_schema: dict[str, DType] | None = None
         validate_backend_version(self._implementation, self._backend_version)
 
     @property
@@ -122,18 +121,14 @@ 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, *, validate_column_names: bool = True
-    ) -> Self:
+    def _from_native_frame(self: Self, df: DataFrame) -> 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:
@@ -184,7 +179,7 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
 
     @property
     def columns(self: Self) -> list[str]:
-        return self._native_frame.columns  # type: ignore[no-any-return]
+        return list(self.schema)
 
     def collect(
         self: Self,
@@ -201,7 +196,7 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
                 implementation=Implementation.PANDAS,
                 backend_version=parse_version(pd),
                 version=self._version,
-                validate_column_names=False,
+                validate_column_names=True,
             )
 
         elif backend is None or backend is Implementation.PYARROW:
@@ -213,7 +208,7 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
                 self._collect_to_arrow(),
                 backend_version=parse_version(pa),
                 version=self._version,
-                validate_column_names=False,
+                validate_column_names=True,
             )
 
         elif backend is Implementation.POLARS:
@@ -232,26 +227,22 @@ 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), validate_column_names=False
-        )
+        return self._from_native_frame(self._native_frame.select(*column_names))
 
     def aggregate(
         self: Self,
         *exprs: SparkLikeExpr,
     ) -> Self:
-        new_columns = parse_exprs(self, *exprs)
+        new_columns = evaluate_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), validate_column_names=False
-        )
+        new_columns_list = [col.alias(col_name) for col_name, col in new_columns]
+        return self._from_native_frame(self._native_frame.agg(*new_columns_list))
 
     def select(
         self: Self,
         *exprs: SparkLikeExpr,
     ) -> Self:
-        new_columns = parse_exprs(self, *exprs)
+        new_columns = evaluate_exprs(self, *exprs)
 
         if not new_columns:
             # return empty dataframe, like Polars does
@@ -259,35 +250,33 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
                 [], self._native_dtypes.StructType([])
             )
 
-            return self._from_native_frame(spark_df, validate_column_names=False)
+            return self._from_native_frame(spark_df)
 
-        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), validate_column_names=False
-        )
+        new_columns_list = [col.alias(col_name) for (col_name, col) in new_columns]
+        return self._from_native_frame(self._native_frame.select(*new_columns_list))
 
     def with_columns(self: Self, *exprs: SparkLikeExpr) -> Self:
-        new_columns = parse_exprs(self, *exprs)
-        return self._from_native_frame(self._native_frame.withColumns(new_columns))
+        new_columns = evaluate_exprs(self, *exprs)
+        return self._from_native_frame(self._native_frame.withColumns(dict(new_columns)))
 
     def filter(self: Self, predicate: SparkLikeExpr) -> Self:
         # `[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, validate_column_names=False)
+        return self._from_native_frame(spark_df)
 
     @property
     def schema(self: Self) -> dict[str, DType]:
-        return {
-            field.name: native_to_narwhals_dtype(
-                dtype=field.dataType,
-                version=self._version,
-                spark_types=self._native_dtypes,
-            )
-            for field in self._native_frame.schema
-        }
+        if self._cached_schema is None:
+            self._cached_schema = {
+                field.name: native_to_narwhals_dtype(
+                    dtype=field.dataType,
+                    version=self._version,
+                    spark_types=self._native_dtypes,
+                )
+                for field in self._native_frame.schema
+            }
+        return self._cached_schema
 
     def collect_schema(self: Self) -> dict[str, DType]:
         return self.schema
@@ -296,14 +285,10 @@ 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), validate_column_names=False
-        )
+        return self._from_native_frame(self._native_frame.drop(*columns_to_drop))
 
     def head(self: Self, n: int) -> Self:
-        return self._from_native_frame(
-            self._native_frame.limit(num=n), validate_column_names=False
-        )
+        return self._from_native_frame(self._native_frame.limit(num=n))
 
     def group_by(self: Self, *keys: str, drop_null_keys: bool) -> SparkLikeLazyGroupBy:
         from narwhals._spark_like.group_by import SparkLikeLazyGroupBy
@@ -333,14 +318,10 @@ 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), validate_column_names=False
-        )
+        return self._from_native_frame(self._native_frame.sort(*sort_cols))
 
     def drop_nulls(self: Self, subset: list[str] | None) -> Self:
-        return self._from_native_frame(
-            self._native_frame.dropna(subset=subset), validate_column_names=False
-        )
+        return self._from_native_frame(self._native_frame.dropna(subset=subset))
 
     def rename(self: Self, mapping: dict[str, str]) -> Self:
         rename_mapping = {
@@ -362,9 +343,7 @@ 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), validate_column_names=False
-        )
+        return self._from_native_frame(self._native_frame.dropDuplicates(subset=subset))
 
     def join(
         self: Self,
@@ -380,11 +359,6 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
         left_columns = self.columns
         right_columns = other.columns
 
-        if isinstance(left_on, str):
-            left_on = [left_on]
-        if isinstance(right_on, str):
-            right_on = [right_on]
-
         # create a mapping for columns on other
         # `right_on` columns will be renamed as `left_on`
         # the remaining columns will be either added the suffix or left unchanged.
@@ -450,7 +424,6 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
                         for col_name in column_names
                     ]
                 ),
-                validate_column_names=False,
             )
         elif self._implementation.is_sqlframe():
             # Not every sqlframe dialect supports `explode_outer` function
@@ -480,7 +453,6 @@ class SparkLikeLazyFrame(CompliantLazyFrame):
                         ]
                     )
                 ),
-                validate_column_names=False,
             )
         else:  # pragma: no cover
             msg = "Unreachable code, please report an issue at https://github.com/narwhals-dev/narwhals/issues"
diff --git a/narwhals/_spark_like/namespace.py b/narwhals/_spark_like/namespace.py
index d64401e0..1b68d073 100644
--- a/narwhals/_spark_like/namespace.py
+++ b/narwhals/_spark_like/namespace.py
@@ -246,7 +246,6 @@ class SparkLikeNamespace(CompliantNamespace["SparkLikeLazyFrame", "Column"]):
                 backend_version=self._backend_version,
                 version=self._version,
                 implementation=self._implementation,
-                validate_column_names=False,
             )
 
         if how == "diagonal":
@@ -257,7 +256,6 @@ 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/_spark_like/utils.py b/narwhals/_spark_like/utils.py
index 566caa02..d262ee1d 100644
--- a/narwhals/_spark_like/utils.py
+++ b/narwhals/_spark_like/utils.py
@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-from functools import lru_cache
 from typing import TYPE_CHECKING
 from typing import Any
 
@@ -20,7 +19,6 @@ if TYPE_CHECKING:
     from narwhals.utils import Version
 
 
-@lru_cache(maxsize=16)
 def native_to_narwhals_dtype(
     dtype: pyspark_types.DataType,
     version: Version,
@@ -145,8 +143,10 @@ def narwhals_to_native_dtype(
     raise AssertionError(msg)
 
 
-def parse_exprs(df: SparkLikeLazyFrame, /, *exprs: SparkLikeExpr) -> dict[str, Column]:
-    native_results: dict[str, list[Column]] = {}
+def evaluate_exprs(
+    df: SparkLikeLazyFrame, /, *exprs: SparkLikeExpr
+) -> list[tuple[str, Column]]:
+    native_results: list[tuple[str, list[Column]]] = []
 
     for expr in exprs:
         native_series_list = expr._call(df)
@@ -156,7 +156,7 @@ def parse_exprs(df: SparkLikeLazyFrame, /, *exprs: SparkLikeExpr) -> dict[str, C
         if len(output_names) != len(native_series_list):  # pragma: no cover
             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))
+        native_results.extend(zip(output_names, native_series_list))
 
     return native_results
 
diff --git a/narwhals/dataframe.py b/narwhals/dataframe.py
index a461a0e5..347f5e77 100644
--- a/narwhals/dataframe.py
+++ b/narwhals/dataframe.py
@@ -2212,7 +2212,7 @@ class LazyFrame(BaseFrame[FrameT]):
                     "  `over` and they will be supported."
                 )
                 raise OrderDependentExprError(msg)
-            if arg._metadata.kind.is_changes_length():
+            if arg._metadata.kind.is_filtration():
                 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 36dc6cda..7d774834 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_extra_open_window(ExprKind.CHANGES_LENGTH),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.FILTRATION),
         )
 
     def skew(self: Self) -> Self:
@@ -869,7 +869,7 @@ class Expr:
         """
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).unique(),
-            self._metadata.with_kind(ExprKind.CHANGES_LENGTH),
+            self._metadata.with_kind(ExprKind.FILTRATION),
         )
 
     def abs(self: Self) -> Self:
@@ -1233,7 +1233,7 @@ class Expr:
                 str_as_lit=False,
             ),
             combine_metadata(self, *flat_predicates, str_as_lit=False).with_kind(
-                ExprKind.CHANGES_LENGTH
+                ExprKind.FILTRATION
             ),
         )
 
@@ -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_extra_open_window(ExprKind.CHANGES_LENGTH),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.FILTRATION),
         )
 
     def fill_null(
@@ -1449,7 +1449,7 @@ class Expr:
         """
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).drop_nulls(),
-            self._metadata.with_kind(ExprKind.CHANGES_LENGTH),
+            self._metadata.with_kind(ExprKind.FILTRATION),
         )
 
     def sample(
@@ -1490,7 +1490,7 @@ class Expr:
             lambda plx: self._to_compliant_expr(plx).sample(
                 n, fraction=fraction, with_replacement=with_replacement, seed=seed
             ),
-            self._metadata.with_kind(ExprKind.CHANGES_LENGTH),
+            self._metadata.with_kind(ExprKind.FILTRATION),
         )
 
     def over(
@@ -1535,7 +1535,7 @@ class Expr:
             |2  4  y                    4|
             └────────────────────────────┘
         """
-        if self._metadata.kind.is_changes_length():
+        if self._metadata.kind.is_filtration():
             msg = "`.over()` can not be used for expressions which change length."
             raise LengthChangingExprError(msg)
         kind = ExprKind.TRANSFORM
@@ -1754,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_extra_open_window(ExprKind.CHANGES_LENGTH),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.FILTRATION),
         )
 
     def tail(self: Self, n: int = 10) -> Self:
@@ -1782,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_extra_open_window(ExprKind.CHANGES_LENGTH),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.FILTRATION),
         )
 
     def round(self: Self, decimals: int = 0) -> Self:
@@ -1877,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_extra_open_window(ExprKind.CHANGES_LENGTH),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.FILTRATION),
         )
 
     # need to allow numeric typing
@@ -1949,7 +1949,7 @@ class Expr:
         """
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).mode(),
-            self._metadata.with_kind(ExprKind.CHANGES_LENGTH),
+            self._metadata.with_kind(ExprKind.FILTRATION),
         )
 
     def is_finite(self: Self) -> Self:
diff --git a/narwhals/expr_cat.py b/narwhals/expr_cat.py
index ca5c1688..4b08d656 100644
--- a/narwhals/expr_cat.py
+++ b/narwhals/expr_cat.py
@@ -65,5 +65,5 @@ class ExprCatNamespace(Generic[ExprT]):
         """
         return self._expr.__class__(
             lambda plx: self._expr._to_compliant_expr(plx).cat.get_categories(),
-            self._expr._metadata.with_kind(ExprKind.CHANGES_LENGTH),
+            self._expr._metadata.with_kind(ExprKind.FILTRATION),
         )
diff --git a/narwhals/stable/v1/__init__.py b/narwhals/stable/v1/__init__.py
index 3d001af3..c43143a0 100644
--- a/narwhals/stable/v1/__init__.py
+++ b/narwhals/stable/v1/__init__.py
@@ -268,7 +268,7 @@ class LazyFrame(NwLazyFrame[IntoFrameT]):
             msg = "Mixing Series with LazyFrame is not supported."
             raise TypeError(msg)
         if isinstance(arg, Expr):
-            # After stable.v1, we raise if arg._is_order_dependent or arg._changes_length
+            # After stable.v1, we raise for order-dependent exprs or filtrations
             return arg._to_compliant_expr(self.__narwhals_namespace__())
         if isinstance(arg, str):
             plx = self.__narwhals_namespace__()
@@ -968,7 +968,7 @@ class Expr(NwExpr):
         """
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).head(n),
-            self._metadata.with_kind_and_extra_open_window(ExprKind.CHANGES_LENGTH),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.FILTRATION),
         )
 
     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_extra_open_window(ExprKind.CHANGES_LENGTH),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.FILTRATION),
         )
 
     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_extra_open_window(ExprKind.CHANGES_LENGTH),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.FILTRATION),
         )
 
     def unique(self: Self, *, maintain_order: bool | None = None) -> Self:
@@ -1019,7 +1019,7 @@ class Expr(NwExpr):
             warn(message=msg, category=UserWarning, stacklevel=find_stacklevel())
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).unique(),
-            self._metadata.with_kind(ExprKind.CHANGES_LENGTH),
+            self._metadata.with_kind(ExprKind.FILTRATION),
         )
 
     def sort(self: Self, *, descending: bool = False, nulls_last: bool = False) -> 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_extra_open_window(ExprKind.CHANGES_LENGTH),
+            self._metadata.with_kind_and_extra_open_window(ExprKind.FILTRATION),
         )
 
     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(ExprKind.CHANGES_LENGTH),
+            self._metadata.with_kind(ExprKind.FILTRATION),
         )
 
 


Test patch:
diff --git a/tests/expr_and_series/over_test.py b/tests/expr_and_series/over_test.py
index 65d4f60e..ba31401b 100644
--- a/tests/expr_and_series/over_test.py
+++ b/tests/expr_and_series/over_test.py
@@ -18,12 +18,14 @@ data = {
     "a": ["a", "a", "b", "b", "b"],
     "b": [1, 2, 3, 5, 3],
     "c": [5, 4, 3, 2, 1],
+    "i": [0, 1, 2, 3, 4],
 }
 
 data_cum = {
     "a": ["a", "a", "b", "b", "b"],
     "b": [1, 2, None, 5, 3],
     "c": [5, 4, 3, 2, 1],
+    "i": [0, 1, 2, 3, 4],
 }
 
 
@@ -41,7 +43,7 @@ def test_over_single(request: pytest.FixtureRequest, constructor: Constructor) -
         "c_max": [5, 5, 3, 3, 3],
     }
 
-    result = df.with_columns(c_max=nw.col("c").max().over("a"))
+    result = df.with_columns(c_max=nw.col("c").max().over("a")).sort("i").drop("i")
     assert_equal_data(result, expected)
 
 
@@ -58,8 +60,13 @@ def test_over_multiple(request: pytest.FixtureRequest, constructor: Constructor)
         "c": [5, 4, 3, 1, 2],
         "c_min": [5, 4, 1, 1, 2],
     }
+    expected = {
+        "a": ["a", "a", "b", "b", "b"],
+        "b": [1, 2, 3, 5, 3],
+        "c": [5, 4, 3, 2, 1],
+    }
 
-    result = df.with_columns(c_min=nw.col("c").min().over("a", "b")).sort("a", "b")
+    result = df.with_columns(c_min=nw.col("c").min().over("a", "b")).sort("i").drop("i")
     assert_equal_data(result, expected)
 
 
@@ -80,7 +87,11 @@ def test_over_cumsum(
         "c_cumsum": [5, 9, 3, 5, 6],
     }
 
-    result = df.with_columns(nw.col("b", "c").cum_sum().over("a").name.suffix("_cumsum"))
+    result = (
+        df.with_columns(nw.col("b", "c").cum_sum().over("a").name.suffix("_cumsum"))
+        .sort("i")
+        .drop("i")
+    )
     assert_equal_data(result, expected)
 
 
@@ -99,8 +110,10 @@ def test_over_cumcount(
         "c_cumcount": [1, 2, 1, 2, 3],
     }
 
-    result = df.with_columns(
-        nw.col("b", "c").cum_count().over("a").name.suffix("_cumcount")
+    result = (
+        df.with_columns(nw.col("b", "c").cum_count().over("a").name.suffix("_cumcount"))
+        .sort("i")
+        .drop("i")
     )
     assert_equal_data(result, expected)
 
@@ -120,7 +133,11 @@ def test_over_cummax(
         "b_cummax": [1, 2, None, 5, 5],
         "c_cummax": [5, 5, 3, 3, 3],
     }
-    result = df.with_columns(nw.col("b", "c").cum_max().over("a").name.suffix("_cummax"))
+    result = (
+        df.with_columns(nw.col("b", "c").cum_max().over("a").name.suffix("_cummax"))
+        .sort("i")
+        .drop("i")
+    )
     assert_equal_data(result, expected)
 
 
@@ -141,7 +158,11 @@ def test_over_cummin(
         "c_cummin": [5, 4, 3, 2, 1],
     }
 
-    result = df.with_columns(nw.col("b", "c").cum_min().over("a").name.suffix("_cummin"))
+    result = (
+        df.with_columns(nw.col("b", "c").cum_min().over("a").name.suffix("_cummin"))
+        .sort("i")
+        .drop("i")
+    )
     assert_equal_data(result, expected)
 
 
@@ -162,8 +183,10 @@ def test_over_cumprod(
         "c_cumprod": [5, 20, 3, 6, 6],
     }
 
-    result = df.with_columns(
-        nw.col("b", "c").cum_prod().over("a").name.suffix("_cumprod")
+    result = (
+        df.with_columns(nw.col("b", "c").cum_prod().over("a").name.suffix("_cumprod"))
+        .sort("i")
+        .drop("i")
     )
     assert_equal_data(result, expected)
 
@@ -238,7 +261,7 @@ def test_over_shift(
         "c": [5, 4, 3, 2, 1],
         "b_shift": [None, None, None, None, 3],
     }
-    result = df.with_columns(b_shift=nw.col("b").shift(2).over("a"))
+    result = df.with_columns(b_shift=nw.col("b").shift(2).over("a")).sort("i").drop("i")
     assert_equal_data(result, expected)
 
 
diff --git a/tests/frame/join_test.py b/tests/frame/join_test.py
index 41a9823d..8fb2997e 100644
--- a/tests/frame/join_test.py
+++ b/tests/frame/join_test.py
@@ -676,7 +676,11 @@ def test_join_duplicate_column_names(
         # need to investigate.
     ):
         request.applymarker(pytest.mark.xfail)
-    if "pyspark" in str(constructor) and "sqlframe" not in str(constructor):
+    if "sqlframe" in str(constructor):
+        import duckdb
+
+        exception: type[Exception] = duckdb.BinderException
+    elif "pyspark" in str(constructor):
         from pyspark.errors import AnalysisException
 
         exception = AnalysisException
@@ -687,4 +691,4 @@ def test_join_duplicate_column_names(
     df = constructor({"a": [1, 2, 3, 4, 5], "b": [6, 6, 6, 6, 6]})
     dfn = nw.from_native(df)
     with pytest.raises(exception):
-        dfn.join(dfn, on=["a"]).join(dfn, on=["a"])  # type: ignore[arg-type]
+        dfn.join(dfn, on=["a"]).join(dfn, on=["a"]).lazy().collect()  # type: ignore[arg-type]
diff --git a/tests/frame/schema_test.py b/tests/frame/schema_test.py
index fc2a4220..f2756cd6 100644
--- a/tests/frame/schema_test.py
+++ b/tests/frame/schema_test.py
@@ -227,7 +227,7 @@ def test_validate_not_duplicated_columns_duckdb() -> None:
     with pytest.raises(
         ValueError, match="Expected unique column names, got:\n- 'a' 2 times"
     ):
-        nw.from_native(rel, eager_only=False)
+        nw.from_native(rel, eager_only=False).lazy().collect()
 
 
 @pytest.mark.skipif(
diff --git a/tests/frame/select_test.py b/tests/frame/select_test.py
index 3ba2110a..5e58b7a2 100644
--- a/tests/frame/select_test.py
+++ b/tests/frame/select_test.py
@@ -143,7 +143,7 @@ def test_alias_invalid(constructor: Constructor) -> None:
         df.lazy().select(nw.all().alias("c")).collect()
 
 
-def test_changes_length_vs_aggregation(constructor_eager: ConstructorEager) -> None:
+def test_filtration_vs_aggregation(constructor_eager: ConstructorEager) -> None:
     df = nw.from_native(constructor_eager({"a": [1, None, 3]}))
     result = df.select(nw.col("a").drop_nulls(), b=nw.col("a").mean())
     expected: dict[str, Any] = {"a": [1, 3], "b": [2.0, 2.0]}
@@ -153,6 +153,12 @@ def test_changes_length_vs_aggregation(constructor_eager: ConstructorEager) -> N
     assert_equal_data(result, expected)
 
 
+def test_select_duplicates(constructor: Constructor) -> None:
+    df = nw.from_native(constructor({"a": [1, 2]})).lazy()
+    with pytest.raises(ValueError, match="Expected unique|duplicate|more than one"):
+        df.select("a", nw.col("a") + 1).collect()
+
+
 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())