Repository: narwhals-dev/narwhals (python)

Issue:
[Bug]: `Expr.mode` inconsistent output when applied to multiple columns
### Describe the bug

When applying expr.mode() on multiple columns (e.g. `df.select(nw.all().mode())`) and one returns one and another multiple values the output is incosistent. This is due to to an issue on the `polars` side of the operation. 

In `polars` this is dealt by using `df.select(nw.all().mode().first())` (https://docs.pola.rs/api/python/dev/reference/expressions/api/polars.Expr.mode.html).

### Steps or code to reproduce the bug

```python

import polars as pl
import pandas as pd
import narwhals as nw
from narwhals.typing import IntoFrame

data = {
    "a": [1, 1, 2, 2, 3],
    "b": [1, 2, 3, 3, 4],
}

df_pl = pl.DataFrame(data)
df_pd = pd.DataFrame(data)

def agnostic_mode(df_native: IntoFrame) -> IntoFrame:
    df = nw.from_native(df_native)
    return df.select(nw.all().mode())

agnostic_mode(df_pd)

agnostic_mode(df_pl)
```

### Expected results

```python
agnostic_mode(df_pd)
┌──────────────────┐
|Narwhals DataFrame|
|------------------|
|        a  b      |
|     0  1  3      |
|     1  2  3      |
└──────────────────┘

agnostic_mode(df_pl)
┌──────────────────┐
|Narwhals DataFrame|
|------------------|
|        a  b      |
|     0  1  3      |
|     1  2  3      |
└──────────────────┘
```

### Actual results

```python
agnostic_mode(df_pd)
┌──────────────────┐
|Narwhals DataFrame|
|------------------|
|        a  b      |
|     0  1  3      |
|     1  2  3      |
└──────────────────┘

agnostic_mode(df_pl)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/kstelios/Projects/narwhals-dev/narwhals/dataframe.py", line 1321, in select
    return super().select(*exprs, **named_exprs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/kstelios/Projects/narwhals-dev/narwhals/dataframe.py", line 180, in select
    self._compliant_frame.select(*compliant_exprs),
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/kstelios/Projects/narwhals-dev/narwhals/_polars/dataframe.py", line 171, in func
    raise catch_polars_exception(e, self._backend_version) from None
narwhals.exceptions.ShapeError: Series b, length 1 doesn't match the DataFrame height of 2

If you want expression: col("b").mode() to be broadcasted, ensure it is a scalar (for instance by adding '.first()').
```

### Please run narwhals.show_version() and enter the output below.

```shell
System:
    python: 3.12.7 (main, Oct 16 2024, 04:37:19) [Clang 18.1.8 ]
executable: /home/kstelios/Projects/narwhals-dev/.venv/bin/python3
   machine: Linux-5.15.167.4-microsoft-standard-WSL2-x86_64-with-glibc2.35

Python dependencies:
     narwhals: 1.31.0
       pandas: 2.2.3
       polars: 1.24.0
         cudf:
        modin: 0.32.0
      pyarrow: 19.0.1
        numpy: 2.2.3
```

### Relevant log output

```shell

```

Pull request description:
fix: Expr.mode broadcasting
closes #2273

This is definitely not THE solution, we need to find a better way of preserving the `_broadcast` attribute in some cases

<!--
# Thanks for contributing a pull request! 
## Please make sure you see our contribution guidelines: https://github.com/narwhals-dev/narwhals/blob/main/CONTRIBUTING.md
-->

## What type of PR is this? (check all applicable)

- [ ] 💾 Refactor
- [ ] ✨ Feature
- [ ] 🐛 Bug Fix
- [ ] 🔧 Optimization
- [ ] 📝 Documentation
- [ ] ✅ Test
- [ ] 🐳 Other

## Related issues

- Related issue #\<issue number\>
- Closes #\<issue number\>

## Checklist

- [ ] Code follows style guide (ruff)
- [ ] Tests added
- [ ] Documented the changes

## If you have comments or can explain your changes, please do so below


Patch:
diff --git a/narwhals/_arrow/series.py b/narwhals/_arrow/series.py
index 08ec2e51..d16d2e9d 100644
--- a/narwhals/_arrow/series.py
+++ b/narwhals/_arrow/series.py
@@ -28,6 +28,7 @@ from narwhals._arrow.utils import native_to_narwhals_dtype
 from narwhals._arrow.utils import nulls_like
 from narwhals._arrow.utils import pad_series
 from narwhals._compliant import EagerSeries
+from narwhals._expression_parsing import ExprKind
 from narwhals.dependencies import is_numpy_array_1d
 from narwhals.exceptions import InvalidOperationError
 from narwhals.utils import Implementation
@@ -131,13 +132,18 @@ class ArrowSeries(EagerSeries["ArrowChunkedArray"]):
     def _from_native_series(
         self: Self,
         series: ArrowArray | ArrowChunkedArray,
+        *,
+        preserve_broadcast: bool = False,
     ) -> Self:
-        return self.__class__(
+        result = self.__class__(
             chunked_array(series),
             name=self._name,
             backend_version=self._backend_version,
             version=self._version,
         )
+        if preserve_broadcast:
+            result._broadcast = self._broadcast
+        return result
 
     @classmethod
     def from_iterable(
@@ -434,12 +440,14 @@ class ArrowSeries(EagerSeries["ArrowChunkedArray"]):
         return self.native.to_numpy()
 
     def alias(self: Self, name: str) -> Self:
-        return self.__class__(
+        result = self.__class__(
             self.native,
             name=name,
             backend_version=self._backend_version,
             version=self._version,
         )
+        result._broadcast = self._broadcast
+        return result
 
     @property
     def dtype(self: Self) -> DType:
@@ -504,14 +512,16 @@ class ArrowSeries(EagerSeries["ArrowChunkedArray"]):
         return self._from_native_series(res)
 
     def is_null(self: Self) -> Self:
-        return self._from_native_series(self.native.is_null())
+        return self._from_native_series(self.native.is_null(), preserve_broadcast=True)
 
     def is_nan(self: Self) -> Self:
-        return self._from_native_series(pc.is_nan(self.native))
+        return self._from_native_series(pc.is_nan(self.native), preserve_broadcast=True)
 
     def cast(self: Self, dtype: DType | type[DType]) -> Self:
         data_type = narwhals_to_native_dtype(dtype, self._version)
-        return self._from_native_series(pc.cast(self.native, data_type))
+        return self._from_native_series(
+            pc.cast(self.native, data_type), preserve_broadcast=True
+        )
 
     def null_count(self: Self, *, _return_py_scalar: bool = True) -> int:
         return maybe_extract_py_scalar(self.native.null_count, _return_py_scalar)
@@ -654,7 +664,7 @@ class ArrowSeries(EagerSeries["ArrowChunkedArray"]):
             series = fill_func(self.native)
         else:
             series = fill_aux(self.native, limit, strategy)
-        return self._from_native_series(series)
+        return self._from_native_series(series, preserve_broadcast=True)
 
     def to_frame(self: Self) -> ArrowDataFrame:
         from narwhals._arrow.dataframe import ArrowDataFrame
@@ -806,10 +816,22 @@ class ArrowSeries(EagerSeries["ArrowChunkedArray"]):
     def clip(
         self: Self, lower_bound: Self | Any | None, upper_bound: Self | Any | None
     ) -> Self:
-        _, lower_bound = extract_native(self, lower_bound)
-        _, upper_bound = extract_native(self, upper_bound)
-        arr = pc.max_element_wise(self.native, lower_bound)
-        return self._from_native_series(pc.min_element_wise(arr, upper_bound))
+        _, lower_bound = (
+            extract_native(self, lower_bound) if lower_bound else (None, None)
+        )
+        _, upper_bound = (
+            extract_native(self, upper_bound) if upper_bound else (None, None)
+        )
+
+        if lower_bound is None:
+            return self._from_native_series(pc.min_element_wise(self.native, upper_bound))
+        if upper_bound is None:
+            return self._from_native_series(pc.max_element_wise(self.native, lower_bound))
+        return self._from_native_series(
+            pc.max_element_wise(
+                pc.min_element_wise(self.native, upper_bound), lower_bound
+            )
+        )
 
     def to_arrow(self: Self) -> ArrowArray:
         return self.native.combine_chunks()
@@ -817,9 +839,13 @@ class ArrowSeries(EagerSeries["ArrowChunkedArray"]):
     def mode(self: Self) -> ArrowSeries:
         plx = self.__narwhals_namespace__()
         col_token = generate_temporary_column_name(n_bytes=8, columns=[self.name])
-        return self.value_counts(
+        counts = self.value_counts(
             name=col_token, normalize=False, sort=False, parallel=False
-        ).filter(plx.col(col_token) == plx.col(col_token).max())[self.name]
+        )
+        return counts.filter(
+            plx.col(col_token)
+            == plx.col(col_token).max().broadcast(kind=ExprKind.AGGREGATION)
+        )[self.name]
 
     def is_finite(self: Self) -> Self:
         return self._from_native_series(pc.is_finite(self.native))
diff --git a/narwhals/_arrow/utils.py b/narwhals/_arrow/utils.py
index ce10365f..7bdfc346 100644
--- a/narwhals/_arrow/utils.py
+++ b/narwhals/_arrow/utils.py
@@ -13,6 +13,7 @@ from typing import overload
 import pyarrow as pa
 import pyarrow.compute as pc
 
+from narwhals.exceptions import ShapeError
 from narwhals.utils import _SeriesNamespace
 from narwhals.utils import import_dtypes_module
 from narwhals.utils import isinstance_or_issubclass
@@ -232,7 +233,7 @@ def extract_native(
     from narwhals._arrow.dataframe import ArrowDataFrame
     from narwhals._arrow.series import ArrowSeries
 
-    if rhs is None:
+    if rhs is None:  # pragma: no cover
         return lhs.native, lit(None, type=lhs._type)
 
     if isinstance(rhs, ArrowDataFrame):
@@ -260,10 +261,9 @@ def align_series_full_broadcast(*series: ArrowSeries) -> Sequence[ArrowSeries]:
     if fast_path:
         return series
 
-    is_max_length_gt_1 = max_length > 1
     reshaped = []
-    for s, length in zip(series, lengths):
-        if is_max_length_gt_1 and length == 1:
+    for s in series:
+        if s._broadcast:
             value = s.native[0]
             if s._backend_version < (13,) and hasattr(value, "as_py"):
                 value = value.as_py()
@@ -271,6 +271,9 @@ def align_series_full_broadcast(*series: ArrowSeries) -> Sequence[ArrowSeries]:
                 s._from_native_series(pa.array([value] * max_length, type=s._type))
             )
         else:
+            if (actual_len := len(s)) != max_length:
+                msg = f"Expected object of length {max_length}, got {actual_len}."
+                raise ShapeError(msg)
             reshaped.append(s)
 
     return reshaped
diff --git a/narwhals/_compliant/expr.py b/narwhals/_compliant/expr.py
index b8dcca8c..7cc4db4c 100644
--- a/narwhals/_compliant/expr.py
+++ b/narwhals/_compliant/expr.py
@@ -882,8 +882,6 @@ class LazyExpr(
     sample: not_implemented = not_implemented()
     map_batches: not_implemented = not_implemented()
     ewm_mean: not_implemented = not_implemented()
-    rolling_var: not_implemented = not_implemented()
-    rolling_std: not_implemented = not_implemented()
     gather_every: not_implemented = not_implemented()
     replace_strict: not_implemented = not_implemented()
     cat: not_implemented = not_implemented()  # pyright: ignore[reportAssignmentType]
diff --git a/narwhals/_compliant/series.py b/narwhals/_compliant/series.py
index 4e1ef8f3..6d4ed559 100644
--- a/narwhals/_compliant/series.py
+++ b/narwhals/_compliant/series.py
@@ -280,6 +280,19 @@ class EagerSeries(CompliantSeries[NativeSeriesT_co], Protocol[NativeSeriesT_co])
     def _from_scalar(self, value: Any) -> Self:
         return self.from_iterable([value], name=self.name, context=self)
 
+    def _from_native_series(
+        self, series: Any, *, preserve_broadcast: bool = False
+    ) -> Self:
+        """Return a new `CompliantSeries`, wrapping the native `series`.
+
+        In cases when operations are known to not affect whether a result should
+        be broadcast, we can pass `preverse_broadcast=True`.
+        Set this with care - it should only be set for unary expressions which don't
+        change length or order, such as `.alias` or `.fill_null`. If in doubt, don't
+        set it, you probably don't need it.
+        """
+        ...
+
     def __narwhals_namespace__(self) -> EagerNamespace[Any, Self, Any]: ...
 
     def _to_expr(self) -> EagerExpr[Any, Any]:
diff --git a/narwhals/_dask/expr.py b/narwhals/_dask/expr.py
index bbd659c9..324705b0 100644
--- a/narwhals/_dask/expr.py
+++ b/narwhals/_dask/expr.py
@@ -402,6 +402,34 @@ class DaskExpr(
             "rolling_mean",
         )
 
+    def rolling_var(
+        self: Self, window_size: int, *, min_samples: int, center: bool, ddof: int
+    ) -> Self:
+        if ddof == 1:
+            return self._from_call(
+                lambda _input: _input.rolling(
+                    window=window_size, min_periods=min_samples, center=center
+                ).var(),
+                "rolling_var",
+            )
+        else:
+            msg = "Dask backend only supports `ddof=1` for `rolling_var`"
+            raise NotImplementedError(msg)
+
+    def rolling_std(
+        self: Self, window_size: int, *, min_samples: int, center: bool, ddof: int
+    ) -> Self:
+        if ddof == 1:
+            return self._from_call(
+                lambda _input: _input.rolling(
+                    window=window_size, min_periods=min_samples, center=center
+                ).std(),
+                "rolling_std",
+            )
+        else:
+            msg = "Dask backend only supports `ddof=1` for `rolling_std`"
+            raise NotImplementedError(msg)
+
     def sum(self: Self) -> Self:
         return self._from_call(lambda _input: _input.sum().to_series(), "sum")
 
diff --git a/narwhals/_duckdb/expr.py b/narwhals/_duckdb/expr.py
index 6f1fc24c..685ca947 100644
--- a/narwhals/_duckdb/expr.py
+++ b/narwhals/_duckdb/expr.py
@@ -105,7 +105,9 @@ class DuckDBExpr(LazyExpr["DuckDBLazyFrame", "duckdb.Expression"]):
         center: bool,
         window_size: int,
         min_samples: int,
+        ddof: int | None = None,
     ) -> WindowFunction:
+        supported_funcs = ["sum", "mean", "std", "var"]
         if center:
             half = (window_size - 1) // 2
             remainder = (window_size - 1) % 2
@@ -119,9 +121,25 @@ class DuckDBExpr(LazyExpr["DuckDBLazyFrame", "duckdb.Expression"]):
             order_by_sql = generate_order_by_sql(*window_inputs.order_by, ascending=True)
             partition_by_sql = generate_partition_by_sql(*window_inputs.partition_by)
             window = f"({partition_by_sql} {order_by_sql} rows between {start} and {end})"
+            if func_name in {"sum", "mean"}:
+                func_: str = func_name
+            elif func_name == "var" and ddof == 0:
+                func_ = "var_pop"
+            elif func_name in "var" and ddof == 1:
+                func_ = "var_samp"
+            elif func_name == "std" and ddof == 0:
+                func_ = "stddev_pop"
+            elif func_name == "std" and ddof == 1:
+                func_ = "stddev_samp"
+            elif func_name in {"var", "std"}:  # pragma: no cover
+                msg = f"Only ddof=0 and ddof=1 are currently supported for rolling_{func_name}."
+                raise ValueError(msg)
+            else:  # pragma: no cover
+                msg = f"Only the following functions are supported: {supported_funcs}.\nGot: {func_name}."
+                raise ValueError(msg)
             sql = (
                 f"case when count({window_inputs.expr}) over {window} >= {min_samples}"
-                f"then {func_name}({window_inputs.expr}) over {window} else null end"
+                f"then {func_}({window_inputs.expr}) over {window} else null end"
             )
             return SQLExpression(sql)  # type: ignore[no-any-return, unused-ignore]
 
@@ -368,14 +386,26 @@ class DuckDBExpr(LazyExpr["DuckDBLazyFrame", "duckdb.Expression"]):
         return self._from_call(func)
 
     def clip(self: Self, lower_bound: Any, upper_bound: Any) -> Self:
-        def func(
+        def _clip_lower(_input: duckdb.Expression, lower_bound: Any) -> duckdb.Expression:
+            return FunctionExpression("greatest", _input, lower_bound)
+
+        def _clip_upper(_input: duckdb.Expression, upper_bound: Any) -> duckdb.Expression:
+            return FunctionExpression("least", _input, upper_bound)
+
+        def _clip_both(
             _input: duckdb.Expression, lower_bound: Any, upper_bound: Any
         ) -> duckdb.Expression:
             return FunctionExpression(
                 "greatest", FunctionExpression("least", _input, upper_bound), lower_bound
             )
 
-        return self._from_call(func, lower_bound=lower_bound, upper_bound=upper_bound)
+        if lower_bound is None:
+            return self._from_call(_clip_upper, upper_bound=upper_bound)
+        if upper_bound is None:
+            return self._from_call(_clip_lower, lower_bound=lower_bound)
+        return self._from_call(
+            _clip_both, lower_bound=lower_bound, upper_bound=upper_bound
+        )
 
     def sum(self: Self) -> Self:
         return self._from_call(lambda _input: FunctionExpression("sum", _input))
@@ -594,6 +624,32 @@ class DuckDBExpr(LazyExpr["DuckDBLazyFrame", "duckdb.Expression"]):
             )
         )
 
+    def rolling_var(
+        self, window_size: int, *, min_samples: int, center: bool, ddof: int
+    ) -> Self:
+        return self._with_window_function(
+            self._rolling_window_func(
+                func_name="var",
+                center=center,
+                window_size=window_size,
+                min_samples=min_samples,
+                ddof=ddof,
+            )
+        )
+
+    def rolling_std(
+        self, window_size: int, *, min_samples: int, center: bool, ddof: int
+    ) -> Self:
+        return self._with_window_function(
+            self._rolling_window_func(
+                func_name="std",
+                center=center,
+                window_size=window_size,
+                min_samples=min_samples,
+                ddof=ddof,
+            )
+        )
+
     def fill_null(
         self: Self, value: Self | Any, strategy: Any, limit: int | None
     ) -> Self:
diff --git a/narwhals/_expression_parsing.py b/narwhals/_expression_parsing.py
index abc8b7a7..d10f0907 100644
--- a/narwhals/_expression_parsing.py
+++ b/narwhals/_expression_parsing.py
@@ -421,7 +421,7 @@ def apply_n_ary_operation(
     )
     kinds = [infer_kind(comparand, str_as_lit=str_as_lit) for comparand in comparands]
 
-    broadcast = any(kind.preserves_length() for kind in kinds)
+    broadcast = any(not kind.is_scalar_like() for kind in kinds)
     compliant_exprs = (
         compliant_expr.broadcast(kind)
         if broadcast and is_compliant_expr(compliant_expr) and is_scalar_like(kind)
diff --git a/narwhals/_pandas_like/series.py b/narwhals/_pandas_like/series.py
index e5a237be..1151a3ee 100644
--- a/narwhals/_pandas_like/series.py
+++ b/narwhals/_pandas_like/series.py
@@ -158,13 +158,18 @@ class PandasLikeSeries(EagerSeries[Any]):
             version=version,
         )
 
-    def _from_native_series(self: Self, series: Any) -> Self:
-        return self.__class__(
+    def _from_native_series(
+        self: Self, series: Any, *, preserve_broadcast: bool = False
+    ) -> Self:
+        result = self.__class__(
             series,
             implementation=self._implementation,
             backend_version=self._backend_version,
             version=self._version,
         )
+        if preserve_broadcast:
+            result._broadcast = self._broadcast
+        return result
 
     @classmethod
     def from_iterable(
@@ -284,7 +289,9 @@ class PandasLikeSeries(EagerSeries[Any]):
             backend_version=self._backend_version,
             version=self._version,
         )
-        return self._from_native_series(self.native.astype(pd_dtype))
+        return self._from_native_series(
+            self.native.astype(pd_dtype), preserve_broadcast=True
+        )
 
     def item(self: Self, index: int | None) -> Any:
         # cuDF doesn't have Series.item().
@@ -540,12 +547,12 @@ class PandasLikeSeries(EagerSeries[Any]):
     # Transformations
 
     def is_null(self: Self) -> PandasLikeSeries:
-        return self._from_native_series(self.native.isna())
+        return self._from_native_series(self.native.isna(), preserve_broadcast=True)
 
     def is_nan(self: Self) -> PandasLikeSeries:
         ser = self.native
         if self.dtype.is_numeric():
-            return self._from_native_series(ser != ser)  # noqa: PLR0124
+            return self._from_native_series(ser != ser, preserve_broadcast=True)  # noqa: PLR0124
         msg = f"`.is_nan` only supported for numeric dtype and not {self.dtype}, did you mean `.is_null`?"
         raise InvalidOperationError(msg)
 
@@ -558,12 +565,15 @@ class PandasLikeSeries(EagerSeries[Any]):
         ser = self.native
         if value is not None:
             _, value = align_and_extract_native(self, value)
-            res_ser = self._from_native_series(ser.fillna(value=value))
+            res_ser = self._from_native_series(
+                ser.fillna(value=value), preserve_broadcast=True
+            )
         else:
             res_ser = self._from_native_series(
                 ser.ffill(limit=limit)
                 if strategy == "forward"
-                else ser.bfill(limit=limit)
+                else ser.bfill(limit=limit),
+                preserve_broadcast=True,
             )
 
         return res_ser
@@ -661,7 +671,8 @@ class PandasLikeSeries(EagerSeries[Any]):
                     name,
                     implementation=self._implementation,
                     backend_version=self._backend_version,
-                )
+                ),
+                preserve_broadcast=True,
             )
         return self
 
@@ -839,8 +850,12 @@ class PandasLikeSeries(EagerSeries[Any]):
     def clip(
         self: Self, lower_bound: Self | Any | None, upper_bound: Self | Any | None
     ) -> Self:
-        _, lower_bound = align_and_extract_native(self, lower_bound)
-        _, upper_bound = align_and_extract_native(self, upper_bound)
+        _, lower_bound = (
+            align_and_extract_native(self, lower_bound) if lower_bound else (None, None)
+        )
+        _, upper_bound = (
+            align_and_extract_native(self, upper_bound) if upper_bound else (None, None)
+        )
         kwargs = {"axis": 0} if self._implementation is Implementation.MODIN else {}
         return self._from_native_series(
             self.native.clip(lower_bound, upper_bound, **kwargs)
diff --git a/narwhals/_pandas_like/utils.py b/narwhals/_pandas_like/utils.py
index fbeeca1d..d4069611 100644
--- a/narwhals/_pandas_like/utils.py
+++ b/narwhals/_pandas_like/utils.py
@@ -15,6 +15,7 @@ import pandas as pd
 
 from narwhals.exceptions import ColumnNotFoundError
 from narwhals.exceptions import DuplicateError
+from narwhals.exceptions import ShapeError
 from narwhals.utils import Implementation
 from narwhals.utils import Version
 from narwhals.utils import import_dtypes_module
@@ -249,6 +250,11 @@ def set_index(
 
     We can set `copy` / `inplace` based on implementation/version.
     """
+    if isinstance(index, implementation.to_native_namespace().Index) and (
+        expected_len := len(index)
+    ) != (actual_len := len(obj)):
+        msg = f"Expected object of length {expected_len}, got length: {actual_len}"
+        raise ShapeError(msg)
     if implementation is Implementation.CUDF:  # pragma: no cover
         obj = obj.copy(deep=False)  # type: ignore[attr-defined]
         obj.index = index  # type: ignore[attr-defined]
@@ -611,9 +617,8 @@ def align_series_full_broadcast(
 
     idx = series[lengths.index(max_length)].native.index
     reindexed = []
-    max_length_gt_1 = max_length > 1
-    for s, length in zip(series, lengths):
-        if max_length_gt_1 and length == 1:
+    for s in series:
+        if s._broadcast:
             reindexed.append(
                 s._from_native_series(
                     native_namespace.Series(
diff --git a/narwhals/_spark_like/expr.py b/narwhals/_spark_like/expr.py
index 4e4f85eb..e2bfe7ec 100644
--- a/narwhals/_spark_like/expr.py
+++ b/narwhals/_spark_like/expr.py
@@ -166,7 +166,9 @@ class SparkLikeExpr(LazyExpr["SparkLikeLazyFrame", "Column"]):
         center: bool,
         window_size: int,
         min_samples: int,
+        ddof: int | None = None,
     ) -> WindowFunction:
+        supported_funcs = ["sum", "mean", "std", "var"]
         if center:
             half = (window_size - 1) // 2
             remainder = (window_size - 1) % 2
@@ -185,9 +187,25 @@ class SparkLikeExpr(LazyExpr["SparkLikeLazyFrame", "Column"]):
                 )
                 .rowsBetween(start, end)
             )
+            if func_name in {"sum", "mean"}:
+                func_: str = func_name
+            elif func_name == "var" and ddof == 0:
+                func_ = "var_pop"
+            elif func_name in "var" and ddof == 1:
+                func_ = "var_samp"
+            elif func_name == "std" and ddof == 0:
+                func_ = "stddev_pop"
+            elif func_name == "std" and ddof == 1:
+                func_ = "stddev_samp"
+            elif func_name in {"var", "std"}:  # pragma: no cover
+                msg = f"Only ddof=0 and ddof=1 are currently supported for rolling_{func_name}."
+                raise ValueError(msg)
+            else:  # pragma: no cover
+                msg = f"Only the following functions are supported: {supported_funcs}.\nGot: {func_name}."
+                raise ValueError(msg)
             return self._F.when(
                 self._F.count(window_inputs.expr).over(window) >= min_samples,
-                getattr(self._F, func_name)(window_inputs.expr).over(window),
+                getattr(self._F, func_)(window_inputs.expr).over(window),
             )
 
         return func
@@ -454,9 +472,11 @@ class SparkLikeExpr(LazyExpr["SparkLikeLazyFrame", "Column"]):
         def _clip_both(
             _input: Column, lower_bound: Column, upper_bound: Column
         ) -> Column:
-            result = _input
-            result = self._F.when(result < lower_bound, lower_bound).otherwise(result)
-            return self._F.when(result > upper_bound, upper_bound).otherwise(result)
+            return (
+                self._F.when(_input < lower_bound, lower_bound)
+                .when(_input > upper_bound, upper_bound)
+                .otherwise(_input)
+            )
 
         if lower_bound is None:
             return self._from_call(_clip_upper, upper_bound=upper_bound)
@@ -675,6 +695,32 @@ class SparkLikeExpr(LazyExpr["SparkLikeLazyFrame", "Column"]):
             )
         )
 
+    def rolling_var(
+        self, window_size: int, *, min_samples: int, center: bool, ddof: int
+    ) -> Self:
+        return self._with_window_function(
+            self._rolling_window_func(
+                func_name="var",
+                center=center,
+                window_size=window_size,
+                min_samples=min_samples,
+                ddof=ddof,
+            )
+        )
+
+    def rolling_std(
+        self, window_size: int, *, min_samples: int, center: bool, ddof: int
+    ) -> Self:
+        return self._with_window_function(
+            self._rolling_window_func(
+                func_name="std",
+                center=center,
+                window_size=window_size,
+                min_samples=min_samples,
+                ddof=ddof,
+            )
+        )
+
     @property
     def str(self: Self) -> SparkLikeExprStringNamespace:
         return SparkLikeExprStringNamespace(self)
diff --git a/narwhals/functions.py b/narwhals/functions.py
index a91d9fd8..1f223e30 100644
--- a/narwhals/functions.py
+++ b/narwhals/functions.py
@@ -904,6 +904,20 @@ def _scan_csv_impl(
         from pyarrow import csv  # ignore-banned-import
 
         native_frame = csv.read_csv(source, **kwargs)
+    elif implementation.is_spark_like():
+        if (session := kwargs.pop("session", None)) is None:
+            msg = "Spark like backends require a session object to be passed in `kwargs`."
+            raise ValueError(msg)
+
+        csv_reader = session.read.format("csv")
+        native_frame = (
+            csv_reader.load(source)
+            if (
+                implementation is Implementation.SQLFRAME
+                and parse_version(version("sqlframe")) < (3, 27, 0)
+            )
+            else csv_reader.options(**kwargs).load(source)
+        )
     else:  # pragma: no cover
         try:
             # implementation is UNKNOWN, Narwhals extension using this feature should
@@ -1100,13 +1114,15 @@ def _scan_parquet_impl(
         if (session := kwargs.pop("session", None)) is None:
             msg = "Spark like backends require a session object to be passed in `kwargs`."
             raise ValueError(msg)
+
+        pq_reader = session.read.format("parquet")
         native_frame = (
-            session.read.format("parquet").load(source)
+            pq_reader.load(source)
             if (
                 implementation is Implementation.SQLFRAME
-                and (parse_version(version("sqlframe"))) < (3, 27, 0)
+                and parse_version(version("sqlframe")) < (3, 27, 0)
             )
-            else session.read.format("parquet").options(**kwargs).load(source)
+            else pq_reader.options(**kwargs).load(source)
         )
 
     else:  # pragma: no cover


Test patch:
diff --git a/tests/expr_and_series/mode_test.py b/tests/expr_and_series/mode_test.py
index c20a57c6..a18c8292 100644
--- a/tests/expr_and_series/mode_test.py
+++ b/tests/expr_and_series/mode_test.py
@@ -3,6 +3,8 @@ from __future__ import annotations
 import pytest
 
 import narwhals.stable.v1 as nw
+from narwhals.exceptions import ShapeError
+from tests.utils import POLARS_VERSION
 from tests.utils import ConstructorEager
 from tests.utils import assert_equal_data
 
@@ -12,25 +14,23 @@ data = {
 }
 
 
-def test_mode_single_expr(
-    constructor_eager: ConstructorEager, request: pytest.FixtureRequest
-) -> None:
-    if "pyarrow_table" in str(constructor_eager):
-        # TODO(unassigned): reimplement mode for pyarrow
-        request.applymarker(pytest.mark.xfail)
+def test_mode_single_expr(constructor_eager: ConstructorEager) -> None:
     df = nw.from_native(constructor_eager(data))
     result = df.select(nw.col("a").mode()).sort("a")
     expected = {"a": [1, 2]}
     assert_equal_data(result, expected)
 
 
-def test_mode_series(
-    constructor_eager: ConstructorEager, request: pytest.FixtureRequest
-) -> None:
-    if "pyarrow_table" in str(constructor_eager):
-        # TODO(unassigned): reimplement mode for pyarrow
-        request.applymarker(pytest.mark.xfail)
+def test_mode_series(constructor_eager: ConstructorEager) -> None:
     series = nw.from_native(constructor_eager(data), eager_only=True)["a"]
     result = series.mode().sort()
     expected = {"a": [1, 2]}
     assert_equal_data({"a": result}, expected)
+
+
+def test_mode_different_lengths(constructor_eager: ConstructorEager) -> None:
+    if "polars" in str(constructor_eager) and POLARS_VERSION < (1, 10):
+        pytest.skip()
+    df = nw.from_native(constructor_eager({"a": [1, 1, 2], "b": [4, 5, 6]}))
+    with pytest.raises(ShapeError):
+        df.select(nw.col("a", "b").mode())
diff --git a/tests/expr_and_series/rolling_std_test.py b/tests/expr_and_series/rolling_std_test.py
index 8d47ba22..4c352313 100644
--- a/tests/expr_and_series/rolling_std_test.py
+++ b/tests/expr_and_series/rolling_std_test.py
@@ -6,7 +6,9 @@ from typing import Any
 import pytest
 
 import narwhals.stable.v1 as nw
+from tests.utils import DUCKDB_VERSION
 from tests.utils import POLARS_VERSION
+from tests.utils import Constructor
 from tests.utils import ConstructorEager
 from tests.utils import assert_equal_data
 
@@ -105,3 +107,247 @@ def test_rolling_std_series(
     result = df.select(df["a"].rolling_std(**kwargs).alias(name))
 
     assert_equal_data(result, {name: expected})
+
+
+@pytest.mark.filterwarnings(
+    "ignore:`Expr.rolling_std` is being called from the stable API although considered an unstable feature."
+)
+@pytest.mark.parametrize(
+    ("expected_a", "window_size", "min_samples", "center", "ddof"),
+    [
+        ([None, None, 0.5, None, None, 1, 2.5], 2, None, False, 0),
+        (
+            [
+                None,
+                None,
+                0.7071067811865476,
+                None,
+                None,
+                1.4142135623730951,
+                3.5355339059327378,
+            ],
+            2,
+            2,
+            False,
+            1,
+        ),
+        (
+            [
+                None,
+                None,
+                0.7071067811865476,
+                0.7071067811865476,
+                1.4142135623730951,
+                1.4142135623730951,
+                3.605551275463989,
+            ],
+            3,
+            2,
+            False,
+            1,
+        ),
+        ([0.0, None, 0.5, 0.5, 1.0, 1.0, 2.943920288775949], 3, 1, False, 0),
+        (
+            [
+                0.7071067811865476,
+                None,
+                0.7071067811865476,
+                1.4142135623730951,
+                1.4142135623730951,
+                3.605551275463989,
+                3.5355339059327378,
+            ],
+            3,
+            1,
+            True,
+            1,
+        ),
+        (
+            [
+                0.7071067811865476,
+                None,
+                0.7071067811865476,
+                1.5275252316519465,
+                2.0,
+                3.605551275463989,
+                3.605551275463989,
+            ],
+            4,
+            1,
+            True,
+            1,
+        ),
+        (
+            [
+                0.5,
+                0.5,
+                1.247219128924647,
+                1.920286436967152,
+                3.344772040064913,
+                2.943920288775949,
+                2.943920288775949,
+            ],
+            5,
+            1,
+            True,
+            0,
+        ),
+    ],
+)
+def test_rolling_std_expr_lazy_ungrouped(
+    constructor: Constructor,
+    expected_a: list[float],
+    window_size: int,
+    min_samples: int,
+    *,
+    center: bool,
+    ddof: int,
+) -> None:
+    if ("polars" in str(constructor) and POLARS_VERSION < (1, 10)) or (
+        "duckdb" in str(constructor) and DUCKDB_VERSION < (1, 3)
+    ):
+        pytest.skip()
+    if "modin" in str(constructor):
+        # unreliable
+        pytest.skip()
+    if "dask" in str(constructor) and ddof != 1:
+        # Only `ddof=1` is supported
+        pytest.skip()
+    data = {
+        "a": [1, None, 2, None, 4, 6, 11],
+        "b": [1, None, 2, 3, 4, 5, 6],
+        "i": list(range(7)),
+    }
+    df = nw.from_native(constructor(data))
+    result = (
+        df.with_columns(
+            nw.col("a")
+            .rolling_std(window_size, min_samples=min_samples, center=center, ddof=ddof)
+            .over(order_by="b")
+        )
+        .select("a", "i")
+        .sort("i")
+    )
+    expected = {"a": expected_a, "i": list(range(7))}
+    assert_equal_data(result, expected)
+
+
+@pytest.mark.filterwarnings(
+    "ignore:`Expr.rolling_std` is being called from the stable API although considered an unstable feature."
+)
+@pytest.mark.parametrize(
+    ("expected_a", "window_size", "min_samples", "center", "ddof"),
+    [
+        ([None, None, 0.5, None, None, 1.0, 2.5], 2, None, False, 0),
+        (
+            [
+                None,
+                None,
+                0.7071067811865476,
+                None,
+                None,
+                1.4142135623730951,
+                3.5355339059327378,
+            ],
+            2,
+            2,
+            False,
+            1,
+        ),
+        (
+            [
+                None,
+                None,
+                0.7071067811865476,
+                0.7071067811865476,
+                None,
+                1.4142135623730951,
+                3.605551275463989,
+            ],
+            3,
+            2,
+            False,
+            1,
+        ),
+        ([0.0, None, 0.5, 0.5, 0.0, 1.0, 2.943920288775949], 3, 1, False, 0),
+        (
+            [
+                0.7071067811865476,
+                None,
+                0.7071067811865476,
+                None,
+                1.4142135623730951,
+                3.605551275463989,
+                3.5355339059327378,
+            ],
+            3,
+            1,
+            True,
+            1,
+        ),
+        (
+            [
+                0.7071067811865476,
+                None,
+                0.7071067811865476,
+                0.7071067811865476,
+                1.4142135623730951,
+                3.605551275463989,
+                3.605551275463989,
+            ],
+            4,
+            1,
+            True,
+            1,
+        ),
+        (
+            [0.5, 0.5, 0.5, 0.5, 2.943920288775949, 2.943920288775949, 2.943920288775949],
+            5,
+            1,
+            True,
+            0,
+        ),
+    ],
+)
+def test_rolling_std_expr_lazy_grouped(
+    constructor: Constructor,
+    expected_a: list[float],
+    window_size: int,
+    min_samples: int,
+    request: pytest.FixtureRequest,
+    *,
+    center: bool,
+    ddof: int,
+) -> None:
+    if ("polars" in str(constructor) and POLARS_VERSION < (1, 10)) or (
+        "duckdb" in str(constructor) and DUCKDB_VERSION < (1, 3)
+    ):
+        pytest.skip()
+    if "pandas" in str(constructor):
+        pytest.skip()
+    if any(x in str(constructor) for x in ("dask", "pyarrow_table")):
+        request.applymarker(pytest.mark.xfail)
+    if "cudf" in str(constructor) and center:
+        # center is not implemented for offset-based windows
+        request.applymarker(pytest.mark.xfail)
+    if "modin" in str(constructor):
+        # unreliable
+        pytest.skip()
+    data = {
+        "a": [1, None, 2, None, 4, 6, 11],
+        "g": [1, 1, 1, 1, 2, 2, 2],
+        "b": [1, None, 2, 3, 4, 5, 6],
+        "i": list(range(7)),
+    }
+    df = nw.from_native(constructor(data))
+    result = (
+        df.with_columns(
+            nw.col("a")
+            .rolling_std(window_size, min_samples=min_samples, center=center, ddof=ddof)
+            .over("g", order_by="b")
+        )
+        .sort("i")
+        .select("a")
+    )
+    expected = {"a": expected_a}
+    assert_equal_data(result, expected)
diff --git a/tests/expr_and_series/rolling_var_test.py b/tests/expr_and_series/rolling_var_test.py
index e0a3827e..09383e7d 100644
--- a/tests/expr_and_series/rolling_var_test.py
+++ b/tests/expr_and_series/rolling_var_test.py
@@ -9,8 +9,10 @@ import pytest
 from hypothesis import given
 
 import narwhals.stable.v1 as nw
+from tests.utils import DUCKDB_VERSION
 from tests.utils import PANDAS_VERSION
 from tests.utils import POLARS_VERSION
+from tests.utils import Constructor
 from tests.utils import ConstructorEager
 from tests.utils import assert_equal_data
 
@@ -171,3 +173,143 @@ def test_rolling_var_hypothesis_polars(center: bool, values: list[float]) -> Non
     )
     expected_dict = nw.from_native(expected, eager_only=True).to_dict(as_series=False)
     assert_equal_data(result, expected_dict)
+
+
+@pytest.mark.filterwarnings(
+    "ignore:`Expr.rolling_var` is being called from the stable API although considered an unstable feature."
+)
+@pytest.mark.parametrize(
+    ("expected_a", "window_size", "min_samples", "center", "ddof"),
+    [
+        ([None, None, 0.25, None, None, 1, 6.25], 2, None, False, 0),
+        ([None, None, 0.5, None, None, 2, 12.5], 2, 2, False, 1),
+        ([None, None, 0.5, 0.5, 2, 2, 13], 3, 2, False, 1),
+        ([0, None, 0.25, 0.25, 1, 1, 8.666666666666666], 3, 1, False, 0),
+        ([0.5, None, 0.5, 2, 2, 13, 12.5], 3, 1, True, 1),
+        ([0.5, None, 0.5, 2.333333333333333, 4, 13, 13], 4, 1, True, 1),
+        (
+            [
+                0.25,
+                0.25,
+                1.5555555555555554,
+                3.6874999999999996,
+                11.1875,
+                8.666666666666666,
+                8.666666666666666,
+            ],
+            5,
+            1,
+            True,
+            0,
+        ),
+    ],
+)
+def test_rolling_var_expr_lazy_ungrouped(
+    constructor: Constructor,
+    expected_a: list[float],
+    window_size: int,
+    min_samples: int,
+    *,
+    center: bool,
+    ddof: int,
+) -> None:
+    if ("polars" in str(constructor) and POLARS_VERSION < (1, 10)) or (
+        "duckdb" in str(constructor) and DUCKDB_VERSION < (1, 3)
+    ):
+        pytest.skip()
+    if "modin" in str(constructor):
+        # unreliable
+        pytest.skip()
+    if "dask" in str(constructor) and ddof != 1:
+        # Only `ddof=1` is supported
+        pytest.skip()
+    data = {
+        "a": [1, None, 2, None, 4, 6, 11],
+        "b": [1, None, 2, 3, 4, 5, 6],
+        "i": list(range(7)),
+    }
+    df = nw.from_native(constructor(data))
+    result = (
+        df.with_columns(
+            nw.col("a")
+            .rolling_var(window_size, min_samples=min_samples, center=center, ddof=ddof)
+            .over(order_by="b")
+        )
+        .select("a", "i")
+        .sort("i")
+    )
+    expected = {"a": expected_a, "i": list(range(7))}
+    assert_equal_data(result, expected)
+
+
+@pytest.mark.filterwarnings(
+    "ignore:`Expr.rolling_var` is being called from the stable API although considered an unstable feature."
+)
+@pytest.mark.parametrize(
+    ("expected_a", "window_size", "min_samples", "center", "ddof"),
+    [
+        ([None, None, 0.25, None, None, 1, 6.25], 2, None, False, 0),
+        ([None, None, 0.5, None, None, 2, 12.5], 2, 2, False, 1),
+        ([None, None, 0.5, 0.5, None, 2, 13], 3, 2, False, 1),
+        ([0, None, 0.25, 0.25, 0, 1, 8.666666666666666], 3, 1, False, 0),
+        ([0.5, None, 0.5, None, 2, 13, 12.5], 3, 1, True, 1),
+        ([0.5, None, 0.5, 0.5, 2, 13, 13], 4, 1, True, 1),
+        (
+            [
+                0.25,
+                0.25,
+                0.25,
+                0.25,
+                8.666666666666666,
+                8.666666666666666,
+                8.666666666666666,
+            ],
+            5,
+            1,
+            True,
+            0,
+        ),
+    ],
+)
+def test_rolling_var_expr_lazy_grouped(
+    constructor: Constructor,
+    expected_a: list[float],
+    window_size: int,
+    min_samples: int,
+    request: pytest.FixtureRequest,
+    *,
+    center: bool,
+    ddof: int,
+) -> None:
+    if ("polars" in str(constructor) and POLARS_VERSION < (1, 10)) or (
+        "duckdb" in str(constructor) and DUCKDB_VERSION < (1, 3)
+    ):
+        pytest.skip()
+    if "pandas" in str(constructor):
+        pytest.skip()
+    if any(x in str(constructor) for x in ("dask", "pyarrow_table")):
+        request.applymarker(pytest.mark.xfail)
+    if "cudf" in str(constructor) and center:
+        # center is not implemented for offset-based windows
+        request.applymarker(pytest.mark.xfail)
+    if "modin" in str(constructor):
+        # unreliable
+        pytest.skip()
+    data = {
+        "a": [1, None, 2, None, 4, 6, 11],
+        "g": [1, 1, 1, 1, 2, 2, 2],
+        "b": [1, None, 2, 3, 4, 5, 6],
+        "i": list(range(7)),
+    }
+    df = nw.from_native(constructor(data))
+    result = (
+        df.with_columns(
+            nw.col("a")
+            .rolling_var(window_size, min_samples=min_samples, center=center, ddof=ddof)
+            .over("g", order_by="b")
+        )
+        .sort("i")
+        .select("a")
+    )
+    expected = {"a": expected_a}
+    assert_equal_data(result, expected)
diff --git a/tests/read_scan_test.py b/tests/read_scan_test.py
index f18109e1..7a758ff7 100644
--- a/tests/read_scan_test.py
+++ b/tests/read_scan_test.py
@@ -1,6 +1,8 @@
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
 from typing import Any
+from typing import Literal
 from typing import Mapping
 
 import pandas as pd
@@ -17,6 +19,9 @@ from tests.utils import assert_equal_data
 pytest.importorskip("polars")
 import polars as pl
 
+if TYPE_CHECKING:
+    from types import ModuleType
+
 data: Mapping[str, Any] = {"a": [1, 2, 3], "b": [4.5, 6.7, 8.9], "z": ["x", "y", "w"]}
 TEST_EAGER_BACKENDS = [
     Implementation.POLARS,
@@ -28,35 +33,19 @@ TEST_EAGER_BACKENDS = [
 ]
 
 
-@pytest.mark.parametrize(
-    "backend",
-    TEST_EAGER_BACKENDS,
-)
+@pytest.mark.parametrize("backend", TEST_EAGER_BACKENDS)
+@pytest.mark.parametrize("nw_namespace", [nw, nw_v1])
 def test_read_csv(
     tmpdir: pytest.TempdirFactory,
     backend: Implementation | str,
+    nw_namespace: ModuleType,
 ) -> None:
     df_pl = pl.DataFrame(data)
     filepath = str(tmpdir / "file.csv")  # type: ignore[operator]
     df_pl.write_csv(filepath)
-    result = nw.read_csv(filepath, backend=backend)
-    assert_equal_data(result, data)
-    assert isinstance(result, nw.DataFrame)
-
-
-@pytest.mark.parametrize(
-    "backend",
-    TEST_EAGER_BACKENDS,
-)
-def test_read_csv_v1(
-    tmpdir: pytest.TempdirFactory, backend: Implementation | str
-) -> None:
-    df_pl = pl.DataFrame(data)
-    filepath = str(tmpdir / "file.csv")  # type: ignore[operator]
-    df_pl.write_csv(filepath)
-    result = nw_v1.read_csv(filepath, backend=backend)
+    result = nw_namespace.read_csv(filepath, backend=backend)
     assert_equal_data(result, data)
-    assert isinstance(result, nw_v1.DataFrame)
+    assert isinstance(result, nw_namespace.DataFrame)
 
 
 @pytest.mark.skipif(PANDAS_VERSION < (1, 5), reason="too old for pyarrow")
@@ -68,38 +57,48 @@ def test_read_csv_kwargs(tmpdir: pytest.TempdirFactory) -> None:
     assert_equal_data(result, data)
 
 
+@pytest.mark.parametrize("nw_namespace", [nw, nw_v1])
 def test_scan_csv(
     tmpdir: pytest.TempdirFactory,
-    request: pytest.FixtureRequest,
     constructor: Constructor,
+    nw_namespace: ModuleType,
 ) -> None:
-    if "pyspark" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-    df_pl = pl.DataFrame(data)
-    filepath = str(tmpdir / "file.csv")  # type: ignore[operator]
-    df_pl.write_csv(filepath)
-    df = nw.from_native(constructor(data))
-    backend = nw.get_native_namespace(df)
-    result = nw.scan_csv(filepath, backend=backend)
-    assert_equal_data(result, data)
-    assert isinstance(result, nw.LazyFrame)
+    kwargs: dict[str, Any]
+    if "sqlframe" in str(constructor):
+        from sqlframe.duckdb import DuckDBSession
+
+        kwargs = {
+            "session": DuckDBSession(),
+            "inferSchema": True,
+            "header": True,
+        }
+    elif "pyspark" in str(constructor):
+        from pyspark.sql import SparkSession
 
+        kwargs = {
+            "session": (
+                SparkSession.builder.appName("unit-tests")  # pyright: ignore[reportAttributeAccessIssue]
+                .master("local[1]")
+                .config("spark.ui.enabled", "false")
+                .config("spark.default.parallelism", "1")
+                .config("spark.sql.shuffle.partitions", "2")
+                .config("spark.sql.session.timeZone", "UTC")
+                .getOrCreate()
+            ),
+            "inferSchema": True,
+            "header": True,
+        }
+    else:
+        kwargs = {}
 
-def test_scan_csv_v1(
-    tmpdir: pytest.TempdirFactory,
-    request: pytest.FixtureRequest,
-    constructor: Constructor,
-) -> None:
-    if "pyspark" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
     df_pl = pl.DataFrame(data)
     filepath = str(tmpdir / "file.csv")  # type: ignore[operator]
     df_pl.write_csv(filepath)
-    df = nw_v1.from_native(constructor(data))
-    backend = nw_v1.get_native_namespace(df)
-    result = nw_v1.scan_csv(filepath, backend=backend)
+    df = nw_namespace.from_native(constructor(data))
+    backend = nw_namespace.get_native_namespace(df)
+    result = nw_namespace.scan_csv(filepath, backend=backend, **kwargs)
     assert_equal_data(result, data)
-    assert isinstance(result, nw_v1.LazyFrame)
+    assert isinstance(result, nw_namespace.LazyFrame)
 
 
 @pytest.mark.skipif(PANDAS_VERSION < (1, 5), reason="too old for pyarrow")
@@ -112,32 +111,20 @@ def test_scan_csv_kwargs(tmpdir: pytest.TempdirFactory) -> None:
 
 
 @pytest.mark.skipif(PANDAS_VERSION < (1, 5), reason="too old for pyarrow")
+@pytest.mark.parametrize("nw_namespace", [nw, nw_v1])
 def test_read_parquet(
     tmpdir: pytest.TempdirFactory,
     constructor_eager: ConstructorEager,
+    nw_namespace: ModuleType,
 ) -> None:
     df_pl = pl.DataFrame(data)
     filepath = str(tmpdir / "file.parquet")  # type: ignore[operator]
     df_pl.write_parquet(filepath)
-    df = nw.from_native(constructor_eager(data))
-    backend = nw.get_native_namespace(df)
-    result = nw.read_parquet(filepath, backend=backend)
+    df = nw_namespace.from_native(constructor_eager(data))
+    backend = nw_namespace.get_native_namespace(df)
+    result = nw_namespace.read_parquet(filepath, backend=backend)
     assert_equal_data(result, data)
-    assert isinstance(result, nw.DataFrame)
-
-
-@pytest.mark.skipif(PANDAS_VERSION < (1, 5), reason="too old for pyarrow")
-def test_read_parquet_v1(
-    tmpdir: pytest.TempdirFactory, constructor_eager: ConstructorEager
-) -> None:
-    df_pl = pl.DataFrame(data)
-    filepath = str(tmpdir / "file.parquet")  # type: ignore[operator]
-    df_pl.write_parquet(filepath)
-    df = nw_v1.from_native(constructor_eager(data))
-    backend = nw_v1.get_native_namespace(df)
-    result = nw_v1.read_parquet(filepath, backend=backend)
-    assert_equal_data(result, data)
-    assert isinstance(result, nw_v1.DataFrame)
+    assert isinstance(result, nw_namespace.DataFrame)
 
 
 @pytest.mark.skipif(PANDAS_VERSION < (1, 5), reason="too old for pyarrow")
@@ -150,9 +137,11 @@ def test_read_parquet_kwargs(tmpdir: pytest.TempdirFactory) -> None:
 
 
 @pytest.mark.skipif(PANDAS_VERSION < (1, 5), reason="too old for pyarrow")
+@pytest.mark.parametrize("nw_namespace", [nw, nw_v1])
 def test_scan_parquet(
     tmpdir: pytest.TempdirFactory,
     constructor: Constructor,
+    nw_namespace: ModuleType,
 ) -> None:
     kwargs: dict[str, Any]
     if "sqlframe" in str(constructor):
@@ -179,16 +168,30 @@ def test_scan_parquet(
     df_pl = pl.DataFrame(data)
     filepath = str(tmpdir / "file.parquet")  # type: ignore[operator]
     df_pl.write_parquet(filepath)
-    df = nw.from_native(constructor(data))
-    backend = nw.get_native_namespace(df)
-    result = nw.scan_parquet(filepath, backend=backend, **kwargs)
+    df = nw_namespace.from_native(constructor(data))
+    backend = nw_namespace.get_native_namespace(df)
+    result = nw_namespace.scan_parquet(filepath, backend=backend, **kwargs)
+    assert_equal_data(result, data)
+    assert isinstance(result, nw_namespace.LazyFrame)
+
+
+@pytest.mark.skipif(PANDAS_VERSION < (1, 5), reason="too old for pyarrow")
+def test_scan_parquet_kwargs(tmpdir: pytest.TempdirFactory) -> None:
+    df_pl = pl.DataFrame(data)
+    filepath = str(tmpdir / "file.parquet")  # type: ignore[operator]
+    df_pl.write_parquet(filepath)
+    result = nw.scan_parquet(filepath, backend=pd, engine="pyarrow")
     assert_equal_data(result, data)
-    assert isinstance(result, nw.LazyFrame)
 
 
 @pytest.mark.parametrize("spark_like_backend", ["pyspark", "sqlframe"])
+@pytest.mark.parametrize("scan_method", ["scan_csv", "scan_parquet"])
+@pytest.mark.parametrize("nw_namespace", [nw, nw_v1])
 def test_scan_fail_spark_like_without_session(
-    tmpdir: pytest.TempdirFactory, spark_like_backend: str
+    tmpdir: pytest.TempdirFactory,
+    spark_like_backend: str,
+    scan_method: Literal["scan_csv", "scan_parquet"],
+    nw_namespace: ModuleType,
 ) -> None:
     _ = pytest.importorskip(spark_like_backend)
 
@@ -200,31 +203,4 @@ def test_scan_fail_spark_like_without_session(
         ValueError,
         match="Spark like backends require a session object to be passed in `kwargs`.",
     ):
-        nw.scan_parquet("file.parquet", backend=spark_like_backend)
-
-
-@pytest.mark.skipif(PANDAS_VERSION < (1, 5), reason="too old for pyarrow")
-def test_scan_parquet_v1(
-    tmpdir: pytest.TempdirFactory,
-    request: pytest.FixtureRequest,
-    constructor: Constructor,
-) -> None:
-    if "pyspark" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-    df_pl = pl.DataFrame(data)
-    filepath = str(tmpdir / "file.parquet")  # type: ignore[operator]
-    df_pl.write_parquet(filepath)
-    df = nw_v1.from_native(constructor(data))
-    backend = nw_v1.get_native_namespace(df)
-    result = nw_v1.scan_parquet(filepath, backend=backend)
-    assert_equal_data(result, data)
-    assert isinstance(result, nw_v1.LazyFrame)
-
-
-@pytest.mark.skipif(PANDAS_VERSION < (1, 5), reason="too old for pyarrow")
-def test_scan_parquet_kwargs(tmpdir: pytest.TempdirFactory) -> None:
-    df_pl = pl.DataFrame(data)
-    filepath = str(tmpdir / "file.parquet")  # type: ignore[operator]
-    df_pl.write_parquet(filepath)
-    result = nw.scan_parquet(filepath, backend=pd, engine="pyarrow")
-    assert_equal_data(result, data)
+        getattr(nw_namespace, scan_method)("file.parquet", backend=spark_like_backend)