Repository: narwhals-dev/narwhals

Issue:
[Bug]: Run into safety assertion on `reuse_series_implementation` for drop_nulls
### Describe the bug

When executing `DataFrame.drop_nulls(subset=[...])` we run into safety assertion of `reuse_series_implementation` function

### Steps or code to reproduce the bug

Consider the following tests:

```py
@pytest.mark.parametrize(
    ("subset", "expected"),
    [
        ("a", {"a": [1, 2.0, 4.0], "b": [float("nan"), 3.0, 5.0]}),
        (["a"], {"a": [1, 2.0, 4.0], "b": [float("nan"), 3.0, 5.0]}),
        (["a", "b"], {"a": [2.0, 4.0], "b": [3.0, 5.0]})
    ]
)
def test_drop_nulls_subset(constructor: Constructor, subset: str | list[str], expected: dict[str, float]) -> None:
    result = nw.from_native(constructor(data)).drop_nulls(subset=subset)
    compare_dicts(result, expected)
```

It will raise for arrow and pandas-like engines, not for polars or dask when subset has more than one column

### Expected results

To not raise

### Actual results

Raise AssertionError

### Please run narwhals.show_version() and enter the output below.

```shell
Latest narwhals
```


### Relevant log output

_No response_

Discussion:
FBruzzesi: `parsed_exprs[0]._output_names` can easily be None though.

Do you like: 
```diff
- output_names=parsed_exprs[0]._output_names,
+ output_names=(parsed_exprs[0]._output_names or [None])[:1],
```
MarcoGorelli: are you're right - let's make a function then, just like we have for `combine_root_names` one line above? in there we can explain that we follow the right-hand-rule like Polars does
FBruzzesi: > right-hand-rule like Polars does

left-most?

>  let's make a function then

Not sure I follow 100%. `combine_root_names` seems generic-ish.

For this (horizontal) reduction I would expect to have:

- left-most output name if available
- otherwise? left most root name?

MarcoGorelli: yup, left-most, sorry :)

so `parsed_exprs[0]._output_names[:1] if parsed_exprs[0]._output_names is not None else None`

Patch:
diff --git a/narwhals/_arrow/namespace.py b/narwhals/_arrow/namespace.py
index 6f699fa2..e1fb8c6b 100644
--- a/narwhals/_arrow/namespace.py
+++ b/narwhals/_arrow/namespace.py
@@ -16,6 +16,7 @@ from narwhals._arrow.utils import horizontal_concat
 from narwhals._arrow.utils import vertical_concat
 from narwhals._expression_parsing import combine_root_names
 from narwhals._expression_parsing import parse_into_exprs
+from narwhals._expression_parsing import reduce_output_names
 from narwhals.utils import Implementation
 
 if TYPE_CHECKING:
@@ -183,9 +184,7 @@ class ArrowNamespace:
         parsed_exprs = parse_into_exprs(*exprs, namespace=self)
 
         def func(df: ArrowDataFrame) -> list[ArrowSeries]:
-            series = []
-            for _expr in parsed_exprs:
-                series.extend(list(_expr._call(df)))
+            series = (s for _expr in parsed_exprs for s in _expr._call(df))
             return [reduce(lambda x, y: x & y, series)]
 
         return self._create_expr_from_callable(
@@ -193,16 +192,14 @@ class ArrowNamespace:
             depth=max(x._depth for x in parsed_exprs) + 1,
             function_name="all_horizontal",
             root_names=combine_root_names(parsed_exprs),
-            output_names=parsed_exprs[0]._output_names,
+            output_names=reduce_output_names(parsed_exprs),
         )
 
     def any_horizontal(self, *exprs: IntoArrowExpr) -> ArrowExpr:
         parsed_exprs = parse_into_exprs(*exprs, namespace=self)
 
         def func(df: ArrowDataFrame) -> list[ArrowSeries]:
-            series = []
-            for _expr in parsed_exprs:
-                series.extend(list(_expr._call(df)))
+            series = (s for _expr in parsed_exprs for s in _expr._call(df))
             return [reduce(lambda x, y: x | y, series)]
 
         return self._create_expr_from_callable(
@@ -210,16 +207,14 @@ class ArrowNamespace:
             depth=max(x._depth for x in parsed_exprs) + 1,
             function_name="any_horizontal",
             root_names=combine_root_names(parsed_exprs),
-            output_names=parsed_exprs[0]._output_names,
+            output_names=reduce_output_names(parsed_exprs),
         )
 
     def sum_horizontal(self, *exprs: IntoArrowExpr) -> ArrowExpr:
         parsed_exprs = parse_into_exprs(*exprs, namespace=self)
 
         def func(df: ArrowDataFrame) -> list[ArrowSeries]:
