Repository: narwhals-dev/narwhals (python)

Issue:
[Enh]: Add Expr.dt.offset_by()
### We would like to learn about your use case. For example, if this feature is needed  to adopt Narwhals in an open source project, could you please enter the link to it below?


_No response_

### Please describe the purpose of the new feature or describe the problem to solve.

It would be nice to have method to subtract and add periods from date and datetime columns. My use case for this is calculating year over year change by self joining a dataframe on month = previous year month.  However, there are many other use cases for this when working with timeseries data. 

### Suggest a solution if possible.

_No response_

### If you have tried alternatives, please describe them below.

_No response_

### Additional information that may help us understand your needs.

_No response_

Pull request:
feat: Add `Expr.dt.offset_by()`
<!--
# 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
- [x] ✨ Feature
- [ ] 🐛 Bug Fix
- [ ] 🔧 Optimization
- [ ] 📝 Documentation
- [ ] ✅ Test
- [ ] 🐳 Other

## Related issues

- Related issue #\<issue number\>
- Closes #2573

## Checklist

- [x] Code follows style guide (ruff)
- [x] Tests added
- [x] Documented the changes

## If you have comments or can explain your changes, please do so below

As a follow-up to this, it would be good to test nanoseconds.


Patch:
diff --git a/docs/api-reference/expr_dt.md b/docs/api-reference/expr_dt.md
index 31c8f604..af2c691e 100644
--- a/docs/api-reference/expr_dt.md
+++ b/docs/api-reference/expr_dt.md
@@ -13,6 +13,7 @@
         - minute
         - month
         - nanosecond
+        - offset_by
         - ordinal_day
         - replace_time_zone
         - second
diff --git a/docs/api-reference/series_dt.md b/docs/api-reference/series_dt.md
index 6592445a..e6a8f0a8 100644
--- a/docs/api-reference/series_dt.md
+++ b/docs/api-reference/series_dt.md
@@ -13,6 +13,7 @@
         - minute
         - month
         - nanosecond
+        - offset_by
         - ordinal_day
         - replace_time_zone
         - second
diff --git a/narwhals/_arrow/series_dt.py b/narwhals/_arrow/series_dt.py
index b37c9656..7c1c2bfb 100644
--- a/narwhals/_arrow/series_dt.py
+++ b/narwhals/_arrow/series_dt.py
@@ -18,7 +18,7 @@ from narwhals._constants import (
     US_PER_MINUTE,
     US_PER_SECOND,
 )
-from narwhals._duration import parse_interval_string
+from narwhals._duration import Interval
 
 if TYPE_CHECKING:
     from collections.abc import Mapping
@@ -202,7 +202,25 @@ class ArrowSeriesDateTimeNamespace(ArrowSeriesNamespace):
         return self.with_native(pc.multiply(self.native, factor).cast(pa.int64()))
 
     def truncate(self, every: str) -> ArrowSeries:
-        multiple, unit = parse_interval_string(every)
+        interval = Interval.parse(every)
         return self.with_native(
-            pc.floor_temporal(self.native, multiple=multiple, unit=UNITS_DICT[unit])
+            pc.floor_temporal(self.native, interval.multiple, UNITS_DICT[interval.unit])
         )
