Repository: narwhals-dev/narwhals
Pull request #2729 (2025-06-23 19:09:20)

chore: use `window_expression` for DuckDB
This should help reduce churn when they introduce `duckdb.WindowExpression`

It also reduces how much sql we need to write by concentrating it all in `window_expression`, reducing the potential points of failure

~~`rank` still needs doing~~

<!--
# Thanks for contributing a pull request! 
## Please make sure you see our contribution guidelines: https://github.com/narwhals-dev/narwhals/blob/main/CONTRIBUTING.md
-->

## What type of PR is this? (check all applicable)

- [ ] 💾 Refactor
- [ ] ✨ Feature
- [ ] 🐛 Bug Fix
- [ ] 🔧 Optimization
- [ ] 📝 Documentation
- [ ] ✅ Test
- [ ] 🐳 Other

## Related issues

- Related issue #\<issue number\>
- Closes #\<issue number\>

## Checklist

- [ ] Code follows style guide (ruff)
- [ ] Tests added
- [ ] Documented the changes

## If you have comments or can explain your changes, please do so below



Patch:
diff --git a/docs/backcompat.md b/docs/backcompat.md
index 0742731b..cc605b01 100644
--- a/docs/backcompat.md
+++ b/docs/backcompat.md
@@ -113,6 +113,8 @@ Which should you use? In general we recommend:
 
 - Since Narwhals 1.45:
 
+    - `nw.any_horizontal` and `nw.all_horizontal` have a `ignore_nulls` keyword. In `narwhals.stable.v1`,
+      it defaults to `False`, but in Narwhals 2.0 it will become a required argument in the main namespace.
     - `LazyFrame.with_row_index` requires `order_by` to be specified as it is an order-dependent operation, in the main Narwhals namespace.
 
 - Since Narwhals 1.43:
diff --git a/docs/concepts/boolean.md b/docs/concepts/boolean.md
index c7c166f5..c40e497f 100644
--- a/docs/concepts/boolean.md
+++ b/docs/concepts/boolean.md
@@ -111,3 +111,12 @@ Here, too, pandas backed by NumPy types differs, as its boolean columns cannot s
 
 - For `nw.col('a') | nw.col('b')`, pandas returns `True` if at least one column contains a `True` value, and `False` otherwise.
 - For `nw.col('a') & nw.col('b')`, pandas returns `True` if both columns contain `True` values, and `False` otherwise.