-            series = []
-            for _expr in parsed_exprs:
-                series.extend([_series.fill_null(0) for _series in _expr._call(df)])
+            series = (s.fill_null(0) for _expr in parsed_exprs for s in _expr._call(df))
             return [reduce(lambda x, y: x + y, series)]
 
         return self._create_expr_from_callable(
@@ -227,7 +222,7 @@ class ArrowNamespace:
             depth=max(x._depth for x in parsed_exprs) + 1,
             function_name="sum_horizontal",
             root_names=combine_root_names(parsed_exprs),
-            output_names=parsed_exprs[0]._output_names,
+            output_names=reduce_output_names(parsed_exprs),
         )
 
     def mean_horizontal(self, *exprs: IntoArrowExpr) -> IntoArrowExpr:
diff --git a/narwhals/_dask/namespace.py b/narwhals/_dask/namespace.py
index 7d661f06..39c64719 100644
--- a/narwhals/_dask/namespace.py
+++ b/narwhals/_dask/namespace.py
@@ -17,6 +17,7 @@ from narwhals._dask.utils import narwhals_to_native_dtype
 from narwhals._dask.utils import validate_comparand
 from narwhals._expression_parsing import combine_root_names
 from narwhals._expression_parsing import parse_into_exprs
+from narwhals._expression_parsing import reduce_output_names
 
 if TYPE_CHECKING:
     import dask_expr
@@ -155,9 +156,7 @@ class DaskNamespace:
         parsed_exprs = parse_into_exprs(*exprs, namespace=self)
 
         def func(df: DaskLazyFrame) -> list[dask_expr.Series]:
-            series = []
-            for _expr in parsed_exprs:
-                series.extend(list(_expr._call(df)))
+            series = [s for _expr in parsed_exprs for s in _expr._call(df)]
             return [reduce(lambda x, y: x & y, series).rename(series[0].name)]
 
         return DaskExpr(
@@ -165,7 +164,7 @@ class DaskNamespace:
             depth=max(x._depth for x in parsed_exprs) + 1,
             function_name="all_horizontal",
             root_names=combine_root_names(parsed_exprs),
-            output_names=parsed_exprs[0]._output_names,
+            output_names=reduce_output_names(parsed_exprs),
             returns_scalar=False,
             backend_version=self._backend_version,
         )
@@ -174,9 +173,7 @@ class DaskNamespace:
         parsed_exprs = parse_into_exprs(*exprs, namespace=self)
 
         def func(df: DaskLazyFrame) -> list[dask_expr.Series]:
-            series = []
-            for _expr in parsed_exprs:
-                series.extend(list(_expr._call(df)))
+            series = [s for _expr in parsed_exprs for s in _expr._call(df)]
             return [reduce(lambda x, y: x | y, series).rename(series[0].name)]
 
         return DaskExpr(
@@ -184,7 +181,7 @@ class DaskNamespace:
             depth=max(x._depth for x in parsed_exprs) + 1,
             function_name="any_horizontal",
             root_names=combine_root_names(parsed_exprs),
-            output_names=parsed_exprs[0]._output_names,
+            output_names=reduce_output_names(parsed_exprs),
             returns_scalar=False,
             backend_version=self._backend_version,
         )
@@ -193,9 +190,7 @@ class DaskNamespace:
         parsed_exprs = parse_into_exprs(*exprs, namespace=self)
 
         def func(df: DaskLazyFrame) -> list[dask_expr.Series]:
-            series = []
-            for _expr in parsed_exprs:
-                series.extend([_series.fillna(0) for _series in _expr._call(df)])
+            series = [s.fillna(0) for _expr in parsed_exprs for s in _expr._call(df)]
             return [reduce(lambda x, y: x + y, series).rename(series[0].name)]
 
         return DaskExpr(
@@ -203,7 +198,7 @@ class DaskNamespace:
             depth=max(x._depth for x in parsed_exprs) + 1,
             function_name="sum_horizontal",
             root_names=combine_root_names(parsed_exprs),
-            output_names=parsed_exprs[0]._output_names,
+            output_names=reduce_output_names(parsed_exprs),
             returns_scalar=False,
             backend_version=self._backend_version,
         )
diff --git a/narwhals/_expression_parsing.py b/narwhals/_expression_parsing.py
index 35d0f7ec..d281cc94 100644
--- a/narwhals/_expression_parsing.py
+++ b/narwhals/_expression_parsing.py
@@ -319,3 +319,12 @@ def combine_root_names(parsed_exprs: Sequence[CompliantExpr]) -> list[str] | Non
                 root_names = None
                 break
     return root_names