+
+    def offset_by(self, by: str) -> ArrowSeries:
+        interval = Interval.parse_no_constraints(by)
+        native = self.native
+        if interval.unit in {"y", "q", "mo"}:
+            msg = f"Offsetting by {interval.unit} is not yet supported for pyarrow."
+            raise NotImplementedError(msg)
+        if interval.unit == "d":
+            offset: pa.DurationScalar[Any] = lit(interval.to_timedelta())
+            if time_zone := native.type.tz:
+                native_naive = pc.local_timestamp(native)
+                result = pc.assume_timezone(pc.add(native_naive, offset), time_zone)
+                return self.with_native(result)
+        elif interval.unit == "ns":  # pragma: no cover
+            offset = lit(interval.multiple, pa.duration("ns"))  # type: ignore[assignment]
+        else:
+            offset = lit(interval.to_timedelta())
+        return self.with_native(pc.add(native, offset))
diff --git a/narwhals/_compliant/any_namespace.py b/narwhals/_compliant/any_namespace.py
index fcacfd90..dc41bafc 100644
--- a/narwhals/_compliant/any_namespace.py
+++ b/narwhals/_compliant/any_namespace.py
@@ -48,6 +48,7 @@ class DateTimeNamespace(_StoresCompliant[CompliantT_co], Protocol[CompliantT_co]
     def total_microseconds(self) -> CompliantT_co: ...
     def total_nanoseconds(self) -> CompliantT_co: ...
     def truncate(self, every: str) -> CompliantT_co: ...
+    def offset_by(self, by: str) -> CompliantT_co: ...
 
 
 class ListNamespace(_StoresCompliant[CompliantT_co], Protocol[CompliantT_co]):
diff --git a/narwhals/_compliant/expr.py b/narwhals/_compliant/expr.py
index 487d22f2..d28648e3 100644
--- a/narwhals/_compliant/expr.py
+++ b/narwhals/_compliant/expr.py
@@ -1112,6 +1112,9 @@ class EagerExprDateTimeNamespace(
     def truncate(self, every: str) -> EagerExprT:
         return self.compliant._reuse_series_namespace("dt", "truncate", every=every)
 
+    def offset_by(self, by: str) -> EagerExprT:
+        return self.compliant._reuse_series_namespace("dt", "offset_by", by=by)
+
 
 class EagerExprListNamespace(
     EagerExprNamespace[EagerExprT], ListNamespace[EagerExprT], Generic[EagerExprT]
diff --git a/narwhals/_dask/expr_dt.py b/narwhals/_dask/expr_dt.py
index fe0f7174..5f2f5f26 100644
--- a/narwhals/_dask/expr_dt.py
+++ b/narwhals/_dask/expr_dt.py
@@ -5,9 +5,9 @@ from typing import TYPE_CHECKING
 from narwhals._compliant import LazyExprNamespace
 from narwhals._compliant.any_namespace import DateTimeNamespace
 from narwhals._constants import MS_PER_SECOND, NS_PER_SECOND, US_PER_SECOND
-from narwhals._duration import parse_interval_string
+from narwhals._duration import Interval
 from narwhals._pandas_like.utils import (
-    UNIT_DICT,
+    ALIAS_DICT,
     calculate_timestamp_date,
     calculate_timestamp_datetime,
     native_to_narwhals_dtype,
@@ -154,9 +154,22 @@ class DaskExprDateTimeNamespace(
         )
 
     def truncate(self, every: str) -> DaskExpr:
-        multiple, unit = parse_interval_string(every)
+        interval = Interval.parse(every)
+        unit = interval.unit
         if unit in {"mo", "q", "y"}:
-            msg = f"Truncating to {unit} is not supported yet for dask."
+            msg = f"Truncating to {unit} is not yet supported for dask."
             raise NotImplementedError(msg)
-        freq = f"{multiple}{UNIT_DICT.get(unit, unit)}"
+        freq = f"{interval.multiple}{ALIAS_DICT.get(unit, unit)}"
         return self.compliant._with_callable(lambda expr: expr.dt.floor(freq), "truncate")
+
+    def offset_by(self, by: str) -> DaskExpr:
+        def func(s: dx.Series, by: str) -> dx.Series:
+            interval = Interval.parse_no_constraints(by)
+            unit = interval.unit
+            if unit in {"y", "q", "mo", "d", "ns"}:
+                msg = f"Offsetting by {unit} is not yet supported for dask."
+                raise NotImplementedError(msg)
+            offset = interval.to_timedelta()
+            return s.add(offset)
+
+        return self.compliant._with_callable(func, "offset_by", by=by)
diff --git a/narwhals/_duckdb/expr_dt.py b/narwhals/_duckdb/expr_dt.py
index 1fb8e1c3..110d2946 100644
--- a/narwhals/_duckdb/expr_dt.py
+++ b/narwhals/_duckdb/expr_dt.py
@@ -13,7 +13,7 @@ from narwhals._constants import (
     US_PER_SECOND,
 )
 from narwhals._duckdb.utils import UNITS_DICT, F, fetch_rel_time_zone, lit
-from narwhals._duration import parse_interval_string
+from narwhals._duration import Interval
 from narwhals._utils import not_implemented
 
 if TYPE_CHECKING:
@@ -99,7 +99,8 @@ class DuckDBExprDateTimeNamespace(
         )
 
     def truncate(self, every: str) -> DuckDBExpr:
-        multiple, unit = parse_interval_string(every)
+        interval = Interval.parse(every)
+        multiple, unit = interval.multiple, interval.unit
         if multiple != 1:
             # https://github.com/duckdb/duckdb/issues/17554
             msg = f"Only multiple 1 is currently supported for DuckDB.\nGot {multiple!s}."
@@ -114,6 +115,15 @@ class DuckDBExprDateTimeNamespace(
 
         return self.compliant._with_elementwise(_truncate)
 
+    def offset_by(self, by: str) -> DuckDBExpr:
+        interval = Interval.parse_no_constraints(by)
+        format = lit(f"{interval.multiple!s} {UNITS_DICT[interval.unit]}")
+
+        def _offset_by(expr: Expression) -> Expression:
+            return F("date_add", format, expr)
+
+        return self.compliant._with_callable(_offset_by)
+
     def _no_op_time_zone(self, time_zone: str) -> DuckDBExpr:
         def func(df: DuckDBLazyFrame) -> Sequence[Expression]:
             native_series_list = self.compliant(df)
diff --git a/narwhals/_duration.py b/narwhals/_duration.py
index 87b6d834..d1fe4a6d 100644
--- a/narwhals/_duration.py
+++ b/narwhals/_duration.py
@@ -2,13 +2,16 @@
 
 from __future__ import annotations
 
+import datetime as dt
 import re
 from typing import TYPE_CHECKING, Literal, cast, get_args
 
 if TYPE_CHECKING:
+    from collections.abc import Container, Mapping
+
     from typing_extensions import TypeAlias
 
-__all__ = ["IntervalUnit", "parse_interval_string"]
+__all__ = ["IntervalUnit"]
 
 IntervalUnit: TypeAlias = Literal["ns", "us", "ms", "s", "m", "h", "d", "mo", "q", "y"]
 """A Polars duration string interval unit.
@@ -24,23 +27,43 @@ IntervalUnit: TypeAlias = Literal["ns", "us", "ms", "s", "m", "h", "d", "mo", "q
 - 'q': quarter.
 - 'y': year.
 """
+TimedeltaKwd: TypeAlias = Literal[
+    "days", "hours", "minutes", "seconds", "milliseconds", "microseconds"
+]
 
 PATTERN_INTERVAL: re.Pattern[str] = re.compile(
     r"^(?P<multiple>\d+)(?P<unit>ns|us|ms|mo|m|s|h|d|q|y)\Z"
 )
 MONTH_MULTIPLES = frozenset([1, 2, 3, 4, 6, 12])
 QUARTER_MULTIPLES = frozenset([1, 2, 4])
+UNIT_TO_TIMEDELTA: Mapping[IntervalUnit, TimedeltaKwd] = {
+    "d": "days",
+    "h": "hours",
+    "m": "minutes",
+    "s": "seconds",
+    "ms": "milliseconds",
+    "us": "microseconds",
+}
+
 
+class Interval:
+    def __init__(self, multiple: int, unit: IntervalUnit, /) -> None:
+        self.multiple: int = multiple
+        self.unit: IntervalUnit = unit
 
-def parse_interval_string(every: str) -> tuple[int, IntervalUnit]:
-    """Parse a string like "1d", "2h", "3m" into a tuple of (number, unit).
+    def to_timedelta(
+        self, *, unsupported: Container[IntervalUnit] = frozenset(("ns", "mo", "q", "y"))
+    ) -> dt.timedelta:
+        if self.unit in unsupported:  # pragma: no cover
+            msg = f"Creating timedelta with {self.unit} unit is not supported."
+            raise NotImplementedError(msg)
+        kwd = UNIT_TO_TIMEDELTA[self.unit]
+        # error: Keywords must be strings (bad mypy)
+        return dt.timedelta(**{kwd: self.multiple})  # type: ignore[misc]
 
-    Returns:
-        A tuple of multiple and unit parsed from the interval string.
-    """
-    if match := PATTERN_INTERVAL.match(every):
-        multiple = int(match["multiple"])
-        unit = cast("IntervalUnit", match["unit"])
+    @classmethod
+    def parse(cls, every: str) -> Interval:
+        multiple, unit = cls._parse(every)
         if unit == "mo" and multiple not in MONTH_MULTIPLES:
             msg = f"Only the following multiples are supported for 'mo' unit: {MONTH_MULTIPLES}.\nGot: {multiple}."
             raise ValueError(msg)
@@ -52,9 +75,20 @@ def parse_interval_string(every: str) -> tuple[int, IntervalUnit]:
                 f"Only multiple 1 is currently supported for 'y' unit.\nGot: {multiple}."
             )
             raise ValueError(msg)
-        return multiple, unit
-    msg = (
-        f"Invalid `every` string: {every}. Expected string of kind <number><unit>, "
-        f"where 'unit' is one of: {get_args(IntervalUnit)}."
-    )
-    raise ValueError(msg)
+        return cls(multiple, unit)
+
+    @classmethod
+    def parse_no_constraints(cls, every: str) -> Interval:
+        return cls(*cls._parse(every))
+
+    @staticmethod
+    def _parse(every: str) -> tuple[int, IntervalUnit]:
+        if match := PATTERN_INTERVAL.match(every):
+            multiple = int(match["multiple"])
+            unit = cast("IntervalUnit", match["unit"])
+            return multiple, unit
+        msg = (
+            f"Invalid `every` string: {every}. Expected string of kind <number><unit>, "
+            f"where 'unit' is one of: {get_args(IntervalUnit)}."
+        )
+        raise ValueError(msg)
diff --git a/narwhals/_ibis/expr_dt.py b/narwhals/_ibis/expr_dt.py
index ef594337..35cc9602 100644
--- a/narwhals/_ibis/expr_dt.py
+++ b/narwhals/_ibis/expr_dt.py
@@ -4,7 +4,7 @@ from typing import TYPE_CHECKING, Any, Callable
 
 from narwhals._compliant import LazyExprNamespace
 from narwhals._compliant.any_namespace import DateTimeNamespace
-from narwhals._duration import parse_interval_string
+from narwhals._duration import Interval
 from narwhals._ibis.utils import UNITS_DICT_BUCKET, UNITS_DICT_TRUNCATE
 from narwhals._utils import not_implemented
 
@@ -68,7 +68,8 @@ class IbisExprDateTimeNamespace(
         return fn
 
     def truncate(self, every: str) -> IbisExpr:
-        multiple, unit = parse_interval_string(every)
+        interval = Interval.parse(every)
+        multiple, unit = interval.multiple, interval.unit
         if unit == "q":
             multiple, unit = 3 * multiple, "mo"
         if multiple != 1:
@@ -80,6 +81,15 @@ class IbisExprDateTimeNamespace(
             fn = self._truncate(UNITS_DICT_TRUNCATE[unit])
         return self.compliant._with_callable(fn)
 
+    def offset_by(self, every: str) -> IbisExpr:
+        interval = Interval.parse_no_constraints(every)
+        unit = interval.unit
+        if unit in {"y", "q", "mo", "d", "ns"}:
+            msg = f"Offsetting by {unit} is not yet supported for ibis."
+            raise NotImplementedError(msg)
+        offset = interval.to_timedelta()
+        return self.compliant._with_callable(lambda expr: expr.add(offset))
+
     def replace_time_zone(self, time_zone: str | None) -> IbisExpr:
         if time_zone is None:
             return self.compliant._with_callable(lambda expr: expr.cast("timestamp"))
diff --git a/narwhals/_pandas_like/series_dt.py b/narwhals/_pandas_like/series_dt.py
index 3666eaf0..3c7f7b9a 100644
--- a/narwhals/_pandas_like/series_dt.py
+++ b/narwhals/_pandas_like/series_dt.py
@@ -2,6 +2,8 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING, Any
 
+import pandas as pd
+
 from narwhals._compliant.any_namespace import DateTimeNamespace
 from narwhals._constants import (
     EPOCH_YEAR,
@@ -10,9 +12,10 @@ from narwhals._constants import (
     SECONDS_PER_DAY,
     US_PER_SECOND,
 )
-from narwhals._duration import parse_interval_string
+from narwhals._duration import Interval
 from narwhals._pandas_like.utils import (
-    UNIT_DICT,
+    ALIAS_DICT,
+    UNITS_DICT,
     PandasLikeSeriesNamespace,
     calculate_timestamp_date,
     calculate_timestamp_datetime,
@@ -203,13 +206,14 @@ class PandasLikeSeriesDateTimeNamespace(
         return self.with_native(result)
 
     def truncate(self, every: str) -> PandasLikeSeries:
-        multiple, unit = parse_interval_string(every)
+        interval = Interval.parse(every)
+        multiple, unit = interval.multiple, interval.unit
         native = self.native
         if self.implementation.is_cudf():
             if multiple != 1:
                 msg = f"Only multiple `1` is supported for cuDF, got: {multiple}."
                 raise NotImplementedError(msg)
-            return self.with_native(self.native.dt.floor(UNIT_DICT.get(unit, unit)))
+            return self.with_native(self.native.dt.floor(ALIAS_DICT.get(unit, unit)))
         dtype_backend = get_dtype_backend(native.dtype, self.compliant._implementation)
         if unit in {"mo", "q", "y"}:
             if self.implementation.is_cudf():
@@ -218,8 +222,6 @@ class PandasLikeSeriesDateTimeNamespace(
             if dtype_backend == "pyarrow":
                 import pyarrow.compute as pc  # ignore-banned-import
 
-                from narwhals._arrow.utils import UNITS_DICT
-
                 ca = native.array._pa_array
                 result_arr = pc.floor_temporal(ca, multiple, UNITS_DICT[unit])
             else:
@@ -230,7 +232,7 @@ class PandasLikeSeriesDateTimeNamespace(
                     np_unit = "M"
                 else:
                     np_unit = "Y"
-                arr = native.values
+                arr = native.values  # noqa: PD011
                 arr_dtype = arr.dtype
                 result_arr = arr.astype(f"datetime64[{multiple}{np_unit}]").astype(
                     arr_dtype
@@ -240,5 +242,48 @@ class PandasLikeSeriesDateTimeNamespace(
             )
             return self.with_native(result_native)
         return self.with_native(
-            self.native.dt.floor(f"{multiple}{UNIT_DICT.get(unit, unit)}")
+            self.native.dt.floor(f"{multiple}{ALIAS_DICT.get(unit, unit)}")
         )
+
+    def offset_by(self, by: str) -> PandasLikeSeries:
+        if self.implementation.is_cudf():
+            msg = "Not implemented for cuDF."
+            raise NotImplementedError(msg)
+        native = self.native
+        if self._is_pyarrow():
+            import pyarrow as pa  # ignore-banned-import
+
+            compliant = self.compliant
+            ca = pa.chunked_array([compliant.to_arrow()])  # type: ignore[arg-type]
+            result = (
+                compliant._version.namespace.from_backend("pyarrow")
+                .compliant.from_native(ca)
+                .dt.offset_by(by)
+                .native
+            )
+            result_pd = native.__class__(
+                result, dtype=native.dtype, index=native.index, name=native.name
+            )
+        else:
+            interval = Interval.parse_no_constraints(by)
+            multiple, unit = interval.multiple, interval.unit
+            if unit == "q":
+                multiple *= 3
+                unit = "mo"
+            offset: pd.DateOffset | pd.Timedelta
+            if unit == "y":
+                offset = pd.DateOffset(years=multiple)
+            elif unit == "mo":
+                offset = pd.DateOffset(months=multiple)
+            else:
+                offset = pd.Timedelta(multiple, unit=UNITS_DICT[unit])  # type: ignore[arg-type]
+            if unit == "d":
+                original_timezone = native.dt.tz
+                native_without_timezone = native.dt.tz_localize(None)
+                result_pd = native_without_timezone + offset
+                if original_timezone is not None:
+                    result_pd = result_pd.dt.tz_localize(original_timezone)
+            else:
+                result_pd = native + offset
+
+        return self.with_native(result_pd)
diff --git a/narwhals/_pandas_like/utils.py b/narwhals/_pandas_like/utils.py
index ef345b6f..65c15e72 100644
--- a/narwhals/_pandas_like/utils.py
+++ b/narwhals/_pandas_like/utils.py
@@ -25,12 +25,14 @@ from narwhals._utils import (
 from narwhals.exceptions import ShapeError
 
 if TYPE_CHECKING:
+    from collections.abc import Mapping
     from types import ModuleType
 
     from pandas._typing import Dtype as PandasDtype
     from pandas.core.dtypes.dtypes import BaseMaskedDtype
-    from typing_extensions import TypeIs
+    from typing_extensions import TypeAlias, TypeIs
 
+    from narwhals._duration import IntervalUnit
     from narwhals._pandas_like.expr import PandasLikeExpr
     from narwhals._pandas_like.series import PandasLikeSeries
     from narwhals._pandas_like.typing import (
@@ -95,7 +97,32 @@ PA_DURATION_RGX = r"""^
 $"""
 PATTERN_PA_DURATION = re.compile(PA_DURATION_RGX, re.VERBOSE)
 
-UNIT_DICT = {"d": "D", "m": "min"}
+NativeIntervalUnit: TypeAlias = Literal[
+    "year",
+    "quarter",
+    "month",
+    "week",
+    "day",
+    "hour",
+    "minute",
+    "second",
+    "millisecond",
+    "microsecond",
+    "nanosecond",
+]
+ALIAS_DICT = {"d": "D", "m": "min"}
+UNITS_DICT: Mapping[IntervalUnit, NativeIntervalUnit] = {
+    "y": "year",
+    "q": "quarter",
+    "mo": "month",
+    "d": "day",
+    "h": "hour",
+    "m": "minute",
+    "s": "second",
+    "ms": "millisecond",
+    "us": "microsecond",
+    "ns": "nanosecond",
+}
 
 
 def align_and_extract_native(
diff --git a/narwhals/_polars/expr.py b/narwhals/_polars/expr.py
index 5a2dc21f..1e0c9fbd 100644
--- a/narwhals/_polars/expr.py
+++ b/narwhals/_polars/expr.py
@@ -4,7 +4,7 @@ from typing import TYPE_CHECKING, Any, Callable, Literal
 
 import polars as pl
 
-from narwhals._duration import parse_interval_string
+from narwhals._duration import Interval
 from narwhals._polars.utils import (
     extract_args_kwargs,
     extract_native,
@@ -326,9 +326,14 @@ class PolarsExprNamespace:
 
 class PolarsExprDateTimeNamespace(PolarsExprNamespace):
     def truncate(self, every: str) -> PolarsExpr:
-        parse_interval_string(every)  # Ensure consistent error message is raised.
+        Interval.parse(every)  # Ensure consistent error message is raised.
         return self.compliant._with_native(self.native.dt.truncate(every))
 
+    def offset_by(self, by: str) -> PolarsExpr:
+        # Ensure consistent error message is raised.
+        Interval.parse_no_constraints(by)
+        return self.compliant._with_native(self.native.dt.offset_by(by))
+
     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/_spark_like/expr_dt.py b/narwhals/_spark_like/expr_dt.py
index aa8a975a..f03ed337 100644
--- a/narwhals/_spark_like/expr_dt.py
+++ b/narwhals/_spark_like/expr_dt.py
@@ -5,7 +5,7 @@ from typing import TYPE_CHECKING
 from narwhals._compliant import LazyExprNamespace
 from narwhals._compliant.any_namespace import DateTimeNamespace
 from narwhals._constants import US_PER_SECOND
-from narwhals._duration import parse_interval_string
+from narwhals._duration import Interval
 from narwhals._spark_like.utils import (
     UNITS_DICT,
     fetch_session_time_zone,
@@ -111,7 +111,8 @@ class SparkLikeExprDateTimeNamespace(
         return self.compliant._with_elementwise(self._weekday)
 
     def truncate(self, every: str) -> SparkLikeExpr:
-        multiple, unit = parse_interval_string(every)
+        interval = Interval.parse(every)
+        multiple, unit = interval.multiple, interval.unit
         if multiple != 1:
             msg = f"Only multiple 1 is currently supported for Spark-like.\nGot {multiple!s}."
             raise ValueError(msg)
@@ -125,6 +126,23 @@ class SparkLikeExprDateTimeNamespace(
 
         return self.compliant._with_elementwise(_truncate)
 
+    def offset_by(self, by: str) -> SparkLikeExpr:
+        interval = Interval.parse_no_constraints(by)
+        multiple, unit = interval.multiple, interval.unit
+        if unit == "ns":  # pragma: no cover
+            msg = "Offsetting by nanoseconds is not yet supported for Spark-like."
+            raise NotImplementedError(msg)
+
+        F = self.compliant._F  # noqa: N806
+
+        def _offset_by(expr: Column) -> Column:
+            # https://github.com/eakmanrq/sqlframe/issues/441
+            return F.timestamp_add(  # pyright: ignore[reportAttributeAccessIssue]
+                UNITS_DICT[unit], F.lit(multiple), expr
+            )
+
+        return self.compliant._with_callable(_offset_by)
+
     def _no_op_time_zone(self, time_zone: str) -> SparkLikeExpr:  # pragma: no cover
         def func(df: SparkLikeLazyFrame) -> Sequence[Column]:
             native_series_list = self.compliant(df)
diff --git a/narwhals/expr_dt.py b/narwhals/expr_dt.py
index 4619a7a5..4c87bfe0 100644
--- a/narwhals/expr_dt.py
+++ b/narwhals/expr_dt.py
@@ -782,3 +782,50 @@ class ExprDateTimeNamespace(Generic[ExprT]):
         return self._expr._with_elementwise(
             lambda plx: self._expr._to_compliant_expr(plx).dt.truncate(every)
         )
+
+    def offset_by(self, by: str) -> ExprT:
+        """Offset this date by a relative time offset.
+
+        Arguments:
+            by: The offset. Must be of form `<multiple><unit>`,
+                where `multiple` is a positive integer and `unit` is one of
+
+                - 'ns': nanosecond.
+                - 'us': microsecond.
+                - 'ms': millisecond.
+                - 's': second.
+                - 'm': minute.
+                - 'h': hour.
+                - 'd': day.
+                - 'mo': month.
+                - 'q': quarter.
+                - 'y': year.
+
+        Returns:
+            Expression of data type `Date` or `Datetime`.
+
+        Examples:
+            >>> from datetime import datetime
+            >>> import polars as pl
+            >>> import narwhals as nw
+            >>> df_native = pl.DataFrame({"datetime": [datetime(2021, 3, 1, 12, 34)]})
+            >>> df = nw.from_native(df_native)
+            >>> df.with_columns(
+            ...     nw.col("datetime").dt.offset_by("1h").alias("datetime_offset_by_1h")
+            ... )
+            ┌───────────────────────────────────────────────┐
+            |              Narwhals DataFrame               |
+            |-----------------------------------------------|
+            |shape: (1, 2)                                  |
+            |┌─────────────────────┬───────────────────────┐|
+            |│ datetime            ┆ datetime_offset_by_1h │|
+            |│ ---                 ┆ ---                   │|
+            |│ datetime[μs]        ┆ datetime[μs]          │|
+            |╞═════════════════════╪═══════════════════════╡|
+            |│ 2021-03-01 12:34:00 ┆ 2021-03-01 13:34:00   │|
+            |└─────────────────────┴───────────────────────┘|
+            └───────────────────────────────────────────────┘
+        """
+        return self._expr._with_elementwise(
+            lambda plx: self._expr._to_compliant_expr(plx).dt.offset_by(by)
+        )
diff --git a/narwhals/series_dt.py b/narwhals/series_dt.py
index 23f86f7d..a48c231a 100644
--- a/narwhals/series_dt.py
+++ b/narwhals/series_dt.py
@@ -681,3 +681,38 @@ class SeriesDateTimeNamespace(Generic[SeriesT]):
         return self._narwhals_series._with_compliant(
             self._narwhals_series._compliant_series.dt.truncate(every)
         )
+
+    def offset_by(self, by: str) -> SeriesT:
+        """Offset this date by a relative time offset.
+
+        Arguments:
+            by: The offset. Must be of form `<multiple><unit>`,
+                where `multiple` is a positive integer and `unit` is one of
+
+                - 'ns': nanosecond.
+                - 'us': microsecond.
+                - 'ms': millisecond.
+                - 's': second.
+                - 'm': minute.
+                - 'h': hour.
+                - 'd': day.
+                - 'mo': month.
+                - 'q': quarter.
+                - 'y': year.
+
+        Returns:
+            Series of data type `Date` or `Datetime`.
+
+        Examples:
+            >>> from datetime import datetime
+            >>> import pandas as pd
+            >>> import narwhals as nw
+            >>> s_native = pd.Series([datetime(2021, 3, 1, 12, 34)])
+            >>> s = nw.from_native(s_native, series_only=True)
+            >>> s.dt.offset_by("1h").to_native()
+            0   2021-03-01 13:34:00
+            dtype: datetime64[ns]
+        """
+        return self._narwhals_series._with_compliant(
+            self._narwhals_series._compliant_series.dt.offset_by(by)
+        )


Test patch:
diff --git a/tests/expr_and_series/dt/offset_by_test.py b/tests/expr_and_series/dt/offset_by_test.py
new file mode 100644
index 00000000..a660b3be
--- /dev/null
+++ b/tests/expr_and_series/dt/offset_by_test.py
@@ -0,0 +1,212 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+import pytest
+
+import narwhals as nw
+from tests.utils import (
+    PANDAS_VERSION,
+    PYARROW_VERSION,
+    Constructor,
+    ConstructorEager,
+    assert_equal_data,
+    is_windows,
+)
+
+data = {
+    "a": [datetime(2021, 3, 1, 12, 34, 56, 49012), datetime(2020, 1, 2, 2, 4, 14, 715123)]
+}
+
+data_tz = {"a": [datetime(2024, 1, 1, tzinfo=timezone.utc)]}
+
+data_dst = {"a": [datetime(2020, 10, 25, tzinfo=timezone.utc)]}
+
+
+@pytest.mark.parametrize(
+    ("by", "expected"),
+    [
+        (
+            "2us",
+            [
+                datetime(2021, 3, 1, 12, 34, 56, 49014),
+                datetime(2020, 1, 2, 2, 4, 14, 715125),
+            ],
+        ),
+        (
+            "2ms",
+            [
+                datetime(2021, 3, 1, 12, 34, 56, 51012),
+                datetime(2020, 1, 2, 2, 4, 14, 717123),
+            ],
+        ),
+        (
+            "10s",
+            [
+                datetime(2021, 3, 1, 12, 35, 6, 49012),
+                datetime(2020, 1, 2, 2, 4, 24, 715123),
+            ],
+        ),
+        (
+            "7m",
+            [
+                datetime(2021, 3, 1, 12, 41, 56, 49012),
+                datetime(2020, 1, 2, 2, 11, 14, 715123),
+            ],
+        ),
+        (
+            "7h",
+            [
+                datetime(2021, 3, 1, 19, 34, 56, 49012),
+                datetime(2020, 1, 2, 9, 4, 14, 715123),
+            ],
+        ),
+        (
+            "13d",
+            [
+                datetime(2021, 3, 14, 12, 34, 56, 49012),
+                datetime(2020, 1, 15, 2, 4, 14, 715123),
+            ],
+        ),
+        (
+            "3mo",
+            [
+                datetime(2021, 6, 1, 12, 34, 56, 49012),
+                datetime(2020, 4, 2, 2, 4, 14, 715123),
+            ],
+        ),
+        (
+            "2q",
+            [
+                datetime(2021, 9, 1, 12, 34, 56, 49012),
+                datetime(2020, 7, 2, 2, 4, 14, 715123),
+            ],
+        ),
+        (
+            "3y",
+            [
+                datetime(2024, 3, 1, 12, 34, 56, 49012),
+                datetime(2023, 1, 2, 2, 4, 14, 715123),
+            ],
+        ),
+    ],
+)
+def test_offset_by(
+    request: pytest.FixtureRequest,
+    constructor: Constructor,
+    by: str,
+    expected: list[datetime],
+) -> None:
+    df = nw.from_native(constructor(data))
+    if df.implementation.is_pyspark_connect():
+        # missing feature upstream
+        request.applymarker(pytest.mark.xfail())
+    if by == "2q" and "sqlframe" in str(constructor):
+        # https://github.com/eakmanrq/sqlframe/issues/443
+        request.applymarker(pytest.mark.xfail())
+    if any(x in by for x in ("y", "q", "mo")) and any(
+        x in str(constructor) for x in ("dask", "pyarrow", "ibis")
+    ):
+        request.applymarker(pytest.mark.xfail())
+    if by.endswith("d") and any(x in str(constructor) for x in ("dask", "ibis")):
+        request.applymarker(pytest.mark.xfail())
+    result = df.select(nw.col("a").dt.offset_by(by))
+    assert_equal_data(result, {"a": expected})
+
+
+@pytest.mark.parametrize(
+    ("by", "expected"),
+    [
+        ("2d", ["2024-01-03T05:45+0545"]),
+        ("5mo", ["2024-06-01T05:45+0545"]),
+        ("7q", ["2025-10-01T05:45+0545"]),
+        ("5y", ["2029-01-01T05:45+0545"]),
+    ],
+)
+def test_offset_by_tz(
+    request: pytest.FixtureRequest,
+    constructor: Constructor,
+    by: str,
+    expected: list[datetime],
+) -> None:
+    if (
+        ("pyarrow" in str(constructor) and is_windows())
+        or ("pyarrow_table" in str(constructor) and is_windows())
+        or ("pandas_pyarrow" in str(constructor) and PANDAS_VERSION < (2, 1))
+        or ("modin_pyarrow" in str(constructor) and PANDAS_VERSION < (2, 1))
+        or ("pyarrow_table" in str(constructor) and PYARROW_VERSION < (12,))
+    ):
+        pytest.skip()
+    if any(x in str(constructor) for x in ("duckdb", "pyspark", "sqlframe", "ibis")):
+        # pyspark,duckdb don't support changing time zones.
+        # convert_time_zone is not supported for ibis.
+        request.applymarker(pytest.mark.xfail())
+    if any(x in by for x in ("y", "q", "mo")) and any(
+        x in str(constructor) for x in ("dask", "pyarrow", "ibis")
+    ):
+        request.applymarker(pytest.mark.xfail())
+    if by.endswith("d") and any(x in str(constructor) for x in ("dask",)):
+        request.applymarker(pytest.mark.xfail())
+    df = nw.from_native(constructor(data_tz))
+    df = df.select(nw.col("a").dt.convert_time_zone("Asia/Kathmandu"))
+    result = df.select(nw.col("a").dt.offset_by(by))
+    result_str = result.select(nw.col("a").dt.to_string("%Y-%m-%dT%H:%M%z"))
+    assert_equal_data(result_str, {"a": expected})
+
+
+@pytest.mark.parametrize(
+    ("by", "expected"),
+    [
+        ("2d", ["2020-10-27T02:00+0100"]),
+        ("5mo", ["2021-03-25T02:00+0100"]),
+        ("1q", ["2021-01-25T02:00+0100"]),
+    ],
+)
+def test_offset_by_dst(
+    request: pytest.FixtureRequest,
+    constructor: Constructor,
+    by: str,
+    expected: list[datetime],
+) -> None:
+    if (
+        ("pyarrow" in str(constructor) and is_windows())
+        or ("pyarrow_table" in str(constructor) and is_windows())
+        or ("pandas_pyarrow" in str(constructor) and PANDAS_VERSION < (2, 1))
+        or ("modin_pyarrow" in str(constructor) and PANDAS_VERSION < (2, 1))
+        or ("pyarrow_table" in str(constructor) and PYARROW_VERSION < (12,))
+    ):
+        pytest.skip()
+    if any(x in str(constructor) for x in ("duckdb", "sqlframe", "pyspark", "ibis")):
+        # pyspark,duckdb don't support changing time zones.
+        # convert_time_zone is not supported for ibis.
+        request.applymarker(pytest.mark.xfail())
+    if any(x in by for x in ("y", "q", "mo")) and any(
+        x in str(constructor) for x in ("dask", "pyarrow")
+    ):
+        request.applymarker(pytest.mark.xfail())
+    if by.endswith("d") and any(x in str(constructor) for x in ("dask",)):
+        request.applymarker(pytest.mark.xfail())
+    df = nw.from_native(constructor(data_dst))
+    df = df.with_columns(a=nw.col("a").dt.convert_time_zone("Europe/Amsterdam"))
+    result = df.select(nw.col("a").dt.offset_by(by))
+    result_str = result.select(nw.col("a").dt.to_string("%Y-%m-%dT%H:%M%z"))
+    assert_equal_data(result_str, {"a": expected})
+
+
+def test_offset_by_series(constructor_eager: ConstructorEager) -> None:
+    df = nw.from_native(constructor_eager(data), eager_only=True)
+    result = df.select(df["a"].dt.offset_by("1h"))
+    expected = {
+        "a": [
+            datetime(2021, 3, 1, 13, 34, 56, 49012),
+            datetime(2020, 1, 2, 3, 4, 14, 715123),
+        ]
+    }
+    assert_equal_data(result, expected)
+
+
+def test_offset_by_invalid_interval(constructor: Constructor) -> None:
+    df = nw.from_native(constructor(data))
+    msg = "Invalid `every` string"
+    with pytest.raises(ValueError, match=msg):
+        df.select(nw.col("a").dt.offset_by("1r"))