Repository: narwhals-dev/narwhals (python)

Issue:
remove unnecessary `echo "setuptools<78" |` from CI installs
this shouldn't be necessary any more

Pull request description:
chore: remove `setuptool<78` in CI
<!--
# 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
- [x] 🐳 Other

## Related issues

- Closes #2435 

## Checklist

- [x] Code follows style guide (ruff)
- [x] Tests added
- [x] Documented the changes

## If you have comments or can explain your changes, please do so below


Patch:
diff --git a/.github/workflows/typing.yml b/.github/workflows/typing.yml
index 92f4e1c3..3da4e24f 100644
--- a/.github/workflows/typing.yml
+++ b/.github/workflows/typing.yml
@@ -32,7 +32,7 @@ jobs:
         # TODO: add more dependencies/backends incrementally
         run: |
           source .venv/bin/activate
-          echo "setuptools<78" | uv pip install -b - -e ".[pyspark]" --group core --group typing
+          uv pip install -e ".[pyspark]" --group core --group typing
       - name: show-deps
         run: |
           source .venv/bin/activate
diff --git a/narwhals/_duckdb/expr.py b/narwhals/_duckdb/expr.py
index 981d7a73..a216a3fe 100644
--- a/narwhals/_duckdb/expr.py
+++ b/narwhals/_duckdb/expr.py
@@ -44,6 +44,7 @@ if TYPE_CHECKING:
     from narwhals._duckdb.typing import WindowFunction
     from narwhals._expression_parsing import ExprMetadata
     from narwhals.dtypes import DType
+    from narwhals.typing import FillNullStrategy
     from narwhals.typing import NonNestedLiteral
     from narwhals.typing import NumericLiteral
     from narwhals.typing import RankMethod
@@ -670,16 +671,41 @@ class DuckDBExpr(LazyExpr["DuckDBLazyFrame", "duckdb.Expression"]):
         )
 
     def fill_null(
-        self, value: Self | NonNestedLiteral, strategy: Any, limit: int | None
+        self,
+        value: Self | NonNestedLiteral,
+        strategy: FillNullStrategy | None,
+        limit: int | None,
     ) -> Self:
         if strategy is not None:
-            msg = "todo"
-            raise NotImplementedError(msg)
+            if self._backend_version < (1, 3):  # pragma: no cover
+                msg = f"`fill_null` with `strategy={strategy}` is only available in DuckDB>=1.3.0."
+                raise NotImplementedError(msg)
+
+            def _fill_with_strategy(window_inputs: WindowInputs) -> 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)
+
+                fill_func = "last_value" if strategy == "forward" else "first_value"
+                _limit = "unbounded" if limit is None else limit
+                rows_between = (
+                    f"{_limit} preceding and current row"
+                    if strategy == "forward"
+                    else f"current row and {_limit} following"
+                )
+                sql = (
+                    f"{fill_func}({window_inputs.expr} ignore nulls) over "
+                    f"({partition_by_sql} {order_by_sql} rows between {rows_between})"
+                )
+                return SQLExpression(sql)  # type: ignore[no-any-return, unused-ignore]
+
+            return self._with_window_function(_fill_with_strategy)
 
-        def func(_input: duckdb.Expression, value: Any) -> duckdb.Expression:
+        def _fill_constant(_input: duckdb.Expression, value: Any) -> duckdb.Expression:
             return CoalesceOperator(_input, value)
 
-        return self._with_callable(func, value=value)
+        return self._with_callable(_fill_constant, value=value)
 
     def cast(self, dtype: DType | type[DType]) -> Self:
         def func(_input: duckdb.Expression) -> duckdb.Expression:
diff --git a/narwhals/_spark_like/expr.py b/narwhals/_spark_like/expr.py
index 088468b0..b6fcd176 100644
--- a/narwhals/_spark_like/expr.py
+++ b/narwhals/_spark_like/expr.py
@@ -704,13 +704,40 @@ class SparkLikeExpr(LazyExpr["SparkLikeLazyFrame", "Column"]):
         limit: int | None,
     ) -> Self:
         if strategy is not None:
-            msg = "Support for strategies is not yet implemented."
-            raise NotImplementedError(msg)
 
-        def _fill_null(_input: Column, value: Column) -> Column:
+            def _fill_with_strategy(window_inputs: WindowInputs) -> Column:
+                fill_func = (
+                    self._F.last_value if strategy == "forward" else self._F.first_value
+                )
+
+                if strategy == "forward":
+                    start = (
+                        -limit if limit is not None else self._Window().unboundedPreceding
+                    )
+                    end = self._Window().currentRow
+                else:
+                    start = self._Window().currentRow
+                    end = (
+                        limit if limit is not None else self._Window().unboundedFollowing
+                    )
+
+                window = (
+                    self._Window()
+                    .partitionBy(list(window_inputs.partition_by) or self._F.lit(1))
+                    .orderBy(
+                        [self._F.col(x).asc_nulls_first() for x in window_inputs.order_by]
+                    )
+                    .rowsBetween(start, end)
+                )
+
+                return fill_func(window_inputs.expr, ignoreNulls=True).over(window)
+
+            return self._with_window_function(_fill_with_strategy)
+
+        def _fill_constant(_input: Column, value: Column) -> Column:
             return self._F.ifnull(_input, value)
 
