Repository: narwhals-dev/narwhals
Pull request #2693 (2025-06-17 16:35:41)

typing: restore NativeSeriesT to EagerDataFrame and EagerNamespace, avoid unnecessary broadcast in eager when/then/otherwise
This addresses part of https://github.com/narwhals-dev/narwhals/issues/2666

## Related
- #2666
- [remove NativeSeriesT from EagerDataFrame / EagerNamespace ](https://github.com/narwhals-dev/narwhals/pull/2662/commits/e19d5e17ce089ecea4944eae6e2753832bb8c730)

## Description

`EagerNamespace` leaves me in covariance/contravariance hell, I'm trying to put together a smaller repro to isolate the issue i'm facing

OK, I've finally fully (I think) made sense of the issues with `EagerNamespace` and why I wasn't able to put `NativeSeriesT` there too

- If you define a protocol `P` generic in `T` and `T` doesn't appear in any of `P`'s method's arguments, then type checkers expect `P` to be covariant in `T`
- If you define a method `foo(self, a: T) -> None`, then type checkers expect `P` to be contravariant in `T`. But, if you instead define it to be `foo(self, a: T | Any) -> None`, then that turns off the type checking for `a` and so type checkers expect `P` to be covariant in `T`. 
- If you then overload `foo` and one of the overloads matches just `T`, then you'll run into a bad place, as type checkers simultaneously expect `P` to be contravariant in `T` and also covariant

In this case, `EagerNamespace.from_native` accepts `NativeSeriesT | Any`, and it's the `| Any` that's the really problematic part. It might as well just be `Any`, at which point we don't need `EagerNamespace` to be generic in `NativeSeriesT` if it's not used anywhere

Having said that, there might also be a bug in MyPy? https://gist.github.com/mypy-play/021d9f12a5e02e1e837767f0e6fd0e4a

I think possible solutions on that side are:
- keep `EagerNamespace` non-generic in `NativeSeriesT` (i.e. status-quo)
- make `EagerNamespace` generic in `NativeSeriesT_contra`, and then `type: ignore` a couple of lines in `_from_native_impl` when `ModinSeries` can't be assigned to `NativeSeries`

dangotbanned: Thanks @MarcoGorelli - love to see all this detail in the description! 😍

Will do my best to review tomorrow, just wanted to say I really appreciate this
dangotbanned: I haven't tried these out, but two ideas I had

## Modin
Change these defs

https://github.com/narwhals-dev/narwhals/blob/f7d94519c73d05993d15195ce1cdb671262b8c86/narwhals/_namespace.py#L113-L117

To this, which *should* make them compatible

```py
    class _ModinDataFrame(NativeFrame, Protocol):
        _pandas_class: type[pd.DataFrame]

    class _ModinSeries(NativeSeries, Protocol):
        _pandas_class: type[pd.Series[Any]]
```

## Variance
For the `TypeVar`(s) or `Protocol`(s) that are problematic, trying defining the `TypeVar`(s) using [`infer_variance`](https://docs.python.org/3/library/typing.html#typing.TypeVar)

```py
from typing import Protocol
from narwhals._typing_compat import TypeVar

class CompliantWhat(Protocol):
    def hello(self) -> str: ...

CompliantWhatT = TypeVar("CompliantWhatT", bound=CompliantWhat, infer_variance=True)
```

 
MarcoGorelli: thanks - i tried the first one but it didn't resolve the issue
MarcoGorelli: thanks @dangotbanned , awesome additions and rewrites!

this helped notice that there was a further chance to improve by avoiding an unnecessary broadcast if `otherwise_value` is a scalar

Patch:
diff --git a/docs/api-reference/expr_str.md b/docs/api-reference/expr_str.md
index bb6abb7e..9fd31704 100644
--- a/docs/api-reference/expr_str.md
+++ b/docs/api-reference/expr_str.md
@@ -18,5 +18,6 @@
         - to_datetime
         - to_lowercase
         - to_uppercase
+        - zfill
       show_source: false
       show_bases: false
diff --git a/docs/api-reference/series_str.md b/docs/api-reference/series_str.md
index 42810161..d2d70e69 100644
--- a/docs/api-reference/series_str.md
+++ b/docs/api-reference/series_str.md
@@ -18,5 +18,6 @@
         - to_datetime
         - to_lowercase
         - to_uppercase
+        - zfill
       show_source: false
       show_bases: false
diff --git a/narwhals/_arrow/dataframe.py b/narwhals/_arrow/dataframe.py
index 4f370d9c..16ff9085 100644
--- a/narwhals/_arrow/dataframe.py
+++ b/narwhals/_arrow/dataframe.py
@@ -8,7 +8,7 @@ import pyarrow as pa
 import pyarrow.compute as pc
 
 from narwhals._arrow.series import ArrowSeries
-from narwhals._arrow.utils import align_series_full_broadcast, native_to_narwhals_dtype
+from narwhals._arrow.utils import native_to_narwhals_dtype
 from narwhals._compliant import EagerDataFrame
 from narwhals._expression_parsing import ExprKind
 from narwhals._utils import (
@@ -74,7 +74,9 @@ if TYPE_CHECKING:
     PromoteOptions: TypeAlias = Literal["none", "default", "permissive"]
 
 
-class ArrowDataFrame(EagerDataFrame["ArrowSeries", "ArrowExpr", "pa.Table"]):
+class ArrowDataFrame(
+    EagerDataFrame["ArrowSeries", "ArrowExpr", "pa.Table", "ChunkedArrayAny"]
+):
     def __init__(
         self,
         native_dataframe: pa.Table,
@@ -330,7 +332,8 @@ class ArrowDataFrame(EagerDataFrame["ArrowSeries", "ArrowExpr", "pa.Table"]):
                 self.native.__class__.from_arrays([]), validate_column_names=False
             )
         names = [s.name for s in new_series]
-        reshaped = align_series_full_broadcast(*new_series)
+        align = new_series[0]._align_full_broadcast
+        reshaped = align(*new_series)
         df = pa.Table.from_arrays([s.native for s in reshaped], names=names)
         return self._with_native(df, validate_column_names=True)
 
diff --git a/narwhals/_arrow/namespace.py b/narwhals/_arrow/namespace.py
index e4817504..84c3475a 100644
--- a/narwhals/_arrow/namespace.py
+++ b/narwhals/_arrow/namespace.py
@@ -12,10 +12,7 @@ from narwhals._arrow.dataframe import ArrowDataFrame
 from narwhals._arrow.expr import ArrowExpr
 from narwhals._arrow.selectors import ArrowSelectorNamespace
 from narwhals._arrow.series import ArrowSeries
-from narwhals._arrow.utils import (
-    align_series_full_broadcast,
-    cast_to_comparable_string_types,
-)
+from narwhals._arrow.utils import cast_to_comparable_string_types
 from narwhals._compliant import CompliantThen, EagerNamespace, EagerWhen
 from narwhals._expression_parsing import (
     combine_alias_output_names,
@@ -26,12 +23,14 @@ from narwhals._utils import Implementation
 if TYPE_CHECKING:
     from collections.abc import Sequence
 
-    from narwhals._arrow.typing import Incomplete
+    from narwhals._arrow.typing import ArrayOrScalar, ChunkedArrayAny, Incomplete
     from narwhals._utils import Version
     from narwhals.typing import IntoDType, NonNestedLiteral
 
 
-class ArrowNamespace(EagerNamespace[ArrowDataFrame, ArrowSeries, ArrowExpr, pa.Table]):
+class ArrowNamespace(
+    EagerNamespace[ArrowDataFrame, ArrowSeries, ArrowExpr, pa.Table, "ChunkedArrayAny"]
+):
     @property
     def _dataframe(self) -> type[ArrowDataFrame]:
         return ArrowDataFrame
@@ -86,7 +85,8 @@ class ArrowNamespace(EagerNamespace[ArrowDataFrame, ArrowSeries, ArrowExpr, pa.T
     def all_horizontal(self, *exprs: ArrowExpr) -> ArrowExpr:
         def func(df: ArrowDataFrame) -> list[ArrowSeries]:
             series = chain.from_iterable(expr(df) for expr in exprs)
-            return [reduce(operator.and_, align_series_full_broadcast(*series))]
+            align = self._series._align_full_broadcast
+            return [reduce(operator.and_, align(*series))]
 
         return self._expr._from_callable(
             func=func,
@@ -100,7 +100,8 @@ class ArrowNamespace(EagerNamespace[ArrowDataFrame, ArrowSeries, ArrowExpr, pa.T
     def any_horizontal(self, *exprs: ArrowExpr) -> ArrowExpr:
         def func(df: ArrowDataFrame) -> list[ArrowSeries]:
             series = chain.from_iterable(expr(df) for expr in exprs)
-            return [reduce(operator.or_, align_series_full_broadcast(*series))]
+            align = self._series._align_full_broadcast
+            return [reduce(operator.or_, align(*series))]
 
         return self._expr._from_callable(
             func=func,
@@ -115,7 +116,8 @@ class ArrowNamespace(EagerNamespace[ArrowDataFrame, ArrowSeries, ArrowExpr, pa.T
         def func(df: ArrowDataFrame) -> list[ArrowSeries]:
             it = chain.from_iterable(expr(df) for expr in exprs)
             series = (s.fill_null(0, strategy=None, limit=None) for s in it)
-            return [reduce(operator.add, align_series_full_broadcast(*series))]
+            align = self._series._align_full_broadcast
+            return [reduce(operator.add, align(*series))]
 
         return self._expr._from_callable(
             func=func,
@@ -131,12 +133,11 @@ class ArrowNamespace(EagerNamespace[ArrowDataFrame, ArrowSeries, ArrowExpr, pa.T
 
         def func(df: ArrowDataFrame) -> list[ArrowSeries]:
             expr_results = list(chain.from_iterable(expr(df) for expr in exprs))
-            series = align_series_full_broadcast(
+            align = self._series._align_full_broadcast
+            series = align(
                 *(s.fill_null(0, strategy=None, limit=None) for s in expr_results)
             )
-            non_na = align_series_full_broadcast(
-                *(1 - s.is_null().cast(int_64) for s in expr_results)
-            )
+            non_na = align(*(1 - s.is_null().cast(int_64) for s in expr_results))
             return [reduce(operator.add, series) / reduce(operator.add, non_na)]
 
         return self._expr._from_callable(
@@ -150,8 +151,9 @@ class ArrowNamespace(EagerNamespace[ArrowDataFrame, ArrowSeries, ArrowExpr, pa.T
 
     def min_horizontal(self, *exprs: ArrowExpr) -> ArrowExpr:
         def func(df: ArrowDataFrame) -> list[ArrowSeries]:
+            align = self._series._align_full_broadcast
             init_series, *series = list(chain.from_iterable(expr(df) for expr in exprs))
-            init_series, *series = align_series_full_broadcast(init_series, *series)
+            init_series, *series = align(init_series, *series)
             native_series = reduce(
                 pc.min_element_wise, [s.native for s in series], init_series.native
             )
@@ -175,8 +177,9 @@ class ArrowNamespace(EagerNamespace[ArrowDataFrame, ArrowSeries, ArrowExpr, pa.T
 
     def max_horizontal(self, *exprs: ArrowExpr) -> ArrowExpr:
         def func(df: ArrowDataFrame) -> list[ArrowSeries]:
+            align = self._series._align_full_broadcast
             init_series, *series = list(chain.from_iterable(expr(df) for expr in exprs))
-            init_series, *series = align_series_full_broadcast(init_series, *series)
+            init_series, *series = align(init_series, *series)
             native_series = reduce(
                 pc.max_element_wise, [s.native for s in series], init_series.native
             )
@@ -232,7 +235,8 @@ class ArrowNamespace(EagerNamespace[ArrowDataFrame, ArrowSeries, ArrowExpr, pa.T
         self, *exprs: ArrowExpr, separator: str, ignore_nulls: bool
     ) -> ArrowExpr:
         def func(df: ArrowDataFrame) -> list[ArrowSeries]:
-            compliant_series_list = align_series_full_broadcast(
+            align = self._series._align_full_broadcast
+            compliant_series_list = align(
                 *(chain.from_iterable(expr(df) for expr in exprs))
             )
             name = compliant_series_list[0].name
@@ -263,23 +267,20 @@ class ArrowNamespace(EagerNamespace[ArrowDataFrame, ArrowSeries, ArrowExpr, pa.T
         )
 
 
-class ArrowWhen(EagerWhen[ArrowDataFrame, ArrowSeries, ArrowExpr]):
+class ArrowWhen(EagerWhen[ArrowDataFrame, ArrowSeries, ArrowExpr, "ChunkedArrayAny"]):
     @property
     def _then(self) -> type[ArrowThen]:
         return ArrowThen
 
     def _if_then_else(
-        self, when: ArrowSeries, then: ArrowSeries, otherwise: ArrowSeries | None, /
-    ) -> ArrowSeries:
-        if otherwise is None:
-            when, then = align_series_full_broadcast(when, then)
-            res_native = pc.if_else(
-                when.native, then.native, pa.nulls(len(when.native), then.native.type)
-            )
-        else:
-            when, then, otherwise = align_series_full_broadcast(when, then, otherwise)
-            res_native = pc.if_else(when.native, then.native, otherwise.native)
-        return then._with_native(res_native)
+        self,
+        when: ChunkedArrayAny,
+        then: ChunkedArrayAny,
+        otherwise: ArrayOrScalar | NonNestedLiteral,
+        /,
+    ) -> ChunkedArrayAny:
+        otherwise = pa.nulls(len(when), then.type) if otherwise is None else otherwise
+        return pc.if_else(when, then, otherwise)
 
 
 class ArrowThen(CompliantThen[ArrowDataFrame, ArrowSeries, ArrowExpr], ArrowExpr): ...
diff --git a/narwhals/_arrow/series.py b/narwhals/_arrow/series.py
index abb460d4..3aeef13d 100644
--- a/narwhals/_arrow/series.py
+++ b/narwhals/_arrow/series.py
@@ -32,7 +32,7 @@ from narwhals._utils import (
     validate_backend_version,
 )
 from narwhals.dependencies import is_numpy_array_1d
-from narwhals.exceptions import InvalidOperationError
+from narwhals.exceptions import InvalidOperationError, ShapeError
 
 if TYPE_CHECKING:
     from collections.abc import Iterable, Iterator, Mapping, Sequence
@@ -192,6 +192,25 @@ class ArrowSeries(EagerSeries["ChunkedArrayAny"]):
             data if is_numpy_array_1d(data) else [data], context=context
         )
 
+    @classmethod
+    def _align_full_broadcast(cls, *series: Self) -> Sequence[Self]:
+        lengths = [len(s) for s in series]
+        max_length = max(lengths)
+        fast_path = all(_len == max_length for _len in lengths)
+        if fast_path:
+            return series
+        reshaped = []
+        for s in series:
+            if s._broadcast:
+                compliant = s._with_native(pa.repeat(s.native[0], max_length))
+            elif (actual_len := len(s)) != max_length:
+                msg = f"Expected object of length {max_length}, got {actual_len}."
+                raise ShapeError(msg)
+            else:
+                compliant = s
+            reshaped.append(compliant)
+        return reshaped
+
     def __narwhals_namespace__(self) -> ArrowNamespace:
         from narwhals._arrow.namespace import ArrowNamespace
 
diff --git a/narwhals/_arrow/series_str.py b/narwhals/_arrow/series_str.py
index 64dcce8c..3731e078 100644
--- a/narwhals/_arrow/series_str.py
+++ b/narwhals/_arrow/series_str.py
@@ -3,12 +3,14 @@ from __future__ import annotations
 import string
 from typing import TYPE_CHECKING
 
+import pyarrow as pa
 import pyarrow.compute as pc
 
 from narwhals._arrow.utils import ArrowSeriesNamespace, lit, parse_datetime_format
 
 if TYPE_CHECKING:
     from narwhals._arrow.series import ArrowSeries
+    from narwhals._arrow.typing import Incomplete
 
 
 class ArrowSeriesStringNamespace(ArrowSeriesNamespace):
@@ -60,3 +62,36 @@ class ArrowSeriesStringNamespace(ArrowSeriesNamespace):
 
     def to_lowercase(self) -> ArrowSeries:
         return self.with_native(pc.utf8_lower(self.native))
+
+    def zfill(self, width: int) -> ArrowSeries:
+        binary_join: Incomplete = pc.binary_join_element_wise
+        native = self.native
+        hyphen, plus = lit("-"), lit("+")
+        first_char, remaining_chars = self.slice(0, 1).native, self.slice(1, None).native
+
+        # Conditions
+        less_than_width = pc.less(pc.utf8_length(native), lit(width))
+        starts_with_hyphen = pc.equal(first_char, hyphen)
+        starts_with_plus = pc.equal(first_char, plus)
+
+        conditions = pc.make_struct(
+            pc.and_(starts_with_hyphen, less_than_width),
+            pc.and_(starts_with_plus, less_than_width),
+            less_than_width,
+        )
+
+        # Cases
+        padded_remaining_chars = pc.utf8_lpad(remaining_chars, width - 1, padding="0")
+
+        result = pc.case_when(
+            conditions,
+            binary_join(
+                pa.repeat(hyphen, len(native)), padded_remaining_chars, ""
+            ),  # starts with hyphen and less than width
+            binary_join(
+                pa.repeat(plus, len(native)), padded_remaining_chars, ""
+            ),  # starts with plus and less than width
+            pc.utf8_lpad(native, width=width, padding="0"),  # less than width
+            native,
+        )
+        return self.with_native(result)
diff --git a/narwhals/_arrow/utils.py b/narwhals/_arrow/utils.py
index 40bd475d..75965895 100644
--- a/narwhals/_arrow/utils.py
+++ b/narwhals/_arrow/utils.py
@@ -8,10 +8,9 @@ import pyarrow.compute as pc
 
 from narwhals._compliant.series import _SeriesNamespace
 from narwhals._utils import isinstance_or_issubclass
-from narwhals.exceptions import ShapeError
 
 if TYPE_CHECKING:
-    from collections.abc import Iterable, Iterator, Mapping, Sequence
+    from collections.abc import Iterable, Iterator, Mapping
 
     from typing_extensions import TypeAlias, TypeIs
 
@@ -261,31 +260,6 @@ def extract_native(
     return lhs.native, rhs if isinstance(rhs, pa.Scalar) else lit(rhs)
 
 
-def align_series_full_broadcast(*series: ArrowSeries) -> Sequence[ArrowSeries]:
-    # Ensure all of `series` are of the same length.
-    lengths = [len(s) for s in series]
-    max_length = max(lengths)
-    fast_path = all(_len == max_length for _len in lengths)
-
-    if fast_path:
-        return series
-
-    reshaped = []
-    for s in series:
-        if s._broadcast:
-            value = s.native[0]
-            if s._backend_version < (13,) and hasattr(value, "as_py"):
-                value = value.as_py()
-            reshaped.append(s._with_native(pa.array([value] * max_length, type=s._type)))
-        else:
-            if (actual_len := len(s)) != max_length:
-                msg = f"Expected object of length {max_length}, got {actual_len}."
-                raise ShapeError(msg)
-            reshaped.append(s)
-
-    return reshaped
-
-
 def floordiv_compat(left: ArrayOrScalar, right: ArrayOrScalar, /) -> Any:
     # The following lines are adapted from pandas' pyarrow implementation.
     # Ref: https://github.com/pandas-dev/pandas/blob/262fcfbffcee5c3116e86a951d8b693f90411e68/pandas/core/arrays/arrow/array.py#L124-L154
diff --git a/narwhals/_compliant/any_namespace.py b/narwhals/_compliant/any_namespace.py
index 3365d25b..93fc5323 100644
--- a/narwhals/_compliant/any_namespace.py
+++ b/narwhals/_compliant/any_namespace.py
@@ -79,6 +79,7 @@ class StringNamespace(_StoresCompliant[CompliantT_co], Protocol[CompliantT_co]):
     def to_datetime(self, format: str | None) -> CompliantT_co: ...
     def to_lowercase(self) -> CompliantT_co: ...
     def to_uppercase(self) -> CompliantT_co: ...
+    def zfill(self, width: int) -> CompliantT_co: ...
 
 
 class StructNamespace(_StoresCompliant[CompliantT_co], Protocol[CompliantT_co]):
diff --git a/narwhals/_compliant/dataframe.py b/narwhals/_compliant/dataframe.py
index c75aaef3..0474269d 100644
--- a/narwhals/_compliant/dataframe.py
+++ b/narwhals/_compliant/dataframe.py
@@ -13,6 +13,7 @@ from narwhals._compliant.typing import (
     EagerSeriesT,
     NativeExprT,
     NativeFrameT,
+    NativeSeriesT,
 )
 from narwhals._translate import (
     ArrowConvertible,
@@ -391,11 +392,11 @@ class CompliantLazyFrame(
 class EagerDataFrame(
     CompliantDataFrame[EagerSeriesT, EagerExprT, NativeFrameT, "DataFrame[NativeFrameT]"],
     CompliantLazyFrame[EagerExprT, NativeFrameT, "DataFrame[NativeFrameT]"],
-    Protocol[EagerSeriesT, EagerExprT, NativeFrameT],
+    Protocol[EagerSeriesT, EagerExprT, NativeFrameT, NativeSeriesT],
 ):
     def __narwhals_namespace__(
         self,
-    ) -> EagerNamespace[Self, EagerSeriesT, EagerExprT, NativeFrameT]: ...
+    ) -> EagerNamespace[Self, EagerSeriesT, EagerExprT, NativeFrameT, NativeSeriesT]: ...
 
     def to_narwhals(self) -> DataFrame[NativeFrameT]:
         return self._version.dataframe(self, level="full")
@@ -439,10 +440,14 @@ class EagerDataFrame(
     ) -> list[str]:
         return list(columns or (f"column_{x}" for x in range(data.shape[1])))
 
-    def _gather(self, rows: SizedMultiIndexSelector[Any]) -> Self: ...
+    def _gather(self, rows: SizedMultiIndexSelector[NativeSeriesT]) -> Self: ...
     def _gather_slice(self, rows: _SliceIndex | range) -> Self: ...
-    def _select_multi_index(self, columns: SizedMultiIndexSelector[Any]) -> Self: ...
-    def _select_multi_name(self, columns: SizedMultiNameSelector[Any]) -> Self: ...
+    def _select_multi_index(
+        self, columns: SizedMultiIndexSelector[NativeSeriesT]
+    ) -> Self: ...
+    def _select_multi_name(
+        self, columns: SizedMultiNameSelector[NativeSeriesT]
+    ) -> Self: ...
     def _select_slice_index(self, columns: _SliceIndex | range) -> Self: ...
     def _select_slice_name(self, columns: _SliceName) -> Self: ...
     def __getitem__(  # noqa: C901, PLR0912
diff --git a/narwhals/_compliant/expr.py b/narwhals/_compliant/expr.py
index 1ea3fcd8..021dc840 100644
--- a/narwhals/_compliant/expr.py
+++ b/narwhals/_compliant/expr.py
@@ -331,7 +331,7 @@ class EagerExpr(
 
     def __narwhals_namespace__(
         self,
-    ) -> EagerNamespace[EagerDataFrameT, EagerSeriesT, Self, Any]: ...
+    ) -> EagerNamespace[EagerDataFrameT, EagerSeriesT, Self, Any, Any]: ...
     def __narwhals_expr__(self) -> None: ...
 
     @classmethod
@@ -1127,6 +1127,9 @@ class EagerExprStringNamespace(
     def to_uppercase(self) -> EagerExprT:
         return self.compliant._reuse_series_namespace("str", "to_uppercase")
 
+    def zfill(self, width: int) -> EagerExprT:
+        return self.compliant._reuse_series_namespace("str", "zfill", width=width)
+
 
 class EagerExprStructNamespace(
     EagerExprNamespace[EagerExprT], StructNamespace[EagerExprT], Generic[EagerExprT]
diff --git a/narwhals/_compliant/namespace.py b/narwhals/_compliant/namespace.py
index 29067839..07d23199 100644
--- a/narwhals/_compliant/namespace.py
+++ b/narwhals/_compliant/namespace.py
@@ -14,6 +14,7 @@ from narwhals._compliant.typing import (
     LazyExprT,
     NativeFrameT,
     NativeFrameT_co,
+    NativeSeriesT,
 )
 from narwhals._utils import (
     exclude_column_names,
@@ -128,7 +129,7 @@ class LazyNamespace(
 
 class EagerNamespace(
     DepthTrackingNamespace[EagerDataFrameT, EagerExprT],
-    Protocol[EagerDataFrameT, EagerSeriesT, EagerExprT, NativeFrameT],
+    Protocol[EagerDataFrameT, EagerSeriesT, EagerExprT, NativeFrameT, NativeSeriesT],
 ):
     @property
     def _dataframe(self) -> type[EagerDataFrameT]: ...
@@ -136,9 +137,15 @@ class EagerNamespace(
     def _series(self) -> type[EagerSeriesT]: ...
     def when(
         self, predicate: EagerExprT
-    ) -> EagerWhen[EagerDataFrameT, EagerSeriesT, EagerExprT]: ...
+    ) -> EagerWhen[EagerDataFrameT, EagerSeriesT, EagerExprT, NativeSeriesT]: ...
 
-    def from_native(self, data: Any, /) -> EagerDataFrameT | EagerSeriesT:
+    @overload
+    def from_native(self, data: NativeFrameT, /) -> EagerDataFrameT: ...
+    @overload
+    def from_native(self, data: NativeSeriesT, /) -> EagerSeriesT: ...
+    def from_native(
+        self, data: NativeFrameT | NativeSeriesT | Any, /
+    ) -> EagerDataFrameT | EagerSeriesT:
         if self._dataframe._is_native(data):
             return self._dataframe.from_native(data, context=self)
         elif self._series._is_native(data):
diff --git a/narwhals/_compliant/series.py b/narwhals/_compliant/series.py
index 811b2339..0e908859 100644
--- a/narwhals/_compliant/series.py
+++ b/narwhals/_compliant/series.py
@@ -280,6 +280,24 @@ class EagerSeries(CompliantSeries[NativeSeriesT], Protocol[NativeSeriesT]):
     _version: Version
     _broadcast: bool
 
+    @classmethod
+    def _align_full_broadcast(cls, *series: Self) -> Sequence[Self]:
+        """Ensure all of `series` have the same length (and index if `pandas`).
+
+        Scalars get broadcasted to the full length of the longest Series.
+
+        This is useful when you need to construct a full Series anyway, such as:
+
+            DataFrame.select(...)
+
+        It should not be used in binary operations, such as:
+
+            nw.col("a") - nw.col("a").mean()
+
+        because then it's more efficient to extract the right-hand-side's single element as a scalar.
+        """
+        ...
+
     def _from_scalar(self, value: Any) -> Self:
         return self.from_iterable([value], name=self.name, context=self)
 
@@ -296,7 +314,9 @@ class EagerSeries(CompliantSeries[NativeSeriesT], Protocol[NativeSeriesT]):
         """
         ...
 
-    def __narwhals_namespace__(self) -> EagerNamespace[Any, Self, Any, Any]: ...
+    def __narwhals_namespace__(
+        self,
+    ) -> EagerNamespace[Any, Self, Any, Any, NativeSeriesT]: ...
 
     def _to_expr(self) -> EagerExpr[Any, Any]:
         return self.__narwhals_namespace__()._expr._from_series(self)  # type: ignore[no-any-return]
diff --git a/narwhals/_compliant/typing.py b/narwhals/_compliant/typing.py
index b2994f48..ea6f6048 100644
--- a/narwhals/_compliant/typing.py
+++ b/narwhals/_compliant/typing.py
@@ -61,12 +61,10 @@ CompliantNamespaceAny: TypeAlias = "CompliantNamespace[Any, Any]"
 
 DepthTrackingExprAny: TypeAlias = "DepthTrackingExpr[Any, Any]"
 
-EagerDataFrameAny: TypeAlias = "EagerDataFrame[Any, Any, Any]"
+EagerDataFrameAny: TypeAlias = "EagerDataFrame[Any, Any, Any, Any]"
 EagerSeriesAny: TypeAlias = "EagerSeries[Any]"
 EagerExprAny: TypeAlias = "EagerExpr[Any, Any]"
-EagerNamespaceAny: TypeAlias = (
-    "EagerNamespace[EagerDataFrameAny, EagerSeriesAny, EagerExprAny, NativeFrame]"
-)
+EagerNamespaceAny: TypeAlias = "EagerNamespace[EagerDataFrameAny, EagerSeriesAny, EagerExprAny, NativeFrame, NativeSeries]"
 
 LazyExprAny: TypeAlias = "LazyExpr[Any, Any]"
 
@@ -74,6 +72,9 @@ NativeExprT = TypeVar("NativeExprT", bound="NativeExpr")
 NativeExprT_co = TypeVar("NativeExprT_co", bound="NativeExpr", covariant=True)
 NativeSeriesT = TypeVar("NativeSeriesT", bound="NativeSeries")
 NativeSeriesT_co = TypeVar("NativeSeriesT_co", bound="NativeSeries", covariant=True)
+NativeSeriesT_contra = TypeVar(
+    "NativeSeriesT_contra", bound="NativeSeries", contravariant=True
+)
 NativeFrameT = TypeVar("NativeFrameT", bound="NativeFrame")
 NativeFrameT_co = TypeVar("NativeFrameT_co", bound="NativeFrame", covariant=True)
 NativeFrameT_contra = TypeVar(
@@ -127,7 +128,7 @@ EagerSeriesT = TypeVar("EagerSeriesT", bound=EagerSeriesAny)
 EagerSeriesT_co = TypeVar("EagerSeriesT_co", bound=EagerSeriesAny, covariant=True)
 
 # NOTE: `pyright` gives false (8) positives if this uses `EagerDataFrameAny`?
-EagerDataFrameT = TypeVar("EagerDataFrameT", bound="EagerDataFrame[Any, Any, Any]")
+EagerDataFrameT = TypeVar("EagerDataFrameT", bound="EagerDataFrame[Any, Any, Any, Any]")
 
 LazyExprT = TypeVar("LazyExprT", bound=LazyExprAny)
 LazyExprT_contra = TypeVar("LazyExprT_contra", bound=LazyExprAny, contravariant=True)
diff --git a/narwhals/_compliant/when_then.py b/narwhals/_compliant/when_then.py
index 5718dd98..919f4866 100644
--- a/narwhals/_compliant/when_then.py
+++ b/narwhals/_compliant/when_then.py
@@ -13,6 +13,7 @@ from narwhals._compliant.typing import (
     EagerSeriesT,
     LazyExprAny,
     NativeExprT,
+    NativeSeriesT,
     WindowFunction,
 )
 from narwhals._typing_compat import Protocol38
@@ -148,16 +149,21 @@ class LazyThen(
 
 class EagerWhen(
     CompliantWhen[EagerDataFrameT, EagerSeriesT, EagerExprT],
-    Protocol38[EagerDataFrameT, EagerSeriesT, EagerExprT],
+    Protocol38[EagerDataFrameT, EagerSeriesT, EagerExprT, NativeSeriesT],
 ):
     def _if_then_else(
-        self, when: EagerSeriesT, then: EagerSeriesT, otherwise: EagerSeriesT | None, /
-    ) -> EagerSeriesT: ...
+        self,
+        when: NativeSeriesT,
+        then: NativeSeriesT,
+        otherwise: NativeSeriesT | NonNestedLiteral | Scalar,
+        /,
+    ) -> NativeSeriesT: ...
 
     def __call__(self, df: EagerDataFrameT, /) -> Sequence[EagerSeriesT]:
         is_expr = self._condition._is_expr
         when: EagerSeriesT = self._condition(df)[0]
         then: EagerSeriesT
+        align = when._align_full_broadcast
 
         if is_expr(self._then_value):
             then = self._then_value(df)[0]
@@ -167,12 +173,12 @@ class EagerWhen(
 
         if is_expr(self._otherwise_value):
             otherwise = self._otherwise_value(df)[0]
-        elif self._otherwise_value is not None:
-            otherwise = when._from_scalar(self._otherwise_value)
-            otherwise._broadcast = True
+            when, then, otherwise = align(when, then, otherwise)
+            result = self._if_then_else(when.native, then.native, otherwise.native)
         else:
-            otherwise = self._otherwise_value
-        return [self._if_then_else(when, then, otherwise)]
+            when, then = align(when, then)
+            result = self._if_then_else(when.native, then.native, self._otherwise_value)
+        return [then._with_native(result)]
 
 
 class LazyWhen(
diff --git a/narwhals/_dask/expr_str.py b/narwhals/_dask/expr_str.py
index b770b53d..938b0844 100644
--- a/narwhals/_dask/expr_str.py
+++ b/narwhals/_dask/expr_str.py
@@ -96,3 +96,8 @@ class DaskExprStringNamespace:
         return self._compliant_expr._with_callable(
             lambda expr: expr.str.lower(), "to_lowercase"
         )
+
+    def zfill(self, width: int) -> DaskExpr:
+        return self._compliant_expr._with_callable(
+            lambda expr, width: expr.str.zfill(width), "zfill", width=width
+        )
diff --git a/narwhals/_duckdb/expr_str.py b/narwhals/_duckdb/expr_str.py
index 6a9d8c2e..e3969eda 100644
--- a/narwhals/_duckdb/expr_str.py
+++ b/narwhals/_duckdb/expr_str.py
@@ -4,7 +4,7 @@ from typing import TYPE_CHECKING
 
 from duckdb import FunctionExpression
 
-from narwhals._duckdb.utils import lit
+from narwhals._duckdb.utils import lit, when
 from narwhals._utils import not_implemented
 
 if TYPE_CHECKING:
@@ -100,4 +100,31 @@ class DuckDBExprStringNamespace:
             lambda expr: FunctionExpression("strptime", expr, lit(format))
         )
 
+    def zfill(self, width: int) -> DuckDBExpr:
+        # DuckDB does not have a built-in zfill function, so we need to implement it manually
+        # using string manipulation functions.
+
+        def func(expr: Expression) -> Expression:
+            less_than_width = FunctionExpression("length", expr) < lit(width)
+            zero, hyphen, plus = lit("0"), lit("-"), lit("+")
+
+            starts_with_minus = FunctionExpression("starts_with", expr, hyphen)
+            starts_with_plus = FunctionExpression("starts_with", expr, plus)
+            substring = FunctionExpression("substr", expr, lit(2))
+            padded_substring = FunctionExpression("lpad", substring, lit(width - 1), zero)
+            return (
+                when(
+                    starts_with_minus & less_than_width,
+                    FunctionExpression("concat", hyphen, padded_substring),
+                )
+                .when(
+                    starts_with_plus & less_than_width,
+                    FunctionExpression("concat", plus, padded_substring),
+                )
+                .when(less_than_width, FunctionExpression("lpad", expr, lit(width), zero))
+                .otherwise(expr)
+            )
+
+        return self._compliant_expr._with_callable(func)
+
     replace = not_implemented()
diff --git a/narwhals/_ibis/expr_str.py b/narwhals/_ibis/expr_str.py
index 1c0d6e59..6602995e 100644
--- a/narwhals/_ibis/expr_str.py
+++ b/narwhals/_ibis/expr_str.py
@@ -1,14 +1,15 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any, Callable
+from typing import TYPE_CHECKING, Any, Callable, cast
 
+import ibis
+import ibis.expr.types as ir
 from ibis.expr.datatypes import Timestamp
 
+from narwhals._ibis.utils import lit
 from narwhals._utils import _is_naive_format, not_implemented
 
 if TYPE_CHECKING:
-    import ibis.expr.types as ir
-
     from narwhals._ibis.expr import IbisExpr
 
 
@@ -100,4 +101,23 @@ class IbisExprStringNamespace:
         fn = self._to_datetime_naive if _is_naive_format(format) else self._to_datetime
         return self._compliant_expr._with_callable(fn(format))
 
+    def zfill(self, width: int) -> IbisExpr:
+        def func(expr: ir.StringColumn) -> ir.Value:
+            length = expr.length()
+            less_than_width = length < lit(width)
+            zero, hyphen, plus = "0", "-", "+"
+            starts_with_minus = expr.startswith(hyphen)
+            starts_with_plus = expr.startswith(plus)
+            one = cast("ir.IntegerScalar", lit(1))
+            sub_length = cast("ir.IntegerValue", length - one)
+            substring = expr.substr(one, sub_length).lpad(width - 1, zero)
+            return ibis.cases(
+                (starts_with_minus & less_than_width, (substring.lpad(width, hyphen))),
+                (starts_with_plus & less_than_width, (substring.lpad(width, plus))),
+                (less_than_width, expr.lpad(width, zero)),
+                else_=expr,
+            )
+
+        return self._compliant_expr._with_callable(func)
+
     replace = not_implemented()
diff --git a/narwhals/_namespace.py b/narwhals/_namespace.py
index 2d52a167..996ebc78 100644
--- a/narwhals/_namespace.py
+++ b/narwhals/_namespace.py
@@ -101,27 +101,38 @@ if TYPE_CHECKING:
     class _NativeDask(Protocol):
         _partition_type: type[pd.DataFrame]
 
-    class _NativeCuDF(Protocol):
+    class _CuDFDataFrame(NativeFrame, Protocol):
         def to_pylibcudf(self, *args: Any, **kwds: Any) -> Any: ...
 
+    class _CuDFSeries(NativeSeries, Protocol):
+        def to_pylibcudf(self, *args: Any, **kwds: Any) -> Any: ...
+        def where(self, cond: Any, other: Any = ..., **kwds: Any) -> Any: ...
+
     class _NativeIbis(Protocol):
         def sql(self, *args: Any, **kwds: Any) -> Any: ...
         def __pyarrow_result__(self, *args: Any, **kwds: Any) -> Any: ...
         def __pandas_result__(self, *args: Any, **kwds: Any) -> Any: ...
         def __polars_result__(self, *args: Any, **kwds: Any) -> Any: ...
 
-    class _ModinDataFrame(Protocol):
+    class _ModinDataFrame(NativeFrame, Protocol):
         _pandas_class: type[pd.DataFrame]
 
-    class _ModinSeries(Protocol):
+    class _ModinSeries(NativeSeries, Protocol):
         _pandas_class: type[pd.Series[Any]]
 
+        def where(self, cond: Any, other: Any = ..., **kwds: Any) -> Any: ...
+
     _NativePolars: TypeAlias = "pl.DataFrame | pl.LazyFrame | pl.Series"
     _NativeArrow: TypeAlias = "pa.Table | pa.ChunkedArray[Any]"
     _NativeDuckDB: TypeAlias = "duckdb.DuckDBPyRelation"
     _NativePandas: TypeAlias = "pd.DataFrame | pd.Series[Any]"
     _NativeModin: TypeAlias = "_ModinDataFrame | _ModinSeries"
-    _NativePandasLike: TypeAlias = "_NativePandas | _NativeCuDF | _NativeModin"
+    _NativeCuDF: TypeAlias = "_CuDFDataFrame | _CuDFSeries"
+    _NativePandasLikeSeries: TypeAlias = "pd.Series[Any] | _CuDFSeries | _ModinSeries"
+    _NativePandasLikeDataFrame: TypeAlias = (
+        "pd.DataFrame | _CuDFDataFrame | _ModinDataFrame"
+    )
+    _NativePandasLike: TypeAlias = "_NativePandasLikeDataFrame |_NativePandasLikeSeries"
     _NativeSQLFrame: TypeAlias = "SQLFrameDataFrame"
     _NativePySpark: TypeAlias = "pyspark_sql.DataFrame"
     _NativePySparkConnect: TypeAlias = "PySparkConnectDataFrame"
@@ -275,8 +286,8 @@ class Namespace(Generic[CompliantNamespaceT_co]):
     @overload
     @classmethod
     def from_native_object(
-        cls, native: _NativePandasLike, /
-    ) -> Namespace[PandasLikeNamespace]: ...
+        cls, native: _NativePandas, /
+    ) -> Namespace[PandasLikeNamespace[pd.DataFrame, pd.Series[Any]]]: ...
 
     @overload
     @classmethod
@@ -302,6 +313,24 @@ class Namespace(Generic[CompliantNamespaceT_co]):
     @classmethod
     def from_native_object(cls, native: _NativeIbis, /) -> Namespace[IbisNamespace]: ...
 
+    @overload
+    @classmethod
+    def from_native_object(
+        cls, native: _NativeModin, /
+    ) -> Namespace[PandasLikeNamespace[_ModinDataFrame, _ModinSeries]]: ...
+
+    @overload
+    @classmethod
+    def from_native_object(
+        cls, native: _NativeCuDF, /
+    ) -> Namespace[PandasLikeNamespace[_CuDFDataFrame, _CuDFSeries]]: ...
+
+    @overload
+    @classmethod
+    def from_native_object(
+        cls, native: _NativePandasLike, /
+    ) -> Namespace[PandasLikeNamespace[Any, Any]]: ...
+
     @overload
     @classmethod
     def from_native_object(
diff --git a/narwhals/_pandas_like/dataframe.py b/narwhals/_pandas_like/dataframe.py
index e57c1150..881395c8 100644
--- a/narwhals/_pandas_like/dataframe.py
+++ b/narwhals/_pandas_like/dataframe.py
@@ -10,7 +10,6 @@ from narwhals._compliant import EagerDataFrame
 from narwhals._pandas_like.series import PANDAS_TO_NUMPY_DTYPE_MISSING, PandasLikeSeries
 from narwhals._pandas_like.utils import (
     align_and_extract_native,
-    align_series_full_broadcast,
     check_column_names_are_unique,
     get_dtype_backend,
     native_to_narwhals_dtype,
@@ -93,7 +92,9 @@ CLASSICAL_NUMPY_DTYPES: frozenset[np.dtype[Any]] = frozenset(
 )
 
 
-class PandasLikeDataFrame(EagerDataFrame["PandasLikeSeries", "PandasLikeExpr", "Any"]):
+class PandasLikeDataFrame(
+    EagerDataFrame["PandasLikeSeries", "PandasLikeExpr", "Any", "pd.Series[Any]"]
+):
     def __init__(
         self,
         native_dataframe: Any,
@@ -394,7 +395,7 @@ class PandasLikeDataFrame(EagerDataFrame["PandasLikeSeries", "PandasLikeExpr", "
         if not new_series:
             # return empty dataframe, like Polars does
             return self._with_native(self.native.__class__(), validate_column_names=False)
-        new_series = align_series_full_broadcast(*new_series)
+        new_series = new_series[0]._align_full_broadcast(*new_series)
         namespace = self.__narwhals_namespace__()
         df = namespace._concat_horizontal([s.native for s in new_series])
         # `concat` creates a new object, so fine to modify `.columns.name` inplace.
diff --git a/narwhals/_pandas_like/namespace.py b/narwhals/_pandas_like/namespace.py
index 814e1f95..c4f2b9e6 100644
--- a/narwhals/_pandas_like/namespace.py
+++ b/narwhals/_pandas_like/namespace.py
@@ -3,9 +3,8 @@ from __future__ import annotations
 import operator
 import warnings
 from functools import reduce
-from typing import TYPE_CHECKING, Literal
-
-import pandas as pd
+from itertools import chain
+from typing import TYPE_CHECKING, Any, Literal, Protocol, overload
 
 from narwhals._compliant import CompliantThen, EagerNamespace, EagerWhen
 from narwhals._expression_parsing import (
@@ -16,21 +15,37 @@ from narwhals._pandas_like.dataframe import PandasLikeDataFrame
 from narwhals._pandas_like.expr import PandasLikeExpr
 from narwhals._pandas_like.selectors import PandasSelectorNamespace
 from narwhals._pandas_like.series import PandasLikeSeries
-from narwhals._pandas_like.utils import align_series_full_broadcast
+from narwhals._pandas_like.typing import NativeDataFrameT, NativeSeriesT
 
 if TYPE_CHECKING:
-    from collections.abc import Sequence
+    from collections.abc import Iterable, Sequence
+
+    from typing_extensions import TypeAlias
 
-    from narwhals._pandas_like.typing import NDFrameT
     from narwhals._utils import Implementation, Version
     from narwhals.typing import IntoDType, NonNestedLiteral
 
-VERTICAL: Literal[0] = 0
-HORIZONTAL: Literal[1] = 1
+
+Incomplete: TypeAlias = Any
+"""Escape hatch, but leaving a trace that this isn't ideal."""
+
+
+_Vertical: TypeAlias = Literal[0]
+_Horizontal: TypeAlias = Literal[1]
+Axis: TypeAlias = Literal[_Vertical, _Horizontal]
+
+VERTICAL: _Vertical = 0
+HORIZONTAL: _Horizontal = 1
 
 
 class PandasLikeNamespace(
-    EagerNamespace[PandasLikeDataFrame, PandasLikeSeries, PandasLikeExpr, pd.DataFrame]
+    EagerNamespace[
+        PandasLikeDataFrame,
+        PandasLikeSeries,
+        PandasLikeExpr,
+        NativeDataFrameT,
+        NativeSeriesT,
+    ]
 ):
     @property
     def _dataframe(self) -> type[PandasLikeDataFrame]:
@@ -101,8 +116,9 @@ class PandasLikeNamespace(
     # --- horizontal ---
     def sum_horizontal(self, *exprs: PandasLikeExpr) -> PandasLikeExpr:
         def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
-            series = [s for _expr in exprs for s in _expr(df)]
-            series = align_series_full_broadcast(*series)
+            align = self._series._align_full_broadcast
+            it = chain.from_iterable(expr(df) for expr in exprs)
+            series = align(*it)
             native_series = (s.fill_null(0, None, None) for s in series)
             return [reduce(operator.add, native_series)]
 
@@ -117,9 +133,8 @@ class PandasLikeNamespace(
 
     def all_horizontal(self, *exprs: PandasLikeExpr) -> PandasLikeExpr:
         def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
-            series = align_series_full_broadcast(
-                *(s for _expr in exprs for s in _expr(df))
-            )
+            align = self._series._align_full_broadcast
+            series = align(*(s for _expr in exprs for s in _expr(df)))
             return [reduce(operator.and_, series)]
 
         return self._expr._from_callable(
@@ -133,9 +148,8 @@ class PandasLikeNamespace(
 
     def any_horizontal(self, *exprs: PandasLikeExpr) -> PandasLikeExpr:
         def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
-            series = align_series_full_broadcast(
-                *(s for _expr in exprs for s in _expr(df))
-            )
+            align = self._series._align_full_broadcast
+            series = align(*(s for _expr in exprs for s in _expr(df)))
             return [reduce(operator.or_, series)]
 
         return self._expr._from_callable(
@@ -150,10 +164,11 @@ class PandasLikeNamespace(
     def mean_horizontal(self, *exprs: PandasLikeExpr) -> PandasLikeExpr:
         def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
             expr_results = [s for _expr in exprs for s in _expr(df)]
-            series = align_series_full_broadcast(
+            align = self._series._align_full_broadcast
+            series = align(
                 *(s.fill_null(0, strategy=None, limit=None) for s in expr_results)
             )
-            non_na = align_series_full_broadcast(*(1 - s.is_null() for s in expr_results))
+            non_na = align(*(1 - s.is_null() for s in expr_results))
             return [reduce(operator.add, series) / reduce(operator.add, non_na)]
 
         return self._expr._from_callable(
@@ -167,8 +182,9 @@ class PandasLikeNamespace(
 
     def min_horizontal(self, *exprs: PandasLikeExpr) -> PandasLikeExpr:
         def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
-            series = [s for _expr in exprs for s in _expr(df)]
-            series = align_series_full_broadcast(*series)
+            it = chain.from_iterable(expr(df) for expr in exprs)
+            align = self._series._align_full_broadcast
+            series = align(*it)
 
             return [
                 PandasLikeSeries(
@@ -192,14 +208,15 @@ class PandasLikeNamespace(
 
     def max_horizontal(self, *exprs: PandasLikeExpr) -> PandasLikeExpr:
         def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
-            series = [s for _expr in exprs for s in _expr(df)]
-            series = align_series_full_broadcast(*series)
+            it = chain.from_iterable(expr(df) for expr in exprs)
+            align = self._series._align_full_broadcast
+            series = align(*it)
 
             return [
                 PandasLikeSeries(
                     self.concat(
                         (s.to_frame() for s in series), how="horizontal"
-                    )._native_frame.max(axis=1),
+                    ).native.max(axis=1),
                     implementation=self._implementation,
                     backend_version=self._backend_version,
                     version=self._version,
@@ -216,21 +233,21 @@ class PandasLikeNamespace(
         )
 
     @property
-    def _concat(self):  # type: ignore[no-untyped-def] # noqa: ANN202
-        """Return the **native** equivalent of `pd.concat`."""
-        # NOTE: Leave un-annotated to allow `@overload` matching via inference.
-        if TYPE_CHECKING:
-            import pandas as pd
+    def _concat(self) -> _NativeConcat[NativeDataFrameT, NativeSeriesT]:
+        """Concatenate pandas objects along a particular axis.
 
-            return pd.concat
+        Return the **native** equivalent of `pd.concat`.
+        """
         return self._implementation.to_native_namespace().concat
 
-    def _concat_diagonal(self, dfs: Sequence[pd.DataFrame], /) -> pd.DataFrame:
+    def _concat_diagonal(self, dfs: Sequence[NativeDataFrameT], /) -> NativeDataFrameT:
         if self._implementation.is_pandas() and self._backend_version < (3,):
             return self._concat(dfs, axis=VERTICAL, copy=False)
         return self._concat(dfs, axis=VERTICAL)
 
-    def _concat_horizontal(self, dfs: Sequence[NDFrameT], /) -> pd.DataFrame:
+    def _concat_horizontal(
+        self, dfs: Sequence[NativeDataFrameT | NativeSeriesT], /
+    ) -> NativeDataFrameT:
         if self._implementation.is_cudf():
             with warnings.catch_warnings():
                 warnings.filterwarnings(
@@ -243,7 +260,7 @@ class PandasLikeNamespace(
             return self._concat(dfs, axis=HORIZONTAL, copy=False)
         return self._concat(dfs, axis=HORIZONTAL)
 
-    def _concat_vertical(self, dfs: Sequence[pd.DataFrame], /) -> pd.DataFrame:
+    def _concat_vertical(self, dfs: Sequence[NativeDataFrameT], /) -> NativeDataFrameT:
         cols_0 = dfs[0].columns
         for i, df in enumerate(dfs[1:], start=1):
             cols_current = df.columns
@@ -260,8 +277,8 @@ class PandasLikeNamespace(
             return self._concat(dfs, axis=VERTICAL, copy=False)
         return self._concat(dfs, axis=VERTICAL)
 
-    def when(self, predicate: PandasLikeExpr) -> PandasWhen:
-        return PandasWhen.from_expr(predicate, context=self)
+    def when(self, predicate: PandasLikeExpr) -> PandasWhen[NativeSeriesT]:
+        return PandasWhen[NativeSeriesT].from_expr(predicate, context=self)
 
     def concat_str(
         self, *exprs: PandasLikeExpr, separator: str, ignore_nulls: bool
@@ -270,8 +287,9 @@ class PandasLikeNamespace(
 
         def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
             expr_results = [s for _expr in exprs for s in _expr(df)]
-            series = align_series_full_broadcast(*(s.cast(string) for s in expr_results))
-            null_mask = align_series_full_broadcast(*(s.is_null() for s in expr_results))
+            align = self._series._align_full_broadcast
+            series = align(*(s.cast(string) for s in expr_results))
+            null_mask = align(*(s.is_null() for s in expr_results))
 
             if not ignore_nulls:
                 null_mask_result = reduce(operator.or_, null_mask)
@@ -306,25 +324,60 @@ class PandasLikeNamespace(
         )
 
 
-class PandasWhen(EagerWhen[PandasLikeDataFrame, PandasLikeSeries, PandasLikeExpr]):
+class _NativeConcat(Protocol[NativeDataFrameT, NativeSeriesT]):
+    @overload
+    def __call__(
+        self,
+        objs: Iterable[NativeDataFrameT],
+        *,
+        axis: _Vertical,
+        copy: bool | None = ...,
+    ) -> NativeDataFrameT: ...
+    @overload
+    def __call__(
+        self, objs: Iterable[NativeSeriesT], *, axis: _Vertical, copy: bool | None = ...
+    ) -> NativeSeriesT: ...
+    @overload
+    def __call__(
+        self,
+        objs: Iterable[NativeDataFrameT | NativeSeriesT],
+        *,
+        axis: _Horizontal,
+        copy: bool | None = ...,
+    ) -> NativeDataFrameT: ...
+    @overload
+    def __call__(
+        self,
+        objs: Iterable[NativeDataFrameT | NativeSeriesT],
+        *,
+        axis: Axis,
+        copy: bool | None = ...,
+    ) -> NativeDataFrameT | NativeSeriesT: ...
+
+    def __call__(
+        self,
+        objs: Iterable[NativeDataFrameT | NativeSeriesT],
+        *,
+        axis: Axis,
+        copy: bool | None = None,
+    ) -> NativeDataFrameT | NativeSeriesT: ...
+
+
+class PandasWhen(
+    EagerWhen[PandasLikeDataFrame, PandasLikeSeries, PandasLikeExpr, NativeSeriesT]
+):
     @property
     def _then(self) -> type[PandasThen]:
         return PandasThen
 
     def _if_then_else(
         self,
-        when: PandasLikeSeries,
-        then: PandasLikeSeries,
-        otherwise: PandasLikeSeries | None,
-        /,
-    ) -> PandasLikeSeries:
-        if otherwise is None:
-            when, then = align_series_full_broadcast(when, then)
-            res_native = then.native.where(when.native)
-        else:
-            when, then, otherwise = align_series_full_broadcast(when, then, otherwise)
-            res_native = then.native.where(when.native, otherwise.native)
-        return then._with_native(res_native)
+        when: NativeSeriesT,
+        then: NativeSeriesT,
+        otherwise: NativeSeriesT | NonNestedLiteral,
+    ) -> NativeSeriesT:
+        where: Incomplete = then.where
+        return where(when) if otherwise is None else where(when, otherwise)
 
 
 class PandasThen(
diff --git a/narwhals/_pandas_like/series.py b/narwhals/_pandas_like/series.py
index e4a63178..21a65b3c 100644
--- a/narwhals/_pandas_like/series.py
+++ b/narwhals/_pandas_like/series.py
@@ -219,6 +219,35 @@ class PandasLikeSeries(EagerSeries[Any]):
         native = implementation.to_native_namespace().Series(arr, name="")
         return cls.from_native(native, context=context)
 
+    @classmethod
+    def _align_full_broadcast(cls, *series: Self) -> Sequence[Self]:
+        Series = series[0].__native_namespace__().Series  # noqa: N806
+        lengths = [len(s) for s in series]
+        max_length = max(lengths)
+        idx = series[lengths.index(max_length)].native.index
+        reindexed = []
+        for s in series:
+            if s._broadcast:
+                native = Series(
+                    np.repeat(s.native.iloc[0], max_length),
+                    index=idx,
+                    name=s.name,
+                    dtype=s.native.dtype,
+                )
+                compliant = s._with_native(native)
+            elif s.native.index is not idx:
+                native = set_index(
+                    s.native,
+                    idx,
+                    implementation=s._implementation,
+                    backend_version=s._backend_version,
+                )
+                compliant = s._with_native(native)
+            else:
+                compliant = s
+            reindexed.append(compliant)
+        return reindexed
+
     @property
     def name(self) -> str:
         return self._name
diff --git a/narwhals/_pandas_like/series_str.py b/narwhals/_pandas_like/series_str.py
index c4bef09e..8a8d6081 100644
--- a/narwhals/_pandas_like/series_str.py
+++ b/narwhals/_pandas_like/series_str.py
@@ -77,3 +77,6 @@ class PandasLikeSeriesStringNamespace(
 
     def to_lowercase(self) -> PandasLikeSeries:
         return self.with_native(self.native.str.lower())
+
+    def zfill(self, width: int) -> PandasLikeSeries:
+        return self.with_native(self.native.str.zfill(width))
diff --git a/narwhals/_pandas_like/typing.py b/narwhals/_pandas_like/typing.py
index 6f7bcb26..1e2453f7 100644
--- a/narwhals/_pandas_like/typing.py
+++ b/narwhals/_pandas_like/typing.py
@@ -2,14 +2,24 @@ from __future__ import annotations  # pragma: no cover
 
 from typing import TYPE_CHECKING  # pragma: no cover
 
+from narwhals._typing_compat import TypeVar
+
 if TYPE_CHECKING:
-    from typing import Any, TypeVar
+    from typing import Any
 
     import pandas as pd
     from typing_extensions import TypeAlias
 
+    from narwhals._namespace import _NativePandasLikeDataFrame, _NativePandasLikeSeries
     from narwhals._pandas_like.expr import PandasLikeExpr
     from narwhals._pandas_like.series import PandasLikeSeries
 
     IntoPandasLikeExpr: TypeAlias = "PandasLikeExpr | PandasLikeSeries"
     NDFrameT = TypeVar("NDFrameT", "pd.DataFrame", "pd.Series[Any]")
+
+NativeSeriesT = TypeVar(
+    "NativeSeriesT", bound="_NativePandasLikeSeries", default="pd.Series[Any]"
+)
+NativeDataFrameT = TypeVar(
+    "NativeDataFrameT", bound="_NativePandasLikeDataFrame", default="pd.DataFrame"
+)
diff --git a/narwhals/_pandas_like/utils.py b/narwhals/_pandas_like/utils.py
index e02cf4c0..a2028bdd 100644
--- a/narwhals/_pandas_like/utils.py
+++ b/narwhals/_pandas_like/utils.py
@@ -501,48 +501,6 @@ def narwhals_to_native_dtype(  # noqa: C901, PLR0912, PLR0915
     raise AssertionError(msg)
 
 
-def align_series_full_broadcast(*series: PandasLikeSeries) -> list[PandasLikeSeries]:
-    # Ensure all of `series` have the same length and index. Scalars get broadcasted to
-    # the full length of the longest Series. This is useful when you need to construct a
-    # full Series anyway (e.g. `DataFrame.select`). It should not be used in binary operations,
-    # such as `nw.col('a') - nw.col('a').mean()`, because then it's more efficient to extract
-    # the right-hand-side's single element as a scalar.
-    native_namespace = series[0].__native_namespace__()
-
-    lengths = [len(s) for s in series]
-    max_length = max(lengths)
-
-    idx = series[lengths.index(max_length)].native.index
-    reindexed = []
-    for s in series:
-        if s._broadcast:
-            reindexed.append(
-                s._with_native(
-                    native_namespace.Series(
-                        [s.native.iloc[0]] * max_length,
-                        index=idx,
-                        name=s.name,
-                        dtype=s.native.dtype,
-                    )
-                )
-            )
-
-        elif s.native.index is not idx:
-            reindexed.append(
-                s._with_native(
-                    set_index(
-                        s.native,
-                        idx,
-                        implementation=s._implementation,
-                        backend_version=s._backend_version,
-                    )
-                )
-            )
-        else:
-            reindexed.append(s)
-    return reindexed
-
-
 def int_dtype_mapper(dtype: Any) -> str:
     if "pyarrow" in str(dtype):
         return "Int64[pyarrow]"
diff --git a/narwhals/_polars/expr.py b/narwhals/_polars/expr.py
index 426a01dc..261e2fff 100644
--- a/narwhals/_polars/expr.py
+++ b/narwhals/_polars/expr.py
@@ -339,6 +339,27 @@ class PolarsExprStringNamespace:
     def __init__(self, expr: PolarsExpr) -> None:
         self._compliant_expr = expr
 
+    def zfill(self, width: int) -> PolarsExpr:
+        native_expr = self._compliant_expr.native
+        native_result = native_expr.str.zfill(width)
+
+        if self._compliant_expr._backend_version <= (1, 30, 0):
+            length = native_expr.str.len_chars()
+            less_than_width = length < width
+            plus = "+"
+            starts_with_plus = native_expr.str.starts_with(plus)
+            native_result = (
+                pl.when(starts_with_plus & less_than_width)
+                .then(
+                    native_expr.str.slice(1, length)
+                    .str.zfill(width - 1)
+                    .str.pad_start(width, plus)
+                )
+                .otherwise(native_result)
+            )
+
+        return self._compliant_expr._with_native(native_result)
+
     def __getattr__(self, attr: str) -> Callable[[Any], PolarsExpr]:
         def func(*args: Any, **kwargs: Any) -> PolarsExpr:
             pos, kwds = extract_args_kwargs(args, kwargs)
diff --git a/narwhals/_polars/series.py b/narwhals/_polars/series.py
index 3c9cd8ba..adf1ab72 100644
--- a/narwhals/_polars/series.py
+++ b/narwhals/_polars/series.py
@@ -686,6 +686,12 @@ class PolarsSeriesStringNamespace:
     def __init__(self, series: PolarsSeries) -> None:
         self._compliant_series = series
 
+    def zfill(self, width: int) -> PolarsSeries:
+        series = self._compliant_series
+        name = series.name
+        ns = series.__narwhals_namespace__()
+        return series.to_frame().select(ns.col(name).str.zfill(width)).get_column(name)
+
     def __getattr__(self, attr: str) -> Any:
         def func(*args: Any, **kwargs: Any) -> Any:
             pos, kwds = extract_args_kwargs(args, kwargs)
diff --git a/narwhals/_spark_like/expr_str.py b/narwhals/_spark_like/expr_str.py
index 7c659525..4d64d653 100644
--- a/narwhals/_spark_like/expr_str.py
+++ b/narwhals/_spark_like/expr_str.py
@@ -113,3 +113,30 @@ class SparkLikeExprStringNamespace:
         return self._compliant_expr._with_callable(
             lambda expr: function(F.replace(expr, F.lit("T"), F.lit(" ")))
         )
+
+    def zfill(self, width: int) -> SparkLikeExpr:
+        def func(expr: Column) -> Column:
+            F = self._compliant_expr._F  # noqa: N806
+
+            length = F.length(expr)
+            less_than_width = length < width
+            hyphen, plus = F.lit("-"), F.lit("+")
+            starts_with_minus = F.startswith(expr, hyphen)
+            starts_with_plus = F.startswith(expr, plus)
+            sub_length = length - F.lit(1)
+            # NOTE: `len` annotated as `int`, but `Column.substr` accepts `int | Column`
+            substring = F.substring(expr, 2, sub_length)  # pyright: ignore[reportArgumentType]
+            padded_substring = F.lpad(substring, width - 1, "0")
+            return (
+                F.when(
+                    starts_with_minus & less_than_width,
+                    F.concat(hyphen, padded_substring),
+                )
+                .when(
+                    starts_with_plus & less_than_width, F.concat(plus, padded_substring)
+                )
+                .when(less_than_width, F.lpad(expr, width, "0"))
+                .otherwise(expr)
+            )
+
+        return self._compliant_expr._with_callable(func)
diff --git a/narwhals/expr_str.py b/narwhals/expr_str.py
index e598ff78..d6b5115e 100644
--- a/narwhals/expr_str.py
+++ b/narwhals/expr_str.py
@@ -447,3 +447,34 @@ class ExprStringNamespace(Generic[ExprT]):
         return self._expr._with_elementwise_op(
             lambda plx: self._expr._to_compliant_expr(plx).str.to_lowercase()
         )
+
+    def zfill(self, width: int) -> ExprT:
+        """Transform string to zero-padded variant.
+
+        Arguments:
+            width: The desired length of the string after padding. If the length of the
+                string is greater than `width`, no padding is applied.
+                If `width` is less than 0, no padding is applied.
+
+        Returns:
+            A new expression.
+
+        Examples:
+            >>> import pandas as pd
+            >>> import narwhals as nw
+            >>> df_native = pd.DataFrame({"digits": ["+1", "-1", "1", None]})
+            >>> df = nw.from_native(df_native)
+            >>> df.with_columns(zfill_col=nw.col("digits").str.zfill(3))
+            ┌──────────────────┐
+            |Narwhals DataFrame|
+            |------------------|
+            |  digits zfill_col|
+            |0     +1       +01|
+            |1     -1       -01|
+            |2      1       001|
+            |3   None      None|
+            └──────────────────┘
+        """
+        return self._expr._with_elementwise_op(
+            lambda plx: self._expr._to_compliant_expr(plx).str.zfill(width)
+        )
diff --git a/narwhals/series_str.py b/narwhals/series_str.py
index 82d50f5a..e2b149af 100644
--- a/narwhals/series_str.py
+++ b/narwhals/series_str.py
@@ -398,3 +398,28 @@ class SeriesStringNamespace(Generic[SeriesT]):
         return self._narwhals_series._with_compliant(
             self._narwhals_series._compliant_series.str.to_datetime(format=format)
         )
+
+    def zfill(self, width: int) -> SeriesT:
+        r"""Pad strings with zeros on the left.
+
+        Arguments:
+            width: The target width of the string. If the string is shorter than this width, it will be padded with zeros on the left.
+
+        Returns:
+            A new Series with strings padded with zeros on the left.
+
+        Examples:
+            >>> import pandas as pd
+            >>> import narwhals as nw
+            >>> s_native = pd.Series(["+1", "-23", "456", "123456"])
+            >>> s = nw.from_native(s_native, series_only=True)
+            >>> s.str.zfill(5).to_native()
+            0     +0001
+            1     -0023
+            2     00456
+            3    123456
+            dtype: object
+        """
+        return self._narwhals_series._with_compliant(
+            self._narwhals_series._compliant_series.str.zfill(width)
+        )


Test patch:
diff --git a/tests/expr_and_series/str/zfill_test.py b/tests/expr_and_series/str/zfill_test.py
new file mode 100644
index 00000000..8e4eccf2
--- /dev/null
+++ b/tests/expr_and_series/str/zfill_test.py
@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+import pytest
+
+import narwhals as nw
+from tests.utils import PANDAS_VERSION, Constructor, ConstructorEager, assert_equal_data
+
+data = {"a": ["-1", "+1", "1", "12", "123", "99999", "+9999", None]}
+expected = {"a": ["-01", "+01", "001", "012", "123", "99999", "+9999", None]}
+
+
+def uses_pyarrow_backend(constructor: Constructor | ConstructorEager) -> bool:
+    return constructor.__name__ in {
+        "pandas_pyarrow_constructor",
+        "modin_pyarrow_constructor",
+    }
+
+
+@pytest.mark.skipif(PANDAS_VERSION < (1, 5), reason="different zfill behavior")
+def test_str_zfill(request: pytest.FixtureRequest, constructor: Constructor) -> None:
+    if uses_pyarrow_backend(constructor):
+        reason = (
+            "pandas with pyarrow backend doesn't support str.zfill, see "
+            "https://github.com/pandas-dev/pandas/issues/61485"
+        )
+        request.applymarker(pytest.mark.xfail(reason=reason))
+
+    df = nw.from_native(constructor(data))
+    result = df.select(nw.col("a").str.zfill(3))
+    assert_equal_data(result, expected)
+
+
+@pytest.mark.skipif(PANDAS_VERSION < (1, 5), reason="different zfill behavior")
+def test_str_zfill_series(
+    request: pytest.FixtureRequest, constructor_eager: ConstructorEager
+) -> None:
+    if uses_pyarrow_backend(constructor_eager):
+        reason = (
+            "pandas with pyarrow backend doesn't support str.zfill, see "
+            "https://github.com/pandas-dev/pandas/issues/61485"
+        )
+        request.applymarker(pytest.mark.xfail(reason=reason))
+
+    df = nw.from_native(constructor_eager(data), eager_only=True)
+    result = df["a"].str.zfill(3)
+    assert_equal_data({"a": result}, expected)