+
+
+def reduce_output_names(parsed_exprs: Sequence[CompliantExpr]) -> list[str] | None:
+    """Returns the left-most output name"""
+    return (
+        parsed_exprs[0]._output_names[:1]
+        if parsed_exprs[0]._output_names is not None
+        else None
+    )
diff --git a/narwhals/_pandas_like/namespace.py b/narwhals/_pandas_like/namespace.py
index 357ef80a..7356524d 100644
--- a/narwhals/_pandas_like/namespace.py
+++ b/narwhals/_pandas_like/namespace.py
@@ -11,6 +11,7 @@ from typing import cast
 from narwhals import dtypes
 from narwhals._expression_parsing import combine_root_names
 from narwhals._expression_parsing import parse_into_exprs
+from narwhals._expression_parsing import reduce_output_names
 from narwhals._pandas_like.dataframe import PandasLikeDataFrame
 from narwhals._pandas_like.expr import PandasLikeExpr
 from narwhals._pandas_like.selectors import PandasSelectorNamespace
@@ -218,9 +219,7 @@ class PandasLikeNamespace:
         parsed_exprs = parse_into_exprs(*exprs, namespace=self)
 
         def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
-            series = []
-            for _expr in parsed_exprs:
-                series.extend([_series.fill_null(0) for _series in _expr._call(df)])
+            series = (s.fill_null(0) for _expr in parsed_exprs for s in _expr._call(df))
             return [reduce(lambda x, y: x + y, series)]
 
         return self._create_expr_from_callable(
@@ -228,16 +227,14 @@ class PandasLikeNamespace:
             depth=max(x._depth for x in parsed_exprs) + 1,
             function_name="sum_horizontal",
             root_names=combine_root_names(parsed_exprs),
-            output_names=parsed_exprs[0]._output_names,
+            output_names=reduce_output_names(parsed_exprs),
         )
 
     def all_horizontal(self, *exprs: IntoPandasLikeExpr) -> PandasLikeExpr:
         parsed_exprs = parse_into_exprs(*exprs, namespace=self)
 
         def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
-            series = []
-            for _expr in parsed_exprs:
-                series.extend(list(_expr._call(df)))
+            series = (s for _expr in parsed_exprs for s in _expr._call(df))
             return [reduce(lambda x, y: x & y, series)]
 
         return self._create_expr_from_callable(
@@ -245,16 +242,14 @@ class PandasLikeNamespace:
             depth=max(x._depth for x in parsed_exprs) + 1,
             function_name="all_horizontal",
             root_names=combine_root_names(parsed_exprs),
-            output_names=parsed_exprs[0]._output_names,
+            output_names=reduce_output_names(parsed_exprs),
         )
 
     def any_horizontal(self, *exprs: IntoPandasLikeExpr) -> PandasLikeExpr:
         parsed_exprs = parse_into_exprs(*exprs, namespace=self)
 
         def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
-            series = []
-            for _expr in parsed_exprs:
-                series.extend(list(_expr._call(df)))
+            series = (s for _expr in parsed_exprs for s in _expr._call(df))
             return [reduce(lambda x, y: x | y, series)]
 
         return self._create_expr_from_callable(
@@ -262,7 +257,7 @@ class PandasLikeNamespace:
             depth=max(x._depth for x in parsed_exprs) + 1,
             function_name="any_horizontal",
             root_names=combine_root_names(parsed_exprs),
-            output_names=parsed_exprs[0]._output_names,
+            output_names=reduce_output_names(parsed_exprs),
         )
 
     def mean_horizontal(self, *exprs: IntoPandasLikeExpr) -> PandasLikeExpr:


Test patch:
diff --git a/tests/frame/drop_nulls_test.py b/tests/frame/drop_nulls_test.py
index 4c227603..9988aa6b 100644
--- a/tests/frame/drop_nulls_test.py
+++ b/tests/frame/drop_nulls_test.py
@@ -21,11 +21,16 @@ def test_drop_nulls(constructor: Constructor) -> None:
     compare_dicts(result, expected)
 
 
-@pytest.mark.parametrize("subset", ["a", ["a"]])
-def test_drop_nulls_subset(constructor: Constructor, subset: str | list[str]) -> None:
+@pytest.mark.parametrize(
+    ("subset", "expected"),
+    [
+        ("a", {"a": [1, 2.0, 4.0], "b": [float("nan"), 3.0, 5.0]}),
+        (["a"], {"a": [1, 2.0, 4.0], "b": [float("nan"), 3.0, 5.0]}),
+        (["a", "b"], {"a": [2.0, 4.0], "b": [3.0, 5.0]}),
+    ],
+)
+def test_drop_nulls_subset(
+    constructor: Constructor, subset: str | list[str], expected: dict[str, float]
+) -> None:
     result = nw.from_native(constructor(data)).drop_nulls(subset=subset)
-    expected = {
-        "a": [1, 2.0, 4.0],
-        "b": [float("nan"), 3.0, 5.0],
-    }
     compare_dicts(result, expected)