-        return self._with_callable(_fill_null, value=value)
+        return self._with_callable(_fill_constant, value=value)
 
     def rolling_sum(self, window_size: int, *, min_samples: int, center: bool) -> Self:
         return self._with_window_function(
diff --git a/narwhals/expr.py b/narwhals/expr.py
index 6b302e7b..d409b4d5 100644
--- a/narwhals/expr.py
+++ b/narwhals/expr.py
@@ -1440,12 +1440,16 @@ class Expr:
         if strategy is not None and strategy not in {"forward", "backward"}:
             msg = f"strategy not supported: {strategy}"
             raise ValueError(msg)
-        return self._with_callable(
+
+        return self.__class__(
             lambda plx: self._to_compliant_expr(plx).fill_null(
                 value=extract_compliant(plx, value, str_as_lit=True),
                 strategy=strategy,
                 limit=limit,
-            )
+            ),
+            self._metadata.with_kind_and_closeable_window(ExprKind.WINDOW)
+            if strategy is not None
+            else self._metadata,
         )
 
     # --- partial reduction ---


Test patch:
diff --git a/.github/workflows/pytest-pyspark.yml b/.github/workflows/pytest-pyspark.yml
index af226297..c39c5d3d 100644
--- a/.github/workflows/pytest-pyspark.yml
+++ b/.github/workflows/pytest-pyspark.yml
@@ -10,7 +10,7 @@ env:
 jobs:
 
   pytest-pyspark-constructor:
-    if: ${{ contains(github.event.pull_request.labels.*.name, 'pyspark') || contains(github.event.pull_request.labels.*.name, 'release') }}
+    if: ${{ contains(github.event.pull_request.labels.*.name, 'pyspark') || contains(github.event.pull_request.labels.*.name, 'spark-like') || contains(github.event.pull_request.labels.*.name, 'release')}}
     strategy:
       matrix:
         python-version: ["3.10", "3.11"]
@@ -30,7 +30,7 @@ jobs:
       - name: install-reqs
         run: uv pip install -e . --group core-tests --group extra --system
       - name: install pyspark
-        run: echo "setuptools<78" | uv pip install -b - -e ".[pyspark]" --system
+        run: uv pip install -e ".[pyspark]" --system
       - name: show-deps
         run: uv pip freeze
       - name: Run pytest
diff --git a/tests/expr_and_series/fill_null_test.py b/tests/expr_and_series/fill_null_test.py
index 01de0400..26ca35e2 100644
--- a/tests/expr_and_series/fill_null_test.py
+++ b/tests/expr_and_series/fill_null_test.py
@@ -7,6 +7,8 @@ from typing import Any
 import pytest
 
 import narwhals 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
@@ -64,20 +66,23 @@ def test_fill_null_exceptions(constructor: Constructor) -> None:
         df.with_columns(nw.col("a").fill_null(strategy="invalid"))  # type: ignore  # noqa: PGH003
 
 
-def test_fill_null_strategies_with_limit_as_none(
-    constructor: Constructor, request: pytest.FixtureRequest
-) -> None:
-    if ("pyspark" in str(constructor)) or "duckdb" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
+def test_fill_null_strategies_with_limit_as_none(constructor: Constructor) -> None:
+    if ("duckdb" in str(constructor) and DUCKDB_VERSION < (1, 3)) or (
+        "polars" in str(constructor) and POLARS_VERSION < (1, 10)
+    ):
+        pytest.skip()
+
     data_limits = {
         "a": [1, None, None, None, 5, 6, None, None, None, 10],
         "b": ["a", None, None, None, "b", "c", None, None, None, "d"],
+        "idx": list(range(10)),
     }
     df = nw.from_native(constructor(data_limits))
 
     expected_forward = {
         "a": [1, 1, 1, 1, 5, 6, 6, 6, 6, 10],
         "b": ["a", "a", "a", "a", "b", "c", "c", "c", "c", "d"],
+        "idx": list(range(10)),
     }
     if (
         "pandas_pyarrow_constructor" in str(constructor)
@@ -94,12 +99,16 @@ def test_fill_null_strategies_with_limit_as_none(
                 "ignore", message="Falling back on a non-pyarrow code path which"
             )
             result_forward = df.with_columns(
-                nw.col("a", "b").fill_null(strategy="forward", limit=None)
+                nw.col("a", "b")
+                .fill_null(strategy="forward", limit=None)
+                .over(order_by="idx")
             )
             assert_equal_data(result_forward, expected_forward)
     else:
         result_forward = df.with_columns(
-            nw.col("a", "b").fill_null(strategy="forward", limit=None)
+            nw.col("a", "b")
+            .fill_null(strategy="forward", limit=None)
+            .over(order_by="idx")
         )
 
         assert_equal_data(result_forward, expected_forward)
@@ -107,6 +116,7 @@ def test_fill_null_strategies_with_limit_as_none(
     expected_backward = {
         "a": [1, 5, 5, 5, 5, 6, 10, 10, 10, 10],
         "b": ["a", "b", "b", "b", "b", "c", "d", "d", "d", "d"],
+        "idx": list(range(10)),
     }
     if (
         "pandas_pyarrow_constructor" in str(constructor)
@@ -124,21 +134,26 @@ def test_fill_null_strategies_with_limit_as_none(
                 "ignore", message="Falling back on a non-pyarrow code path which"
             )
             result_backward = df.with_columns(
-                nw.col("a", "b").fill_null(strategy="backward", limit=None)
+                nw.col("a", "b")
+                .fill_null(strategy="backward", limit=None)
+                .over(order_by="idx")
             )
             assert_equal_data(result_backward, expected_backward)
     else:
         result_backward = df.with_columns(
-            nw.col("a", "b").fill_null(strategy="backward", limit=None)
+            nw.col("a", "b")
+            .fill_null(strategy="backward", limit=None)
+            .over(order_by="idx")
         )
         assert_equal_data(result_backward, expected_backward)
 
 
-def test_fill_null_limits(
-    constructor: Constructor, request: pytest.FixtureRequest
-) -> None:
-    if ("pyspark" in str(constructor)) or "duckdb" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
+def test_fill_null_limits(constructor: Constructor) -> None:
+    if ("duckdb" in str(constructor) and DUCKDB_VERSION < (1, 3)) or (
+        "polars" in str(constructor) and POLARS_VERSION < (1, 10)
+    ):
+        pytest.skip()
+
     context: Any = (
         pytest.raises(NotImplementedError, match="The limit keyword is not supported")
         if "cudf" in str(constructor)
@@ -149,6 +164,7 @@ def test_fill_null_limits(
     data_limits = {
         "a": [1, None, None, None, 5, 6, None, None, None, 10],
         "b": ["a", None, None, None, "b", "c", None, None, None, "d"],
+        "idx": list(range(10)),
     }
     df = nw.from_native(constructor(data_limits))
     with context:
@@ -158,21 +174,23 @@ def test_fill_null_limits(
             )
 
         result_forward = df.with_columns(
-            nw.col("a", "b").fill_null(strategy="forward", limit=2)
+            nw.col("a", "b").fill_null(strategy="forward", limit=2).over(order_by="idx")
         )
         expected_forward = {
             "a": [1, 1, 1, None, 5, 6, 6, 6, None, 10],
             "b": ["a", "a", "a", None, "b", "c", "c", "c", None, "d"],
+            "idx": list(range(10)),
         }
         assert_equal_data(result_forward, expected_forward)
 
         result_backward = df.with_columns(
-            nw.col("a", "b").fill_null(strategy="backward", limit=2)
+            nw.col("a", "b").fill_null(strategy="backward", limit=2).over(order_by="idx")
         )
 
         expected_backward = {
             "a": [1, None, 5, 5, 5, 6, None, 10, 10, 10],
             "b": ["a", None, "b", "b", "b", "c", None, "d", "d", "d"],
+            "idx": list(range(10)),
         }
         assert_equal_data(result_backward, expected_backward)
 
@@ -348,3 +366,44 @@ def test_fill_null_series_exceptions(constructor_eager: ConstructorEager) -> Non
         df_float.select(
             a_zero_digit=df_float["a"].fill_null(strategy="invalid"),  # type: ignore  # noqa: PGH003
         )
+
+
+def test_fill_null_strategies_with_partition_by(
+    constructor: Constructor, request: pytest.FixtureRequest
+) -> None:
+    if any(x in str(constructor) for x in ("pandas", "modin", "cudf", "pyarrow", "dask")):
+        request.applymarker(pytest.mark.xfail)
+
+    if ("duckdb" in str(constructor) and DUCKDB_VERSION < (1, 3)) or (
+        "polars" in str(constructor) and POLARS_VERSION < (1, 10)
+    ):
+        pytest.skip()
+
+    data = {
+        "partition": ["A", "B", "C", "B", "A", "B", "A", "C", "C"],
+        "values": [1, None, None, 2, None, 3, None, None, 4],
+        "idx": list(range(9)),
+    }
+    df = nw.from_native(constructor(data))
+
+    # Forward fill within each group
+    result_forward = df.with_columns(
+        nw.col("values").fill_null(strategy="forward").over("partition", order_by="idx")
+    ).sort("idx")
+    expected_forward = {
+        "partition": ["A", "B", "C", "B", "A", "B", "A", "C", "C"],
+        "values": [1, None, None, 2, 1, 3, 1, None, 4],
+        "idx": list(range(9)),
+    }
+    assert_equal_data(result_forward, expected_forward)
+
+    # Backward fill within each group
+    result_backward = df.with_columns(
+        nw.col("values").fill_null(strategy="backward").over("partition", order_by="idx")
+    ).sort("idx")
+    expected_backward = {
+        "partition": ["A", "B", "C", "B", "A", "B", "A", "C", "C"],
+        "values": [1, 2, 4, 2, None, 3, None, 4, 4],
+        "idx": list(range(9)),
+    }
+    assert_equal_data(result_backward, expected_backward)