+
+In `any_horizontal` and `all_horizontal` there is an `ignore_nulls` argument, which behaves as follows:
+
+- If `True`, then null values are ignored and contribute nothing to the final result. If there are
+    no values, the result is:
+
+    - `False` for `any_horizontal`.
+    - `True` for `all_horizontal`.
+- If `False`, then Kleene logic is followed. If using pandas backed by classical NumPy types, then this option is not supported.
diff --git a/narwhals/_arrow/dataframe.py b/narwhals/_arrow/dataframe.py
index 7fadd2f4..56540e8e 100644
--- a/narwhals/_arrow/dataframe.py
+++ b/narwhals/_arrow/dataframe.py
@@ -434,7 +434,8 @@ class ArrowDataFrame(
         if subset is None:
             return self._with_native(self.native.drop_null(), validate_column_names=False)
         plx = self.__narwhals_namespace__()
-        return self.filter(~plx.any_horizontal(plx.col(*subset).is_null()))
+        mask = ~plx.any_horizontal(plx.col(*subset).is_null(), ignore_nulls=True)
+        return self.filter(mask)
 
     def sort(self, *by: str, descending: bool | Sequence[bool], nulls_last: bool) -> Self:
         if isinstance(descending, bool):
diff --git a/narwhals/_arrow/namespace.py b/narwhals/_arrow/namespace.py
index 84c3475a..82cc37a7 100644
--- a/narwhals/_arrow/namespace.py
+++ b/narwhals/_arrow/namespace.py
@@ -82,11 +82,16 @@ class ArrowNamespace(
             version=self._version,
         )
 
-    def all_horizontal(self, *exprs: ArrowExpr) -> ArrowExpr:
+    def all_horizontal(self, *exprs: ArrowExpr, ignore_nulls: bool) -> ArrowExpr:
         def func(df: ArrowDataFrame) -> list[ArrowSeries]:
             series = chain.from_iterable(expr(df) for expr in exprs)
             align = self._series._align_full_broadcast
-            return [reduce(operator.and_, align(*series))]
+            it = (
+                (s.fill_null(True, None, None) for s in series)  # noqa: FBT003
+                if ignore_nulls
+                else series
+            )
+            return [reduce(operator.and_, align(*it))]
 
         return self._expr._from_callable(
             func=func,
@@ -97,11 +102,16 @@ class ArrowNamespace(
             context=self,
         )
 
-    def any_horizontal(self, *exprs: ArrowExpr) -> ArrowExpr:
+    def any_horizontal(self, *exprs: ArrowExpr, ignore_nulls: bool) -> ArrowExpr:
         def func(df: ArrowDataFrame) -> list[ArrowSeries]:
             series = chain.from_iterable(expr(df) for expr in exprs)
             align = self._series._align_full_broadcast
-            return [reduce(operator.or_, align(*series))]
+            it = (
+                (s.fill_null(False, None, None) for s in series)  # noqa: FBT003
+                if ignore_nulls
+                else series
+            )
+            return [reduce(operator.or_, align(*it))]
 
         return self._expr._from_callable(
             func=func,
diff --git a/narwhals/_compliant/expr.py b/narwhals/_compliant/expr.py
index 021dc840..3528ad97 100644
--- a/narwhals/_compliant/expr.py
+++ b/narwhals/_compliant/expr.py
@@ -663,7 +663,7 @@ class EagerExpr(
 
     def filter(self, *predicates: Self) -> Self:
         plx = self.__narwhals_namespace__()
-        predicate = plx.all_horizontal(*predicates)
+        predicate = plx.all_horizontal(*predicates, ignore_nulls=False)
         return self._reuse_series("filter", predicate=predicate)
 
     def drop_nulls(self) -> Self:
diff --git a/narwhals/_compliant/namespace.py b/narwhals/_compliant/namespace.py
index 07d23199..384527b3 100644
--- a/narwhals/_compliant/namespace.py
+++ b/narwhals/_compliant/namespace.py
@@ -69,8 +69,12 @@ class CompliantNamespace(Protocol[CompliantFrameT, CompliantExprT]):
 
     def len(self) -> CompliantExprT: ...
     def lit(self, value: NonNestedLiteral, dtype: IntoDType | None) -> CompliantExprT: ...
-    def all_horizontal(self, *exprs: CompliantExprT) -> CompliantExprT: ...
-    def any_horizontal(self, *exprs: CompliantExprT) -> CompliantExprT: ...
+    def all_horizontal(
+        self, *exprs: CompliantExprT, ignore_nulls: bool
+    ) -> CompliantExprT: ...
+    def any_horizontal(
+        self, *exprs: CompliantExprT, ignore_nulls: bool
+    ) -> CompliantExprT: ...
     def sum_horizontal(self, *exprs: CompliantExprT) -> CompliantExprT: ...
     def mean_horizontal(self, *exprs: CompliantExprT) -> CompliantExprT: ...
     def min_horizontal(self, *exprs: CompliantExprT) -> CompliantExprT: ...
diff --git a/narwhals/_dask/dataframe.py b/narwhals/_dask/dataframe.py
index 8e5280c6..679b5ad0 100644
--- a/narwhals/_dask/dataframe.py
+++ b/narwhals/_dask/dataframe.py
@@ -183,7 +183,8 @@ class DaskLazyFrame(
         if subset is None:
             return self._with_native(self.native.dropna())
         plx = self.__narwhals_namespace__()
-        return self.filter(~plx.any_horizontal(plx.col(*subset).is_null()))
+        mask = ~plx.any_horizontal(plx.col(*subset).is_null(), ignore_nulls=True)
+        return self.filter(mask)
 
     @property
     def schema(self) -> dict[str, DType]:
diff --git a/narwhals/_dask/namespace.py b/narwhals/_dask/namespace.py
index d58e7957..92c22550 100644
--- a/narwhals/_dask/namespace.py
+++ b/narwhals/_dask/namespace.py
@@ -91,12 +91,21 @@ class DaskNamespace(
             version=self._version,
         )
 
-    def all_horizontal(self, *exprs: DaskExpr) -> DaskExpr:
+    def all_horizontal(self, *exprs: DaskExpr, ignore_nulls: bool) -> DaskExpr:
         def func(df: DaskLazyFrame) -> list[dx.Series]:
-            series = align_series_full_broadcast(
-                df, *(s for _expr in exprs for s in _expr(df))
+            series = (s for _expr in exprs for s in _expr(df))
+            # Note on `ignore_nulls`: Dask doesn't support storing arbitrary Python
+            # objects in `object` dtype, so we don't need the same check we have for pandas-like.
+            it = (
+                (
+                    # NumPy-backed 'bool' dtype can't contain nulls so doesn't need filling.
+                    s if s.dtype == "bool" else s.fillna(True)  # noqa: FBT003
+                    for s in series
+                )
+                if ignore_nulls
+                else series
             )
-            return [reduce(operator.and_, series)]
+            return [reduce(operator.and_, align_series_full_broadcast(df, *it))]
 
         return self._expr(
             call=func,
@@ -108,12 +117,21 @@ class DaskNamespace(
             version=self._version,
         )
 
-    def any_horizontal(self, *exprs: DaskExpr) -> DaskExpr:
+    def any_horizontal(self, *exprs: DaskExpr, ignore_nulls: bool) -> DaskExpr:
         def func(df: DaskLazyFrame) -> list[dx.Series]:
-            series = align_series_full_broadcast(
-                df, *(s for _expr in exprs for s in _expr(df))
+            series = (s for _expr in exprs for s in _expr(df))
+            # Note on `ignore_nulls`: Dask doesn't support storing arbitrary Python
+            # objects in `object` dtype, so we don't need the same check we have for pandas-like.
+            it = (
+                (
+                    # NumPy-backed 'bool' dtype can't contain nulls so doesn't need filling.
+                    s if s.dtype == "bool" else s.fillna(False)  # noqa: FBT003
+                    for s in series
+                )
+                if ignore_nulls
+                else series
             )
-            return [reduce(operator.or_, series)]
+            return [reduce(operator.or_, align_series_full_broadcast(df, *it))]
 
         return self._expr(
             call=func,
diff --git a/narwhals/_duckdb/dataframe.py b/narwhals/_duckdb/dataframe.py
index 4267e32c..a2869a38 100644
--- a/narwhals/_duckdb/dataframe.py
+++ b/narwhals/_duckdb/dataframe.py
@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import contextlib
 from functools import reduce
 from operator import and_
 from typing import TYPE_CHECKING, Any
@@ -12,10 +11,9 @@ from narwhals._duckdb.utils import (
     DeferredTimeZone,
     col,
     evaluate_exprs,
-    generate_order_by_sql,
-    generate_partition_by_sql,
     lit,
     native_to_narwhals_dtype,
+    window_expression,
 )
 from narwhals._utils import (
     Implementation,
@@ -52,9 +50,6 @@ if TYPE_CHECKING:
     from narwhals.stable.v1 import DataFrame as DataFrameV1
     from narwhals.typing import AsofJoinStrategy, JoinStrategy, LazyUniqueKeepStrategy
 
-with contextlib.suppress(ImportError):  # requires duckdb>=1.3.0
-    from duckdb import SQLExpression
-
 
 class DuckDBLazyFrame(
     CompliantLazyFrame[
@@ -383,24 +378,17 @@ class DuckDBLazyFrame(
         self, subset: Sequence[str] | None, *, keep: LazyUniqueKeepStrategy
     ) -> Self:
         if subset_ := subset if keep == "any" else (subset or self.columns):
-            if self._backend_version < (1, 3):
-                msg = (
-                    "At least version 1.3 of DuckDB is required for `unique` operation\n"
-                    "with `subset` specified."
-                )
-                raise NotImplementedError(msg)
             # Sanitise input
             if error := self._check_columns_exist(subset_):
                 raise error
             idx_name = generate_temporary_column_name(8, self.columns)
             count_name = generate_temporary_column_name(8, [*self.columns, idx_name])
-            partition_by_sql = generate_partition_by_sql(*(subset_))
             name = count_name if keep == "none" else idx_name
-            idx_expr = SQLExpression(
-                f"{FunctionExpression('row_number')} over ({partition_by_sql})"
-            ).alias(idx_name)
-            count_expr = SQLExpression(
-                f"{FunctionExpression('count', StarExpression())} over ({partition_by_sql})"
+            idx_expr = window_expression(FunctionExpression("row_number"), subset_).alias(
+                idx_name
+            )
+            count_expr = window_expression(
+                FunctionExpression("count", StarExpression()), subset_, ()
             ).alias(count_name)
             return self._with_native(
                 self.native.select(StarExpression(), idx_expr, count_expr)
@@ -508,11 +496,11 @@ class DuckDBLazyFrame(
 
     @requires.backend_version((1, 3))
     def with_row_index(self, name: str, order_by: Sequence[str]) -> Self:
-        order_by_sql = generate_order_by_sql(*order_by, ascending=True)
-        row_index_expr = SQLExpression(
-            f"{FunctionExpression('row_number')} over ({order_by_sql}) - 1"
+        expr = (
+            window_expression(FunctionExpression("row_number"), order_by=order_by)
+            - lit(1)
         ).alias(name)
-        return self._with_native(self.native.select(row_index_expr, StarExpression()))
+        return self._with_native(self.native.select(expr, StarExpression()))
 
     gather_every = not_implemented.deprecated(
         "`LazyFrame.gather_every` is deprecated and will be removed in a future version."
diff --git a/narwhals/_duckdb/expr.py b/narwhals/_duckdb/expr.py
index 6f407b54..30a792b1 100644
--- a/narwhals/_duckdb/expr.py
+++ b/narwhals/_duckdb/expr.py
@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import contextlib
 import operator
 from typing import TYPE_CHECKING, Any, Callable, Literal, cast
 
@@ -15,11 +14,10 @@ from narwhals._duckdb.expr_str import DuckDBExprStringNamespace
 from narwhals._duckdb.expr_struct import DuckDBExprStructNamespace
 from narwhals._duckdb.utils import (
     col,
-    generate_order_by_sql,
-    generate_partition_by_sql,
     lit,
     narwhals_to_native_dtype,
     when,
+    window_expression,
 )
 from narwhals._expression_parsing import ExprKind
 from narwhals._utils import Implementation, not_implemented, requires
@@ -38,6 +36,7 @@ if TYPE_CHECKING:
     )
     from narwhals._duckdb.dataframe import DuckDBLazyFrame
     from narwhals._duckdb.namespace import DuckDBNamespace
+    from narwhals._duckdb.typing import WindowExpressionKwargs
     from narwhals._expression_parsing import ExprMetadata
     from narwhals._utils import Version, _FullContext
     from narwhals.typing import (
@@ -54,10 +53,6 @@ if TYPE_CHECKING:
     DuckDBWindowInputs = WindowInputs[Expression]
 
 
-with contextlib.suppress(ImportError):  # requires duckdb>=1.3.0
-    from duckdb import SQLExpression
-
-
 class DuckDBExpr(LazyExpr["DuckDBLazyFrame", "Expression"]):
     _implementation = Implementation.DUCKDB
 
@@ -82,12 +77,13 @@ class DuckDBExpr(LazyExpr["DuckDBLazyFrame", "Expression"]):
     @property
     def window_function(self) -> DuckDBWindowFunction:
         def default_window_func(
-            df: DuckDBLazyFrame, window_inputs: DuckDBWindowInputs
+            df: DuckDBLazyFrame, inputs: DuckDBWindowInputs
         ) -> list[Expression]:
-            assert not window_inputs.order_by  # noqa: S101
-            partition_by_sql = generate_partition_by_sql(*window_inputs.partition_by)
-            template = f"{{expr}} over ({partition_by_sql})"
-            return [SQLExpression(template.format(expr=expr)) for expr in self(df)]
+            assert not inputs.order_by  # noqa: S101
+            return [
+                window_expression(expr, inputs.partition_by, inputs.order_by)
+                for expr in self(df)
+            ]
 
         return self._window_function or default_window_func
 
@@ -111,13 +107,18 @@ class DuckDBExpr(LazyExpr["DuckDBLazyFrame", "Expression"]):
         func_name: Literal["sum", "max", "min", "count", "product"],
     ) -> DuckDBWindowFunction:
         def func(df: DuckDBLazyFrame, inputs: DuckDBWindowInputs) -> list[Expression]:
-            order_by_sql = generate_order_by_sql(*inputs.order_by, ascending=not reverse)
-            partition_by_sql = generate_partition_by_sql(*inputs.partition_by)
-            sql = (
-                f"{func_name} ({{expr}}) over ({partition_by_sql} {order_by_sql} "
-                "rows between unbounded preceding and current row)"
-            )
-            return [SQLExpression(sql.format(expr=expr)) for expr in self(df)]
+            return [
+                window_expression(
+                    FunctionExpression(func_name, expr),
+                    inputs.partition_by,
+                    inputs.order_by,
+                    descending=reverse,
+                    nulls_last=reverse,
+                    rows_start="unbounded preceding",
+                    rows_end="current row",
+                )
+                for expr in self(df)
+            ]
 
         return func
 
@@ -141,9 +142,6 @@ class DuckDBExpr(LazyExpr["DuckDBLazyFrame", "Expression"]):
             end = "current row"
 
         def func(df: DuckDBLazyFrame, inputs: DuckDBWindowInputs) -> list[Expression]:
-            order_by_sql = generate_order_by_sql(*inputs.order_by, ascending=True)
-            partition_by_sql = generate_partition_by_sql(*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:
@@ -160,12 +158,17 @@ class DuckDBExpr(LazyExpr["DuckDBLazyFrame", "Expression"]):
             else:  # pragma: no cover
                 msg = f"Only the following functions are supported: {supported_funcs}.\nGot: {func_name}."
                 raise ValueError(msg)
-            condition_sql = f"count({{expr}}) over {window} >= {min_samples}"
-            value_sql = f"{func_}({{expr}}) over {window}"
+            window_kwargs: WindowExpressionKwargs = {
+                "partition_by": inputs.partition_by,
+                "order_by": inputs.order_by,
+                "rows_start": start,
+                "rows_end": end,
+            }
             return [
                 when(
-                    SQLExpression(condition_sql.format(expr=expr)),
-                    SQLExpression(value_sql.format(expr=expr)),
+                    window_expression(FunctionExpression("count", expr), **window_kwargs)
+                    >= lit(min_samples),
+                    window_expression(FunctionExpression(func_, expr), **window_kwargs),
                 )
                 for expr in self(df)
             ]
@@ -232,17 +235,15 @@ class DuckDBExpr(LazyExpr["DuckDBLazyFrame", "Expression"]):
     def _push_down_window_function(
         self, call: Callable[..., Expression], /, **expressifiable_args: Self | Any
     ) -> DuckDBWindowFunction:
-        def window_f(
-            df: DuckDBLazyFrame, window_inputs: DuckDBWindowInputs
-        ) -> list[Expression]:
+        def window_f(df: DuckDBLazyFrame, inputs: DuckDBWindowInputs) -> list[Expression]:
             # If a function `f` is elementwise, and `g` is another function, then
             # - `f(g) over (window)`
             # - `f(g over (window))
             # are equivalent.
             # Make sure to only use with if `call` is elementwise!
-            native_series_list = self.window_function(df, window_inputs)
+            native_series_list = self.window_function(df, inputs)
             other_native_series = {
-                key: df._evaluate_window_expr(value, window_inputs)
+                key: df._evaluate_window_expr(value, inputs)
                 if self._is_expr(value)
                 else lit(value)
                 for key, value in expressifiable_args.items()
@@ -426,13 +427,12 @@ class DuckDBExpr(LazyExpr["DuckDBLazyFrame", "Expression"]):
         def f(expr: Expression) -> Expression:
             return CoalesceOperator(FunctionExpression("bool_and", expr), lit(True))  # noqa: FBT003
 
-        def window_f(
-            df: DuckDBLazyFrame, window_inputs: DuckDBWindowInputs
-        ) -> list[Expression]:
-            pb = generate_partition_by_sql(*window_inputs.partition_by)
+        def window_f(df: DuckDBLazyFrame, inputs: DuckDBWindowInputs) -> list[Expression]:
             return [
                 CoalesceOperator(
-                    SQLExpression(f"{FunctionExpression('bool_and', expr)} over ({pb})"),
+                    window_expression(
+                        FunctionExpression("bool_and", expr), inputs.partition_by
+                    ),
                     lit(True),  # noqa: FBT003
                 )
                 for expr in self(df)
@@ -444,13 +444,12 @@ class DuckDBExpr(LazyExpr["DuckDBLazyFrame", "Expression"]):
         def f(expr: Expression) -> Expression:
             return CoalesceOperator(FunctionExpression("bool_or", expr), lit(False))  # noqa: FBT003
 
-        def window_f(
-            df: DuckDBLazyFrame, window_inputs: DuckDBWindowInputs
-        ) -> list[Expression]:
-            pb = generate_partition_by_sql(*window_inputs.partition_by)
+        def window_f(df: DuckDBLazyFrame, inputs: DuckDBWindowInputs) -> list[Expression]:
             return [
                 CoalesceOperator(
-                    SQLExpression(f"{FunctionExpression('bool_or', expr)} over ({pb})"),
+                    window_expression(
+                        FunctionExpression("bool_or", expr), inputs.partition_by
+                    ),
                     lit(False),  # noqa: FBT003
                 )
                 for expr in self(df)
@@ -499,13 +498,12 @@ class DuckDBExpr(LazyExpr["DuckDBLazyFrame", "Expression"]):
         def f(expr: Expression) -> Expression:
             return CoalesceOperator(FunctionExpression("sum", expr), lit(0))
 
-        def window_f(
-            df: DuckDBLazyFrame, window_inputs: DuckDBWindowInputs
-        ) -> list[Expression]:
-            pb = generate_partition_by_sql(*window_inputs.partition_by)
+        def window_f(df: DuckDBLazyFrame, inputs: DuckDBWindowInputs) -> list[Expression]:
             return [
                 CoalesceOperator(
-                    SQLExpression(f"{FunctionExpression('sum', expr)} over ({pb})"),
+                    window_expression(
+                        FunctionExpression("sum", expr), inputs.partition_by
+                    ),
                     lit(0),
                 )
                 for expr in self(df)
@@ -612,56 +610,59 @@ class DuckDBExpr(LazyExpr["DuckDBLazyFrame", "Expression"]):
     @requires.backend_version((1, 3))
     def shift(self, n: int) -> Self:
         def func(df: DuckDBLazyFrame, inputs: DuckDBWindowInputs) -> Sequence[Expression]:
-            order_by_sql = generate_order_by_sql(*inputs.order_by, ascending=True)
-            partition_by_sql = generate_partition_by_sql(*inputs.partition_by)
-            sql = f"lag({{expr}}, {n}) over ({partition_by_sql} {order_by_sql})"
-            return [SQLExpression(sql.format(expr=expr)) for expr in self(df)]
+            return [
+                window_expression(
+                    FunctionExpression("lag", expr, lit(n)),
+                    inputs.partition_by,
+                    inputs.order_by,
+                )
+                for expr in self(df)
+            ]
 
         return self._with_window_function(func)
 
     @requires.backend_version((1, 3))
     def is_first_distinct(self) -> Self:
         def func(df: DuckDBLazyFrame, inputs: DuckDBWindowInputs) -> Sequence[Expression]:
-            order_by_sql = generate_order_by_sql(*inputs.order_by, ascending=True)
-            if inputs.partition_by:
-                partition_by_sql = (
-                    generate_partition_by_sql(*inputs.partition_by) + ", {expr}"
+            return [
+                window_expression(
+                    FunctionExpression("row_number"),
+                    (*inputs.partition_by, expr),
+                    inputs.order_by,
                 )
-            else:
-                partition_by_sql = "partition by {expr}"
-            sql = (
-                f"{FunctionExpression('row_number')} "
-                f"over({partition_by_sql} {order_by_sql})"
-            )
-            return [SQLExpression(sql.format(expr=expr)) == lit(1) for expr in self(df)]
+                == lit(1)
+                for expr in self(df)
+            ]
 
         return self._with_window_function(func)
 
     @requires.backend_version((1, 3))
     def is_last_distinct(self) -> Self:
         def func(df: DuckDBLazyFrame, inputs: DuckDBWindowInputs) -> Sequence[Expression]:
-            order_by_sql = generate_order_by_sql(*inputs.order_by, ascending=False)
-            if inputs.partition_by:
-                partition_by_sql = (
-                    generate_partition_by_sql(*inputs.partition_by) + ", {expr}"
+            return [
+                window_expression(
+                    FunctionExpression("row_number"),
+                    (*inputs.partition_by, expr),
+                    inputs.order_by,
+                    descending=True,
+                    nulls_last=True,
                 )
-            else:
-                partition_by_sql = "partition by {expr}"
-            sql = (
-                f"{FunctionExpression('row_number')} "
-                f"over({partition_by_sql} {order_by_sql})"
-            )
-            return [SQLExpression(sql.format(expr=expr)) == lit(1) for expr in self(df)]
+                == lit(1)
+                for expr in self(df)
+            ]
 
         return self._with_window_function(func)
 
     @requires.backend_version((1, 3))
     def diff(self) -> Self:
         def func(df: DuckDBLazyFrame, inputs: DuckDBWindowInputs) -> list[Expression]:
-            order_by_sql = generate_order_by_sql(*inputs.order_by, ascending=True)
-            partition_by_sql = generate_partition_by_sql(*inputs.partition_by)
-            sql = f"lag({{expr}}) over ({partition_by_sql} {order_by_sql})"
-            return [expr - SQLExpression(sql.format(expr=expr)) for expr in self(df)]
+            return [
+                expr
+                - window_expression(
+                    FunctionExpression("lag", expr), inputs.partition_by, inputs.order_by
+                )
+                for expr in self(df)
+            ]
 
         return self._with_window_function(func)
 
@@ -759,21 +760,24 @@ class DuckDBExpr(LazyExpr["DuckDBLazyFrame", "Expression"]):
             def _fill_with_strategy(
                 df: DuckDBLazyFrame, inputs: DuckDBWindowInputs
             ) -> Sequence[Expression]:
-                order_by_sql = generate_order_by_sql(*inputs.order_by, ascending=True)
-                partition_by_sql = generate_partition_by_sql(*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"
+                rows_start, rows_end = (
+                    (f"{_limit} preceding", "current row")
                     if strategy == "forward"
-                    else f"current row and {_limit} following"
+                    else ("current row", f"{_limit} following")
                 )
-                sql = (
-                    f"{fill_func}({{expr}} ignore nulls) over "
-                    f"({partition_by_sql} {order_by_sql} rows between {rows_between})"
-                )
-                return [SQLExpression(sql.format(expr=expr)) for expr in self(df)]
+                return [
+                    window_expression(
+                        FunctionExpression(fill_func, expr),
+                        inputs.partition_by,
+                        inputs.order_by,
+                        rows_start=rows_start,
+                        rows_end=rows_end,
+                        ignore_nulls=True,
+                    )
+                    for expr in self(df)
+                ]
 
             return self._with_window_function(_fill_with_strategy)
 
@@ -792,9 +796,9 @@ class DuckDBExpr(LazyExpr["DuckDBLazyFrame", "Expression"]):
     @requires.backend_version((1, 3))
     def is_unique(self) -> Self:
         def _is_unique(expr: Expression, *partition_by: str | Expression) -> Expression:
-            pb = generate_partition_by_sql(expr, *partition_by)
-            sql = f"{FunctionExpression('count', col('*'))} over ({pb})"
-            return SQLExpression(sql) == lit(1)
+            return window_expression(
+                FunctionExpression("count", StarExpression()), (expr, *partition_by)
+            ) == lit(1)
 
         def _unpartitioned_is_unique(expr: Expression) -> Expression:
             return _is_unique(expr)
@@ -822,36 +826,34 @@ class DuckDBExpr(LazyExpr["DuckDBLazyFrame", "Expression"]):
             expr: Expression,
             *,
             descending: bool,
-            partition_by: Sequence[str | Expression] | None = None,
+            partition_by: Sequence[str | Expression],
         ) -> Expression:
-            order_by_sql = (
-                f"order by {expr} desc nulls last"
-                if descending
-                else f"order by {expr} asc nulls last"
-            )
             count_expr = FunctionExpression("count", StarExpression())
-            if partition_by is not None:
-                window = f"{generate_partition_by_sql(*partition_by)} {order_by_sql}"
-                count_window = f"{generate_partition_by_sql(*partition_by, expr)}"
-            else:
-                window = order_by_sql
-                count_window = generate_partition_by_sql(expr)
+            window_kwargs: WindowExpressionKwargs = {
+                "partition_by": partition_by,
+                "order_by": (expr,),
+                "descending": descending,
+                "nulls_last": True,
+            }
+            count_window_kwargs: WindowExpressionKwargs = {
+                "partition_by": (*partition_by, expr)
+            }
             if method == "max":
                 rank_expr = (
-                    SQLExpression(f"{func} OVER ({window})")
-                    + SQLExpression(f"{count_expr} over ({count_window})")
+                    window_expression(func, **window_kwargs)
+                    + window_expression(count_expr, **count_window_kwargs)
                     - lit(1)
                 )
             elif method == "average":
-                rank_expr = SQLExpression(f"{func} OVER ({window})") + (
-                    SQLExpression(f"{count_expr} over ({count_window})") - lit(1)
+                rank_expr = window_expression(func, **window_kwargs) + (
+                    window_expression(count_expr, **count_window_kwargs) - lit(1)
                 ) / lit(2.0)
             else:
-                rank_expr = SQLExpression(f"{func} OVER ({window})")
+                rank_expr = window_expression(func, **window_kwargs)
             return when(expr.isnotnull(), rank_expr)
 
         def _unpartitioned_rank(expr: Expression) -> Expression:
-            return _rank(expr, descending=descending)
+            return _rank(expr, partition_by=(), descending=descending)
 
         def _partitioned_rank(
             df: DuckDBLazyFrame, inputs: DuckDBWindowInputs
diff --git a/narwhals/_duckdb/namespace.py b/narwhals/_duckdb/namespace.py
index 97a28be9..e4a05d54 100644
--- a/narwhals/_duckdb/namespace.py
+++ b/narwhals/_duckdb/namespace.py
@@ -115,15 +115,25 @@ class DuckDBNamespace(
             version=self._version,
         )
 
-    def all_horizontal(self, *exprs: DuckDBExpr) -> DuckDBExpr:
+    def all_horizontal(self, *exprs: DuckDBExpr, ignore_nulls: bool) -> DuckDBExpr:
         def func(cols: Iterable[Expression]) -> Expression:
-            return reduce(operator.and_, cols)
+            it = (
+                (CoalesceOperator(expr, lit(True)) for expr in cols)  # noqa: FBT003
+                if ignore_nulls
+                else cols
+            )
+            return reduce(operator.and_, it)
 
         return self._expr_from_elementwise(func, *exprs)
 
-    def any_horizontal(self, *exprs: DuckDBExpr) -> DuckDBExpr:
+    def any_horizontal(self, *exprs: DuckDBExpr, ignore_nulls: bool) -> DuckDBExpr:
         def func(cols: Iterable[Expression]) -> Expression:
-            return reduce(operator.or_, cols)
+            it = (
+                (CoalesceOperator(expr, lit(False)) for expr in cols)  # noqa: FBT003
+                if ignore_nulls
+                else cols
+            )
+            return reduce(operator.or_, it)
 
         return self._expr_from_elementwise(func, *exprs)
 
diff --git a/narwhals/_duckdb/typing.py b/narwhals/_duckdb/typing.py
new file mode 100644
index 00000000..cc7a6dcf
--- /dev/null
+++ b/narwhals/_duckdb/typing.py
@@ -0,0 +1,18 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, TypedDict
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+    from duckdb import Expression
+
+
+class WindowExpressionKwargs(TypedDict, total=False):
+    partition_by: Sequence[str | Expression]
+    order_by: Sequence[str | Expression]
+    rows_start: str
+    rows_end: str
+    descending: bool
+    nulls_last: bool
+    ignore_nulls: bool
diff --git a/narwhals/_duckdb/utils.py b/narwhals/_duckdb/utils.py
index c5d4872a..c006a413 100644
--- a/narwhals/_duckdb/utils.py
+++ b/narwhals/_duckdb/utils.py
@@ -8,6 +8,8 @@ import duckdb
 from narwhals._utils import Version, isinstance_or_issubclass
 
 if TYPE_CHECKING:
+    from collections.abc import Sequence
+
     from duckdb import DuckDBPyRelation, Expression
     from duckdb.typing import DuckDBPyType
 
@@ -272,16 +274,59 @@ def narwhals_to_native_dtype(dtype: IntoDType, version: Version) -> str:  # noqa
     raise AssertionError(msg)
 
 
+def parse_into_expression(into_expression: str | Expression) -> Expression:
+    return col(into_expression) if isinstance(into_expression, str) else into_expression
+
+
 def generate_partition_by_sql(*partition_by: str | Expression) -> str:
     if not partition_by:
         return ""
-    by_sql = ", ".join([f"{col(x) if isinstance(x, str) else x}" for x in partition_by])
+    by_sql = ", ".join([f"{parse_into_expression(x)}" for x in partition_by])
     return f"partition by {by_sql}"
 
 
-def generate_order_by_sql(*order_by: str, ascending: bool) -> str:
+def generate_order_by_sql(
+    *order_by: str | Expression, ascending: bool, nulls_first: bool
+) -> str:
+    if not order_by:
+        return ""
+    nulls = "nulls first" if nulls_first else "nulls last"
     if ascending:
-        by_sql = ", ".join([f"{col(x)} asc nulls first" for x in order_by])
+        by_sql = ", ".join([f"{parse_into_expression(x)} asc {nulls}" for x in order_by])
     else:
-        by_sql = ", ".join([f"{col(x)} desc nulls last" for x in order_by])
+        by_sql = ", ".join([f"{parse_into_expression(x)} desc {nulls}" for x in order_by])
     return f"order by {by_sql}"
+
+
+def window_expression(
+    expr: Expression,
+    partition_by: Sequence[str | Expression] = (),
+    order_by: Sequence[str | Expression] = (),
+    rows_start: str = "",
+    rows_end: str = "",
+    *,
+    descending: bool = False,
+    nulls_last: bool = False,
+    ignore_nulls: bool = False,
+) -> Expression:
+    # TODO(unassigned): Replace with `duckdb.WindowExpression` when they release it.
+    # https://github.com/duckdb/duckdb/discussions/14725#discussioncomment-11200348
+    try:
+        from duckdb import SQLExpression
+    except ModuleNotFoundError as exc:  # pragma: no cover
+        msg = f"DuckDB>=1.3.0 is required for this operation. Found: DuckDB {duckdb.__version__}"
+        raise NotImplementedError(msg) from exc
+    pb = generate_partition_by_sql(*partition_by)
+    ob = generate_order_by_sql(
+        *order_by, ascending=not descending, nulls_first=not nulls_last
+    )
+
+    if rows_start and rows_end:
+        rows = f"rows between {rows_start} and {rows_end}"
+    elif rows_start or rows_end:  # pragma: no cover
+        msg = "Either both `rows_start` and `rows_end` must be specified, or neither."
+    else:
+        rows = ""
+
+    func = f"{str(expr).removesuffix(')')} ignore nulls)" if ignore_nulls else str(expr)
+    return SQLExpression(f"{func} over ({pb} {ob} {rows})")
diff --git a/narwhals/_ibis/namespace.py b/narwhals/_ibis/namespace.py
index 4530b05c..03fa7da2 100644
--- a/narwhals/_ibis/namespace.py
+++ b/narwhals/_ibis/namespace.py
@@ -100,15 +100,25 @@ class IbisNamespace(LazyNamespace[IbisLazyFrame, IbisExpr, "ir.Table"]):
             version=self._version,
         )
 
-    def all_horizontal(self, *exprs: IbisExpr) -> IbisExpr:
+    def all_horizontal(self, *exprs: IbisExpr, ignore_nulls: bool) -> IbisExpr:
         def func(cols: Iterable[ir.Value]) -> ir.Value:
-            return reduce(operator.and_, cols)
+            it = (
+                (col.fill_null(lit(True)) for col in cols)  # noqa: FBT003
+                if ignore_nulls
+                else cols
+            )
+            return reduce(operator.and_, it)
 
         return self._expr_from_callable(func, *exprs)
 
-    def any_horizontal(self, *exprs: IbisExpr) -> IbisExpr:
+    def any_horizontal(self, *exprs: IbisExpr, ignore_nulls: bool) -> IbisExpr:
         def func(cols: Iterable[ir.Value]) -> ir.Value:
-            return reduce(operator.or_, cols)
+            it = (
+                (col.fill_null(lit(False)) for col in cols)  # noqa: FBT003
+                if ignore_nulls
+                else cols
+            )
+            return reduce(operator.or_, it)
 
         return self._expr_from_callable(func, *exprs)
 
diff --git a/narwhals/_pandas_like/dataframe.py b/narwhals/_pandas_like/dataframe.py
index 4d9d70a3..738292bd 100644
--- a/narwhals/_pandas_like/dataframe.py
+++ b/narwhals/_pandas_like/dataframe.py
@@ -406,7 +406,8 @@ class PandasLikeDataFrame(
                 self.native.dropna(axis=0), validate_column_names=False
             )
         plx = self.__narwhals_namespace__()
-        return self.filter(~plx.any_horizontal(plx.col(*subset).is_null()))
+        mask = ~plx.any_horizontal(plx.col(*subset).is_null(), ignore_nulls=True)
+        return self.filter(mask)
 
     def estimated_size(self, unit: SizeUnit) -> int | float:
         sz = self.native.memory_usage(deep=True).sum()
diff --git a/narwhals/_pandas_like/namespace.py b/narwhals/_pandas_like/namespace.py
index c4f2b9e6..6e442ffc 100644
--- a/narwhals/_pandas_like/namespace.py
+++ b/narwhals/_pandas_like/namespace.py
@@ -131,11 +131,29 @@ class PandasLikeNamespace(
             context=self,
         )
 
-    def all_horizontal(self, *exprs: PandasLikeExpr) -> PandasLikeExpr:
+    def all_horizontal(
+        self, *exprs: PandasLikeExpr, ignore_nulls: bool
+    ) -> PandasLikeExpr:
         def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
             align = self._series._align_full_broadcast
-            series = align(*(s for _expr in exprs for s in _expr(df)))
-            return [reduce(operator.and_, series)]
+            series = [s for _expr in exprs for s in _expr(df)]
+            if not ignore_nulls and any(
+                s.native.dtype == "object" and s.is_null().any() for s in series
+            ):
+                # classical NumPy boolean columns don't support missing values, so
+                # only do the full scan with `is_null` if we have `object` dtype.
+                msg = "Cannot use `ignore_nulls=False` in `all_horizontal` for non-nullable NumPy-backed pandas Series when nulls are present."
+                raise ValueError(msg)
+            it = (
+                (
+                    # NumPy-backed 'bool' dtype can't contain nulls so doesn't need filling.
+                    s if s.native.dtype == "bool" else s.fill_null(True, None, None)  # noqa: FBT003
+                    for s in series
+                )
+                if ignore_nulls
+                else iter(series)
+            )
+            return [reduce(operator.and_, align(*it))]
 
         return self._expr._from_callable(
             func=func,
@@ -146,11 +164,29 @@ class PandasLikeNamespace(
             context=self,
         )
 
-    def any_horizontal(self, *exprs: PandasLikeExpr) -> PandasLikeExpr:
+    def any_horizontal(
+        self, *exprs: PandasLikeExpr, ignore_nulls: bool
+    ) -> PandasLikeExpr:
         def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
             align = self._series._align_full_broadcast
-            series = align(*(s for _expr in exprs for s in _expr(df)))
-            return [reduce(operator.or_, series)]
+            series = [s for _expr in exprs for s in _expr(df)]
+            if not ignore_nulls and any(
+                s.native.dtype == "object" and s.is_null().any() for s in series
+            ):
+                # classical NumPy boolean columns don't support missing values, so
+                # only do the full scan with `is_null` if we have `object` dtype.
+                msg = "Cannot use `ignore_nulls=False` in `any_horizontal` for non-nullable NumPy-backed pandas Series when nulls are present."
+                raise ValueError(msg)
+            it = (
+                (
+                    # NumPy-backed 'bool' dtype can't contain nulls so doesn't need filling.
+                    s if s.native.dtype == "bool" else s.fill_null(False, None, None)  # noqa: FBT003
+                    for s in series
+                )
+                if ignore_nulls
+                else iter(series)
+            )
+            return [reduce(operator.or_, align(*it))]
 
         return self._expr._from_callable(
             func=func,
diff --git a/narwhals/_polars/namespace.py b/narwhals/_polars/namespace.py
index c2e5c241..f5a78623 100644
--- a/narwhals/_polars/namespace.py
+++ b/narwhals/_polars/namespace.py
@@ -28,8 +28,6 @@ class PolarsNamespace:
     all: Method[PolarsExpr]
     col: Method[PolarsExpr]
     exclude: Method[PolarsExpr]
-    all_horizontal: Method[PolarsExpr]
-    any_horizontal: Method[PolarsExpr]
     sum_horizontal: Method[PolarsExpr]
     min_horizontal: Method[PolarsExpr]
     max_horizontal: Method[PolarsExpr]
@@ -125,12 +123,24 @@ class PolarsNamespace:
     def len(self) -> PolarsExpr:
         if self._backend_version < (0, 20, 5):
             return self._expr(
-                pl.count().alias("len"),
-                version=self._version,
-                backend_version=self._backend_version,
+                pl.count().alias("len"), self._version, self._backend_version
             )
+        return self._expr(pl.len(), self._version, self._backend_version)
+
+    def all_horizontal(self, *exprs: PolarsExpr, ignore_nulls: bool) -> PolarsExpr:
+        it = (expr.fill_null(True) for expr in exprs) if ignore_nulls else iter(exprs)  # noqa: FBT003
+        return self._expr(
+            pl.all_horizontal(*(expr.native for expr in it)),
+            self._version,
+            self._backend_version,
+        )
+
+    def any_horizontal(self, *exprs: PolarsExpr, ignore_nulls: bool) -> PolarsExpr:
+        it = (expr.fill_null(False) for expr in exprs) if ignore_nulls else iter(exprs)  # noqa: FBT003
         return self._expr(
-            pl.len(), version=self._version, backend_version=self._backend_version
+            pl.any_horizontal(*(expr.native for expr in it)),
+            self._version,
+            self._backend_version,
         )
 
     def concat(
diff --git a/narwhals/_spark_like/namespace.py b/narwhals/_spark_like/namespace.py
index f2b6ec84..6f149c8b 100644
--- a/narwhals/_spark_like/namespace.py
+++ b/narwhals/_spark_like/namespace.py
@@ -131,15 +131,25 @@ class SparkLikeNamespace(
             implementation=self._implementation,
         )
 
-    def all_horizontal(self, *exprs: SparkLikeExpr) -> SparkLikeExpr:
+    def all_horizontal(self, *exprs: SparkLikeExpr, ignore_nulls: bool) -> SparkLikeExpr:
         def func(cols: Iterable[Column]) -> Column:
-            return reduce(operator.and_, cols)
+            it = (
+                (self._F.coalesce(col, self._F.lit(True)) for col in cols)  # noqa: FBT003
+                if ignore_nulls
+                else cols
+            )
+            return reduce(operator.and_, it)
 
         return self._expr_from_callable(func, *exprs)
 
-    def any_horizontal(self, *exprs: SparkLikeExpr) -> SparkLikeExpr:
+    def any_horizontal(self, *exprs: SparkLikeExpr, ignore_nulls: bool) -> SparkLikeExpr:
         def func(cols: Iterable[Column]) -> Column:
-            return reduce(operator.or_, cols)
+            it = (
+                (self._F.coalesce(col, self._F.lit(False)) for col in cols)  # noqa: FBT003
+                if ignore_nulls
+                else cols
+            )
+            return reduce(operator.or_, it)
 
         return self._expr_from_callable(func, *exprs)
 
diff --git a/narwhals/dataframe.py b/narwhals/dataframe.py
index 24bf3eea..b3692cb3 100644
--- a/narwhals/dataframe.py
+++ b/narwhals/dataframe.py
@@ -209,7 +209,7 @@ class BaseFrame(Generic[_FrameT]):
                 for name, v in constraints.items()
             )
             predicate = plx.all_horizontal(
-                *chain(compliant_predicates, compliant_constraints)
+                *chain(compliant_predicates, compliant_constraints), ignore_nulls=False
             )
         return self._with_compliant(self._compliant_frame.filter(predicate))
 
diff --git a/narwhals/functions.py b/narwhals/functions.py
index 18bdb726..9e5db828 100644
--- a/narwhals/functions.py
+++ b/narwhals/functions.py
@@ -3,6 +3,7 @@ from __future__ import annotations
 import platform
 import sys
 from collections.abc import Iterable, Mapping, Sequence
+from functools import partial
 from importlib.metadata import version
 from typing import TYPE_CHECKING, Any, Literal, cast
 
@@ -1466,7 +1467,7 @@ def max_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:
 
 class When:
     def __init__(self, *predicates: IntoExpr | Iterable[IntoExpr]) -> None:
-        self._predicate = all_horizontal(*flatten(predicates))
+        self._predicate = all_horizontal(*flatten(predicates), ignore_nulls=False)
 
     def then(self, value: IntoExpr | NonNestedLiteral | _1DArray) -> Then:
         kind = ExprKind.from_into_expr(value, str_as_lit=False)
@@ -1570,16 +1571,20 @@ def when(*predicates: IntoExpr | Iterable[IntoExpr]) -> When:
     return When(*predicates)
 
 
-def all_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:
+def all_horizontal(
+    *exprs: IntoExpr | Iterable[IntoExpr], ignore_nulls: bool | None = None
+) -> Expr:
     r"""Compute the bitwise AND horizontally across columns.
 
-    [Kleene Logic](https://en.wikipedia.org/wiki/Three-valued_logic)
-    is followed, except for pandas' classical NumPy types which can't hold null
-    values, see [Boolean columns](../concepts/boolean.md).
-
     Arguments:
         exprs: Name(s) of the columns to use in the aggregation function. Accepts
             expression input.
+        ignore_nulls: Whether to ignore nulls:
+
+            - If `True`, null values are ignored. If there are no elements, the result
+              is `True`.
+            - If `False` (default), Kleene logic is followed. Note that this is not allowed for
+              pandas with classical NumPy dtypes when null values are present.
 
     Returns:
         A new expression.
@@ -1593,7 +1598,9 @@ def all_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:
         ...     "b": [False, True, True, None, None, None],
         ... }
         >>> df_native = pa.table(data)
-        >>> nw.from_native(df_native).select("a", "b", all=nw.all_horizontal("a", "b"))
+        >>> nw.from_native(df_native).select(
+        ...     "a", "b", all=nw.all_horizontal("a", "b", ignore_nulls=False)
+        ... )
         ┌─────────────────────────────────────────┐
         |           Narwhals DataFrame            |
         |-----------------------------------------|
@@ -1611,10 +1618,19 @@ def all_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:
     if not exprs:
         msg = "At least one expression must be passed to `all_horizontal`"
         raise ValueError(msg)
+    if ignore_nulls is None:
+        issue_deprecation_warning(
+            "`ignore_nulls` will become a required argument in Narwhals 2.0. Please specify `ignore_nulls=True` or `ignore_nulls=False` to silence this warning.",
+            _version="1.45",
+        )
+        ignore_nulls = False
     flat_exprs = flatten(exprs)
     return Expr(
         lambda plx: apply_n_ary_operation(
-            plx, plx.all_horizontal, *flat_exprs, str_as_lit=False
+            plx,
+            partial(plx.all_horizontal, ignore_nulls=ignore_nulls),
+            *flat_exprs,
+            str_as_lit=False,
         ),
         ExprMetadata.from_horizontal_op(*flat_exprs),
     )
@@ -1659,16 +1675,20 @@ def lit(value: NonNestedLiteral, dtype: IntoDType | None = None) -> Expr:
     return Expr(lambda plx: plx.lit(value, dtype), ExprMetadata.literal())
 
 
-def any_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:
+def any_horizontal(
+    *exprs: IntoExpr | Iterable[IntoExpr], ignore_nulls: bool | None = None
+) -> Expr:
     r"""Compute the bitwise OR horizontally across columns.
 
-    [Kleene Logic](https://en.wikipedia.org/wiki/Three-valued_logic)
-    is followed, except for pandas' classical NumPy types which can't hold null
-    values, see [Boolean columns](../concepts/boolean.md).
-
     Arguments:
         exprs: Name(s) of the columns to use in the aggregation function. Accepts
             expression input.
+        ignore_nulls: Whether to ignore nulls:
+
+            - If `True`, null values are ignored. If there are no elements, the result
+              is `False`.
+            - If `False` (default), Kleene logic is followed. Note that this is not allowed for
+              pandas with classical NumPy dtypes when null values are present.
 
     Returns:
         A new expression.
@@ -1682,7 +1702,9 @@ def any_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:
         ...     "b": [False, True, True, None, None, None],
         ... }
         >>> df_native = pl.DataFrame(data)
-        >>> nw.from_native(df_native).select("a", "b", any=nw.any_horizontal("a", "b"))
+        >>> nw.from_native(df_native).select(
+        ...     "a", "b", any=nw.any_horizontal("a", "b", ignore_nulls=False)
+        ... )
         ┌─────────────────────────┐
         |   Narwhals DataFrame    |
         |-------------------------|
@@ -1704,10 +1726,19 @@ def any_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:
     if not exprs:
         msg = "At least one expression must be passed to `any_horizontal`"
         raise ValueError(msg)
+    if ignore_nulls is None:
+        issue_deprecation_warning(
+            "`ignore_nulls` will become a required argument in Narwhals 2.0. Please specify `ignore_nulls=True` or `ignore_nulls=False` to silence this warning.",
+            _version="1.45",
+        )
+        ignore_nulls = False
     flat_exprs = flatten(exprs)
     return Expr(
         lambda plx: apply_n_ary_operation(
-            plx, plx.any_horizontal, *flat_exprs, str_as_lit=False
+            plx,
+            partial(plx.any_horizontal, ignore_nulls=ignore_nulls),
+            *flat_exprs,
+            str_as_lit=False,
         ),
         ExprMetadata.from_horizontal_op(*flat_exprs),
     )
diff --git a/narwhals/stable/v1/__init__.py b/narwhals/stable/v1/__init__.py
index 1b28570a..706b5e64 100644
--- a/narwhals/stable/v1/__init__.py
+++ b/narwhals/stable/v1/__init__.py
@@ -1335,38 +1335,46 @@ def sum_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:
     return _stableify(nw.sum_horizontal(*exprs))
 
 
-def all_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:
+def all_horizontal(
+    *exprs: IntoExpr | Iterable[IntoExpr], ignore_nulls: bool = False
+) -> Expr:
     r"""Compute the bitwise AND horizontally across columns.
 
-    [Kleene Logic](https://en.wikipedia.org/wiki/Three-valued_logic)
-    is followed, except for pandas' classical NumPy types which can't hold null
-    values, see [Boolean columns](../concepts/boolean.md).
-
     Arguments:
         exprs: Name(s) of the columns to use in the aggregation function. Accepts
             expression input.
+        ignore_nulls: Whether to ignore nulls:
+
+            - If `True`, null values are ignored. If there are no elements, the result
+              is `True`.
+            - If `False` (default), Kleene logic is followed. Note that this is not allowed for
+              pandas with classical NumPy dtypes when null values are present.
 
     Returns:
         A new expression.
     """
-    return _stableify(nw.all_horizontal(*exprs))
+    return _stableify(nw.all_horizontal(*exprs, ignore_nulls=ignore_nulls))
 
 
-def any_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:
+def any_horizontal(
+    *exprs: IntoExpr | Iterable[IntoExpr], ignore_nulls: bool = False
+) -> Expr:
     r"""Compute the bitwise OR horizontally across columns.
 
-    [Kleene Logic](https://en.wikipedia.org/wiki/Three-valued_logic)
-    is followed, except for pandas' classical NumPy types which can't hold null
-    values, see [Boolean columns](../concepts/boolean.md).
-
     Arguments:
         exprs: Name(s) of the columns to use in the aggregation function. Accepts
             expression input.
+        ignore_nulls: Whether to ignore nulls:
+
+            - If `True`, null values are ignored. If there are no elements, the result
+              is `False`.
+            - If `False` (default), Kleene logic is followed. Note that this is not allowed for
+              pandas with classical NumPy dtypes when null values are present.
 
     Returns:
         A new expression.
     """
-    return _stableify(nw.any_horizontal(*exprs))
+    return _stableify(nw.any_horizontal(*exprs, ignore_nulls=ignore_nulls))
 
 
 def mean_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:


Test patch:
diff --git a/tests/expr_and_series/all_horizontal_test.py b/tests/expr_and_series/all_horizontal_test.py
index 70597aa3..33a7e219 100644
--- a/tests/expr_and_series/all_horizontal_test.py
+++ b/tests/expr_and_series/all_horizontal_test.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from contextlib import nullcontext as does_not_raise
 from typing import Any
 
 import pytest
@@ -8,38 +9,75 @@ import narwhals as nw
 from tests.utils import POLARS_VERSION, Constructor, ConstructorEager, assert_equal_data
 
 
-@pytest.mark.parametrize("expr1", ["a", nw.col("a")])
-@pytest.mark.parametrize("expr2", ["b", nw.col("b")])
-def test_allh(constructor: Constructor, expr1: Any, expr2: Any) -> None:
+def test_allh(constructor: Constructor) -> None:
     data = {"a": [False, False, True], "b": [False, True, True]}
     df = nw.from_native(constructor(data))
-    result = df.select(all=nw.all_horizontal(expr1, expr2))
+    result = df.select(all=nw.all_horizontal("a", nw.col("b"), ignore_nulls=True))
 
     expected = {"all": [False, False, True]}
     assert_equal_data(result, expected)
 
 
+def test_anyh_ignore_nulls(constructor: Constructor) -> None:
+    if "dask" in str(constructor):
+        # Dask infers `[True, None, None, None]` as `object` dtype, and then `__or__` fails.
+        # test it below separately
+        pytest.skip()
+    data = {"a": [True, True, False], "b": [True, None, None]}
+    df = nw.from_native(constructor(data))
+    result = df.select(any=nw.all_horizontal("a", "b", ignore_nulls=True))
+    expected = [True, True, False]
+    assert_equal_data(result, {"any": expected})
+
+
 def test_allh_kleene(constructor: Constructor, request: pytest.FixtureRequest) -> None:
     if "cudf" in str(constructor):
         # https://github.com/rapidsai/cudf/issues/19171
         request.applymarker(pytest.mark.xfail)
     if "dask" in str(constructor):
         # Dask infers `[True, None, None, None]` as `object` dtype, and then `__or__` fails.
-        request.applymarker(pytest.mark.xfail)
+        # test it below separately
+        pytest.skip()
+    context = (
+        pytest.raises(ValueError, match="ignore_nulls")
+        if "pandas_constructor" in str(constructor)
+        else does_not_raise()
+    )
     data = {"a": [True, True, False], "b": [True, None, None]}
     df = nw.from_native(constructor(data))
-    result = df.select(all=nw.all_horizontal("a", "b"))
-    if any(x in str(constructor) for x in ("pandas_constructor",)):
-        expected: list[bool | None] = [True, False, False]
-    else:
+    with context:
+        result = df.select(all=nw.all_horizontal("a", "b", ignore_nulls=False))
         expected = [True, None, False]
+        assert_equal_data(result, {"all": expected})
+
+
+def test_anyh_dask(constructor: Constructor) -> None:
+    if "dask" not in str(constructor):
+        pytest.skip()
+    import dask.dataframe as dd
+    import pandas as pd
+
+    data = {"a": [True, True, False], "b": [True, None, None]}
+    df = nw.from_native(dd.from_pandas(pd.DataFrame(data, dtype="Boolean[pyarrow]")))
+    result = df.select(all=nw.all_horizontal("a", "b", ignore_nulls=True))
+    expected: list[bool | None] = [True, True, False]
+    assert_equal_data(result, {"all": expected})
+    result = df.select(all=nw.all_horizontal("a", "b", ignore_nulls=False))
+    expected = [True, None, False]
+    assert_equal_data(result, {"all": expected})
+
+    # No nulls, NumPy-backed
+    data = {"a": [True, True, False], "b": [True, False, False]}
+    df = nw.from_native(dd.from_pandas(pd.DataFrame(data)))
+    result = df.select(all=nw.all_horizontal("a", "b", ignore_nulls=True))
+    expected = [True, False, False]
     assert_equal_data(result, {"all": expected})
 
 
 def test_allh_series(constructor_eager: ConstructorEager) -> None:
     data = {"a": [False, False, True], "b": [False, True, True]}
     df = nw.from_native(constructor_eager(data), eager_only=True)
-    result = df.select(all=nw.all_horizontal(df["a"], df["b"]))
+    result = df.select(all=nw.all_horizontal(df["a"], df["b"], ignore_nulls=True))
 
     expected = {"all": [False, False, True]}
     assert_equal_data(result, expected)
@@ -48,10 +86,10 @@ def test_allh_series(constructor_eager: ConstructorEager) -> None:
 def test_allh_all(constructor: Constructor) -> None:
     data = {"a": [False, False, True], "b": [False, True, True]}
     df = nw.from_native(constructor(data))
-    result = df.select(all=nw.all_horizontal(nw.all()))
+    result = df.select(all=nw.all_horizontal(nw.all(), ignore_nulls=True))
     expected = {"all": [False, False, True]}
     assert_equal_data(result, expected)
-    result = df.select(nw.all_horizontal(nw.all()))
+    result = df.select(nw.all_horizontal(nw.all(), ignore_nulls=True))
     expected = {"a": [False, False, True]}
     assert_equal_data(result, expected)
 
@@ -61,10 +99,10 @@ def test_allh_nth(constructor: Constructor) -> None:
         pytest.skip()
     data = {"a": [False, False, True], "b": [False, True, True]}
     df = nw.from_native(constructor(data))
-    result = df.select(nw.all_horizontal(nw.nth(0, 1)))
+    result = df.select(nw.all_horizontal(nw.nth(0, 1), ignore_nulls=True))
     expected = {"a": [False, False, True]}
     assert_equal_data(result, expected)
-    result = df.select(nw.all_horizontal(nw.col("a"), nw.nth(0)))
+    result = df.select(nw.all_horizontal(nw.col("a"), nw.nth(0), ignore_nulls=True))
     expected = {"a": [False, False, True]}
     assert_equal_data(result, expected)
 
@@ -79,11 +117,11 @@ def test_allh_iterator(constructor: Constructor) -> None:
     expr_items = [("a", 3), ("b", "b")]
     expected = {"a": [3], "b": ["b"]}
 
-    eager = nw.all_horizontal(list(iter_eq(expr_items)))
+    eager = nw.all_horizontal(list(iter_eq(expr_items)), ignore_nulls=True)
     assert_equal_data(df.filter(eager), expected)
-    unpacked = nw.all_horizontal(*iter_eq(expr_items))
+    unpacked = nw.all_horizontal(*iter_eq(expr_items), ignore_nulls=True)
     assert_equal_data(df.filter(unpacked), expected)
-    lazy = nw.all_horizontal(iter_eq(expr_items))
+    lazy = nw.all_horizontal(iter_eq(expr_items), ignore_nulls=True)
 
     assert_equal_data(df.filter(lazy), expected)
     assert_equal_data(df.filter(lazy), expected)
diff --git a/tests/expr_and_series/any_horizontal_test.py b/tests/expr_and_series/any_horizontal_test.py
index 88e97dc6..04f0cba7 100644
--- a/tests/expr_and_series/any_horizontal_test.py
+++ b/tests/expr_and_series/any_horizontal_test.py
@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any
+from contextlib import nullcontext as does_not_raise
 
 import pytest
 
@@ -8,12 +8,10 @@ import narwhals as nw
 from tests.utils import Constructor, assert_equal_data
 
 
-@pytest.mark.parametrize("expr1", ["a", nw.col("a")])
-@pytest.mark.parametrize("expr2", ["b", nw.col("b")])
-def test_anyh(constructor: Constructor, expr1: Any, expr2: Any) -> None:
+def test_anyh(constructor: Constructor) -> None:
     data = {"a": [False, False, True], "b": [False, True, True]}
     df = nw.from_native(constructor(data))
-    result = df.select(any=nw.any_horizontal(expr1, expr2))
+    result = df.select(any=nw.any_horizontal(nw.col("a"), "b", ignore_nulls=True))
 
     expected = {"any": [False, True, True]}
     assert_equal_data(result, expected)
@@ -25,23 +23,65 @@ def test_anyh_kleene(constructor: Constructor, request: pytest.FixtureRequest) -
         request.applymarker(pytest.mark.xfail)
     if "dask" in str(constructor):
         # Dask infers `[True, None, None, None]` as `object` dtype, and then `__or__` fails.
-        request.applymarker(pytest.mark.xfail)
+        # test it below separately
+        pytest.skip()
+    context = (
+        pytest.raises(ValueError, match="ignore_nulls")
+        if "pandas_constructor" in str(constructor)
+        else does_not_raise()
+    )
     data = {"a": [True, True, False], "b": [True, None, None]}
     df = nw.from_native(constructor(data))
-    result = df.select(any=nw.any_horizontal("a", "b"))
-    if any(x in str(constructor) for x in ("pandas_constructor",)):
-        expected: list[bool | None] = [True, True, False]
-    else:
+    with context:
+        result = df.select(any=nw.any_horizontal("a", "b", ignore_nulls=False))
         expected = [True, True, None]
+        assert_equal_data(result, {"any": expected})
+
+
+def test_anyh_ignore_nulls(constructor: Constructor) -> None:
+    if "dask" in str(constructor):
+        # Dask infers `[True, None, None, None]` as `object` dtype, and then `__or__` fails.
+        # test it below separately
+        pytest.skip()
+    data = {"a": [True, True, False], "b": [True, None, None]}
+    df = nw.from_native(constructor(data))
+    result = df.select(any=nw.any_horizontal("a", "b", ignore_nulls=True))
+    expected = [True, True, False]
+    assert_equal_data(result, {"any": expected})
+
+
+def test_anyh_dask(constructor: Constructor) -> None:
+    if "dask" not in str(constructor):
+        pytest.skip()
+    import dask.dataframe as dd
+    import pandas as pd
+
+    data = {"a": [True, True, False], "b": [True, None, None]}
+    df = nw.from_native(dd.from_pandas(pd.DataFrame(data, dtype="Boolean[pyarrow]")))
+    result = df.select(any=nw.any_horizontal("a", "b", ignore_nulls=True))
+    expected: list[bool | None] = [True, True, False]
+    assert_equal_data(result, {"any": expected})
+    result = df.select(any=nw.any_horizontal("a", "b", ignore_nulls=False))
+    expected = [True, True, None]
+    assert_equal_data(result, {"any": expected})
+
+    # No nulls, NumPy-backed
+    data = {"a": [True, True, False], "b": [True, False, False]}
+    df = nw.from_native(dd.from_pandas(pd.DataFrame(data)))
+    result = df.select(any=nw.any_horizontal("a", "b", ignore_nulls=True))
+    expected = [True, True, False]
     assert_equal_data(result, {"any": expected})
 
 
 def test_anyh_all(constructor: Constructor) -> None:
+    if "dask" in str(constructor):
+        # Can't use `ignore_nulls` for NumPy-backed Dask, test it separately below
+        pytest.skip()
     data = {"a": [False, False, True], "b": [False, True, True]}
     df = nw.from_native(constructor(data))
-    result = df.select(any=nw.any_horizontal(nw.all()))
+    result = df.select(any=nw.any_horizontal(nw.all(), ignore_nulls=False))
     expected = {"any": [False, True, True]}
     assert_equal_data(result, expected)
-    result = df.select(nw.any_horizontal(nw.all()))
+    result = df.select(nw.any_horizontal(nw.all(), ignore_nulls=False))
     expected = {"a": [False, True, True]}
     assert_equal_data(result, expected)
diff --git a/tests/expr_and_series/horizontal_broadcasts_test.py b/tests/expr_and_series/horizontal_broadcasts_test.py
index 445b9674..6c6bb896 100644
--- a/tests/expr_and_series/horizontal_broadcasts_test.py
+++ b/tests/expr_and_series/horizontal_broadcasts_test.py
@@ -12,8 +12,8 @@ def test_sumh_broadcasting(constructor: Constructor) -> None:
     data = {"a": [1, 2, 3], "b": [4, 5, 6], "i": [0, 1, 2]}
     df = nw.from_native(constructor(data))
     result = df.with_columns(
-        any=nw.any_horizontal(nw.sum("a", "b").cast(nw.Boolean)),
-        all=nw.all_horizontal(nw.sum("a", "b").cast(nw.Boolean)),
+        any=nw.any_horizontal(nw.sum("a", "b").cast(nw.Boolean), ignore_nulls=True),
+        all=nw.all_horizontal(nw.sum("a", "b").cast(nw.Boolean), ignore_nulls=True),
         max=nw.max_horizontal(nw.sum("a"), nw.sum("b")),
         min=nw.min_horizontal(nw.sum("a"), nw.sum("b")),
         sum=nw.sum_horizontal(nw.sum("a"), nw.sum("b")),
diff --git a/tests/v1_test.py b/tests/v1_test.py
index 0ecaa03e..18c057f1 100644
--- a/tests/v1_test.py
+++ b/tests/v1_test.py
@@ -362,6 +362,38 @@ def test_get_level() -> None:
     )
 
 
+def test_any_horizontal() -> None:
+    # here, it defaults to Kleene logic.
+    pytest.importorskip("polars")
+    import polars as pl
+
+    df = nw_v1.from_native(
+        pl.DataFrame({"a": [True, True, False], "b": [True, None, None]})
+    )
+    result = df.select(nw_v1.any_horizontal("a", "b"))
+    expected = {"a": [True, True, None]}
+    assert_equal_data(result, expected)
+    with pytest.deprecated_call(match="ignore_nulls"):
+        result = df.select(nw.any_horizontal("a", "b"))
+    assert_equal_data(result, expected)
+
+
+def test_all_horizontal() -> None:
+    # here, it defaults to Kleene logic.
+    pytest.importorskip("polars")
+    import polars as pl
+
+    df = nw_v1.from_native(
+        pl.DataFrame({"a": [True, True, False], "b": [True, None, None]})
+    )
+    result = df.select(nw_v1.all_horizontal("a", "b"))
+    expected = {"a": [True, None, False]}
+    assert_equal_data(result, expected)
+    with pytest.deprecated_call(match="ignore_nulls"):
+        result = df.select(nw.all_horizontal("a", "b"))
+    assert_equal_data(result, expected)
+
+
 def test_with_row_index(constructor: Constructor) -> None:
     data = {"abc": ["foo", "bars"], "xyz": [100, 200], "const": [42, 42]}