Repository: narwhals-dev/narwhals
Pull request #1708 (2025-01-02 14:16:00)

feat: add missing dunder methods in `SparkLikeExpr` and `SparkLikeNamespace.lit`
<!--
# 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

## 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


MarcoGorelli: thanks @EdAbati ! we recently updated the `returns_scalar`s in dask in https://github.com/narwhals-dev/narwhals/pull/1684/files, should we do the same here in pyspark?
EdAbati: Oh interesting ! I will make a separate PR that we can merge before this one
MarcoGorelli: we can also remove the `__r*__` definitions now 😎 
EdAbati: Hey @MarcoGorelli I had a look at adding `binary_operation_returns_scalar` in Spark too. It takes a bit more time than expected I may have to look into it this weekend or next week.

Would it be ok to have it as a follow up?

Patch:
diff --git a/.github/workflows/check_tpch_queries.yml b/.github/workflows/check_tpch_queries.yml
index 723fa6e8..ce7da6f8 100644
--- a/.github/workflows/check_tpch_queries.yml
+++ b/.github/workflows/check_tpch_queries.yml
@@ -25,7 +25,7 @@ jobs:
           cache-suffix: ${{ matrix.python-version }}
           cache-dependency-glob: "pyproject.toml"
       - name: local-install
-        run: uv pip install -e ".[dev, core, dask]" --system
+        run: uv pip install -U --pre -e ".[dev, core, dask]" --system
       - name: generate-data
         run: cd tpch && python generate_data.py
       - name: tpch-tests 
diff --git a/.github/workflows/extremes.yml b/.github/workflows/extremes.yml
index 47ebc85e..0e7e6a20 100644
--- a/.github/workflows/extremes.yml
+++ b/.github/workflows/extremes.yml
@@ -158,8 +158,6 @@ jobs:
         run: |
           uv pip uninstall pyarrow --system
           uv pip install --extra-index-url https://pypi.fury.io/arrow-nightlies/ --pre pyarrow --system
-      - name: show-deps
-        run: uv pip freeze
       - name: install numpy nightly
         run: |
           uv pip uninstall numpy --system
@@ -167,18 +165,22 @@ jobs:
       - name: install dask
         run: |
           uv pip uninstall dask dask-expr --system
-          python -m pip install git+https://github.com/dask/distributed git+https://github.com/dask/dask git+https://github.com/dask/dask-expr
+          python -m pip install git+https://github.com/dask/distributed git+https://github.com/dask/dask
+      - name: install duckdb
+        run: |
+          python -m pip install -U --pre duckdb
       - name: show-deps
         run: uv pip freeze
       - name: Assert nightlies dependencies
         run: |
           DEPS=$(uv pip freeze)
-          echo "$DEPS" | grep 'polars'
+          echo "$DEPS" | grep 'polars.*@'
           echo "$DEPS" | grep 'pandas.*dev'
           echo "$DEPS" | grep 'pyarrow.*dev'
-          echo "$DEPS" | grep 'numpy'
-          echo "$DEPS" | grep 'dask'
+          echo "$DEPS" | grep 'numpy.*dev'
+          echo "$DEPS" | grep 'dask.*@'
+          echo "$DEPS" | grep 'duckdb.*dev'
       - name: Run pytest
         run: |
           pytest tests --cov=narwhals --cov=tests --cov-fail-under=50 --runslow \
-          --constructors=pandas,pandas[nullable],pandas[pyarrow],pyarrow,polars[eager],polars[lazy],dask
+          --constructors=pandas,pandas[nullable],pandas[pyarrow],pyarrow,polars[eager],polars[lazy],dask,duckdb
diff --git a/narwhals/_dask/expr.py b/narwhals/_dask/expr.py
index 40e7eff9..373c2902 100644
--- a/narwhals/_dask/expr.py
+++ b/narwhals/_dask/expr.py
@@ -23,7 +23,11 @@ from narwhals.utils import generate_temporary_column_name
 from narwhals.utils import import_dtypes_module
 
 if TYPE_CHECKING:
-    import dask_expr
+    try:
+        import dask.dataframe.dask_expr as dx
+    except ModuleNotFoundError:
+        import dask_expr as dx
+
     from typing_extensions import Self
 
     from narwhals._dask.dataframe import DaskLazyFrame
@@ -32,12 +36,12 @@ if TYPE_CHECKING:
     from narwhals.utils import Version
 
 
-class DaskExpr(CompliantExpr["dask_expr.Series"]):
+class DaskExpr(CompliantExpr["dx.Series"]):
     _implementation: Implementation = Implementation.DASK
 
     def __init__(
         self,
-        call: Callable[[DaskLazyFrame], Sequence[dask_expr.Series]],
+        call: Callable[[DaskLazyFrame], Sequence[dx.Series]],
         *,
         depth: int,
         function_name: str,
@@ -60,7 +64,7 @@ class DaskExpr(CompliantExpr["dask_expr.Series"]):
         self._version = version
         self._kwargs = kwargs
 
-    def __call__(self, df: DaskLazyFrame) -> Sequence[dask_expr.Series]:
+    def __call__(self, df: DaskLazyFrame) -> Sequence[dx.Series]:
         return self._call(df)
 
     def __narwhals_expr__(self) -> None: ...
@@ -78,7 +82,7 @@ class DaskExpr(CompliantExpr["dask_expr.Series"]):
         backend_version: tuple[int, ...],
         version: Version,
     ) -> Self:
-        def func(df: DaskLazyFrame) -> list[dask_expr.Series]:
+        def func(df: DaskLazyFrame) -> list[dx.Series]:
             try:
                 return [df._native_frame[column_name] for column_name in column_names]
             except KeyError as e:
@@ -107,7 +111,7 @@ class DaskExpr(CompliantExpr["dask_expr.Series"]):
         backend_version: tuple[int, ...],
         version: Version,
     ) -> Self:
-        def func(df: DaskLazyFrame) -> list[dask_expr.Series]:
+        def func(df: DaskLazyFrame) -> list[dx.Series]:
             return [
                 df._native_frame.iloc[:, column_index] for column_index in column_indices
             ]
@@ -126,14 +130,14 @@ class DaskExpr(CompliantExpr["dask_expr.Series"]):
 
     def _from_call(
         self,
-        # First argument to `call` should be `dask_expr.Series`
-        call: Callable[..., dask_expr.Series],
+        # First argument to `call` should be `dx.Series`
+        call: Callable[..., dx.Series],
         expr_name: str,
         *,
         returns_scalar: bool,
         **kwargs: Any,
     ) -> Self:
-        def func(df: DaskLazyFrame) -> list[dask_expr.Series]:
+        def func(df: DaskLazyFrame) -> list[dx.Series]:
             results = []
             inputs = self._call(df)
             _kwargs = {key: maybe_evaluate(df, value) for key, value in kwargs.items()}
@@ -163,7 +167,7 @@ class DaskExpr(CompliantExpr["dask_expr.Series"]):
         )
 
     def alias(self, name: str) -> Self:
-        def func(df: DaskLazyFrame) -> list[dask_expr.Series]:
+        def func(df: DaskLazyFrame) -> list[dx.Series]:
             inputs = self._call(df)
             return [_input.rename(name) for _input in inputs]
 
@@ -312,7 +316,7 @@ class DaskExpr(CompliantExpr["dask_expr.Series"]):
     def median(self) -> Self:
         from narwhals.exceptions import InvalidOperationError
 
-        def func(s: dask_expr.Series) -> dask_expr.Series:
+        def func(s: dx.Series) -> dx.Series:
             dtype = native_to_narwhals_dtype(s, self._version, Implementation.DASK)
             if not dtype.is_numeric():
                 msg = "`median` operation not supported for non-numeric input type."
@@ -511,11 +515,11 @@ class DaskExpr(CompliantExpr["dask_expr.Series"]):
         limit: int | None = None,
     ) -> DaskExpr:
         def func(
-            _input: dask_expr.Series,
+            _input: dx.Series,
             value: Any | None,
             strategy: str | None,
             limit: int | None,
-        ) -> dask_expr.Series:
+        ) -> dx.Series:
             if value is not None:
                 res_ser = _input.fillna(value)
             else:
@@ -566,7 +570,7 @@ class DaskExpr(CompliantExpr["dask_expr.Series"]):
         )
 
     def is_nan(self: Self) -> Self:
-        def func(_input: dask_expr.Series) -> dask_expr.Series:
+        def func(_input: dx.Series) -> dx.Series:
             dtype = native_to_narwhals_dtype(_input, self._version, self._implementation)
             if dtype.is_numeric():
                 return _input != _input  # noqa: PLR0124
@@ -585,7 +589,7 @@ class DaskExpr(CompliantExpr["dask_expr.Series"]):
     ) -> Self:
         if interpolation == "linear":
 
-            def func(_input: dask_expr.Series, quantile: float) -> dask_expr.Series:
+            def func(_input: dx.Series, quantile: float) -> dx.Series:
                 if _input.npartitions > 1:
                     msg = "`Expr.quantile` is not supported for Dask backend with multiple partitions."
                     raise NotImplementedError(msg)
@@ -599,7 +603,7 @@ class DaskExpr(CompliantExpr["dask_expr.Series"]):
             raise NotImplementedError(msg)
 
     def is_first_distinct(self: Self) -> Self:
-        def func(_input: dask_expr.Series) -> dask_expr.Series:
+        def func(_input: dx.Series) -> dx.Series:
             _name = _input.name
             col_token = generate_temporary_column_name(n_bytes=8, columns=[_name])
             _input = add_row_index(
@@ -618,7 +622,7 @@ class DaskExpr(CompliantExpr["dask_expr.Series"]):
         )
 
     def is_last_distinct(self: Self) -> Self:
-        def func(_input: dask_expr.Series) -> dask_expr.Series:
+        def func(_input: dx.Series) -> dx.Series:
             _name = _input.name
             col_token = generate_temporary_column_name(n_bytes=8, columns=[_name])
             _input = add_row_index(
@@ -635,7 +639,7 @@ class DaskExpr(CompliantExpr["dask_expr.Series"]):
         )
 
     def is_duplicated(self: Self) -> Self:
-        def func(_input: dask_expr.Series) -> dask_expr.Series:
+        def func(_input: dx.Series) -> dx.Series:
             _name = _input.name
             return (
                 _input.to_frame()
@@ -647,7 +651,7 @@ class DaskExpr(CompliantExpr["dask_expr.Series"]):
         return self._from_call(func, "is_duplicated", returns_scalar=self._returns_scalar)
 
     def is_unique(self: Self) -> Self:
-        def func(_input: dask_expr.Series) -> dask_expr.Series:
+        def func(_input: dx.Series) -> dx.Series:
             _name = _input.name
             return (
                 _input.to_frame()
@@ -967,7 +971,7 @@ class DaskExprDateTimeNamespace:
         )
 
     def convert_time_zone(self, time_zone: str) -> DaskExpr:
-        def func(s: dask_expr.Series, time_zone: str) -> dask_expr.Series:
+        def func(s: dx.Series, time_zone: str) -> dx.Series:
             dtype = native_to_narwhals_dtype(
                 s, self._compliant_expr._version, Implementation.DASK
             )
@@ -984,9 +988,7 @@ class DaskExprDateTimeNamespace:
         )
 
     def timestamp(self, time_unit: Literal["ns", "us", "ms"] = "us") -> DaskExpr:
-        def func(
-            s: dask_expr.Series, time_unit: Literal["ns", "us", "ms"] = "us"
-        ) -> dask_expr.Series:
+        def func(s: dx.Series, time_unit: Literal["ns", "us", "ms"] = "us") -> dx.Series:
             dtype = native_to_narwhals_dtype(
                 s, self._compliant_expr._version, Implementation.DASK
             )
diff --git a/narwhals/_dask/group_by.py b/narwhals/_dask/group_by.py
index 243b21b7..60086efa 100644
--- a/narwhals/_dask/group_by.py
+++ b/narwhals/_dask/group_by.py
@@ -12,7 +12,12 @@ from narwhals.utils import remove_prefix
 
 if TYPE_CHECKING:
     import dask.dataframe as dd
-    import dask_expr
+
+    try:
+        import dask.dataframe.dask_expr as dx
+    except ModuleNotFoundError:
+        import dask_expr as dx
+
     import pandas as pd
 
     from narwhals._dask.dataframe import DaskLazyFrame
@@ -43,7 +48,10 @@ def var(
 ]:
     from functools import partial
 
-    import dask_expr as dx
+    try:
+        import dask.dataframe.dask_expr as dx
+    except ModuleNotFoundError:
+        import dask_expr as dx
 
     return partial(dx._groupby.GroupBy.var, ddof=ddof)
 
@@ -55,7 +63,10 @@ def std(
 ]:
     from functools import partial
 
-    import dask_expr as dx
+    try:
+        import dask.dataframe.dask_expr as dx
+    except ModuleNotFoundError:
+        import dask_expr as dx
 
     return partial(dx._groupby.GroupBy.std, ddof=ddof)
 
@@ -127,7 +138,7 @@ class DaskLazyGroupBy:
 def agg_dask(
     df: DaskLazyFrame,
     grouped: Any,
-    exprs: Sequence[CompliantExpr[dask_expr.Series]],
+    exprs: Sequence[CompliantExpr[dx.Series]],
     keys: list[str],
     from_dataframe: Callable[[Any], DaskLazyFrame],
 ) -> DaskLazyFrame:
diff --git a/narwhals/_dask/namespace.py b/narwhals/_dask/namespace.py
index 9a16d7f1..d8b2b7a9 100644
--- a/narwhals/_dask/namespace.py
+++ b/narwhals/_dask/namespace.py
@@ -21,14 +21,17 @@ from narwhals._expression_parsing import reduce_output_names
 from narwhals.typing import CompliantNamespace
 
 if TYPE_CHECKING:
-    import dask_expr
+    try:
+        import dask.dataframe.dask_expr as dx
+    except ModuleNotFoundError:
+        import dask_expr as dx
 
     from narwhals._dask.typing import IntoDaskExpr
     from narwhals.dtypes import DType
     from narwhals.utils import Version
 
 
-class DaskNamespace(CompliantNamespace["dask_expr.Series"]):
+class DaskNamespace(CompliantNamespace["dx.Series"]):
     @property
     def selectors(self) -> DaskSelectorNamespace:
         return DaskSelectorNamespace(
@@ -40,7 +43,7 @@ class DaskNamespace(CompliantNamespace["dask_expr.Series"]):
         self._version = version
 
     def all(self) -> DaskExpr:
-        def func(df: DaskLazyFrame) -> list[dask_expr.Series]:
+        def func(df: DaskLazyFrame) -> list[dx.Series]:
             return [df._native_frame[column_name] for column_name in df.columns]
 
         return DaskExpr(
@@ -69,7 +72,7 @@ class DaskNamespace(CompliantNamespace["dask_expr.Series"]):
         import dask.dataframe as dd
         import pandas as pd
 
-        def func(df: DaskLazyFrame) -> list[dask_expr.Series]:
+        def func(df: DaskLazyFrame) -> list[dx.Series]:
             return [
                 dd.from_pandas(
                     pd.Series(
@@ -99,7 +102,7 @@ class DaskNamespace(CompliantNamespace["dask_expr.Series"]):
         import dask.dataframe as dd
         import pandas as pd
 
-        def func(df: DaskLazyFrame) -> list[dask_expr.Series]:
+        def func(df: DaskLazyFrame) -> list[dx.Series]:
             if not df.columns:
                 return [
                     dd.from_pandas(
@@ -125,7 +128,7 @@ class DaskNamespace(CompliantNamespace["dask_expr.Series"]):
     def all_horizontal(self, *exprs: IntoDaskExpr) -> DaskExpr:
         parsed_exprs = parse_into_exprs(*exprs, namespace=self)
 
-        def func(df: DaskLazyFrame) -> list[dask_expr.Series]:
+        def func(df: DaskLazyFrame) -> list[dx.Series]:
             series = [s for _expr in parsed_exprs for s in _expr(df)]
             return [reduce(lambda x, y: x & y, series).rename(series[0].name)]
 
@@ -144,7 +147,7 @@ class DaskNamespace(CompliantNamespace["dask_expr.Series"]):
     def any_horizontal(self, *exprs: IntoDaskExpr) -> DaskExpr:
         parsed_exprs = parse_into_exprs(*exprs, namespace=self)
 
-        def func(df: DaskLazyFrame) -> list[dask_expr.Series]:
+        def func(df: DaskLazyFrame) -> list[dx.Series]:
             series = [s for _expr in parsed_exprs for s in _expr(df)]
             return [reduce(lambda x, y: x | y, series).rename(series[0].name)]
 
@@ -163,7 +166,7 @@ class DaskNamespace(CompliantNamespace["dask_expr.Series"]):
     def sum_horizontal(self, *exprs: IntoDaskExpr) -> DaskExpr:
         parsed_exprs = parse_into_exprs(*exprs, namespace=self)
 
-        def func(df: DaskLazyFrame) -> list[dask_expr.Series]:
+        def func(df: DaskLazyFrame) -> list[dx.Series]:
             series = [s.fillna(0) for _expr in parsed_exprs for s in _expr(df)]
             return [reduce(lambda x, y: x + y, series).rename(series[0].name)]
 
@@ -239,7 +242,7 @@ class DaskNamespace(CompliantNamespace["dask_expr.Series"]):
     def mean_horizontal(self, *exprs: IntoDaskExpr) -> DaskExpr:
         parsed_exprs = parse_into_exprs(*exprs, namespace=self)
 
-        def func(df: DaskLazyFrame) -> list[dask_expr.Series]:
+        def func(df: DaskLazyFrame) -> list[dx.Series]:
             series = (s.fillna(0) for _expr in parsed_exprs for s in _expr(df))
             non_na = (1 - s.isna() for _expr in parsed_exprs for s in _expr(df))
             return [
@@ -266,7 +269,7 @@ class DaskNamespace(CompliantNamespace["dask_expr.Series"]):
 
         parsed_exprs = parse_into_exprs(*exprs, namespace=self)
 
-        def func(df: DaskLazyFrame) -> list[dask_expr.Series]:
+        def func(df: DaskLazyFrame) -> list[dx.Series]:
             series = [s for _expr in parsed_exprs for s in _expr(df)]
 
             return [dd.concat(series, axis=1).min(axis=1).rename(series[0].name)]
@@ -288,7 +291,7 @@ class DaskNamespace(CompliantNamespace["dask_expr.Series"]):
 
         parsed_exprs = parse_into_exprs(*exprs, namespace=self)
 
-        def func(df: DaskLazyFrame) -> list[dask_expr.Series]:
+        def func(df: DaskLazyFrame) -> list[dx.Series]:
             series = [s for _expr in parsed_exprs for s in _expr(df)]
 
             return [dd.concat(series, axis=1).max(axis=1).rename(series[0].name)]
@@ -327,7 +330,7 @@ class DaskNamespace(CompliantNamespace["dask_expr.Series"]):
             *parse_into_exprs(*more_exprs, namespace=self),
         ]
 
-        def func(df: DaskLazyFrame) -> list[dask_expr.Series]:
+        def func(df: DaskLazyFrame) -> list[dx.Series]:
             series = (s.astype(str) for _expr in parsed_exprs for s in _expr(df))
             null_mask = [s for _expr in parsed_exprs for s in _expr.is_null()(df)]
 
@@ -389,12 +392,12 @@ class DaskWhen:
         self._returns_scalar = returns_scalar
         self._version = version
 
-    def __call__(self, df: DaskLazyFrame) -> Sequence[dask_expr.Series]:
+    def __call__(self, df: DaskLazyFrame) -> Sequence[dx.Series]:
         from narwhals._expression_parsing import parse_into_expr
 
         plx = df.__narwhals_namespace__()
         condition = parse_into_expr(self._condition, namespace=plx)(df)[0]
-        condition = cast("dask_expr.Series", condition)
+        condition = cast("dx.Series", condition)
         try:
             value_series = parse_into_expr(self._then_value, namespace=plx)(df)[0]
         except TypeError:
@@ -402,7 +405,7 @@ class DaskWhen:
             _df = condition.to_frame("a")
             _df["tmp"] = self._then_value
             value_series = _df["tmp"]
-        value_series = cast("dask_expr.Series", value_series)
+        value_series = cast("dx.Series", value_series)
         validate_comparand(condition, value_series)
 
         if self._otherwise_value is None:
diff --git a/narwhals/_dask/selectors.py b/narwhals/_dask/selectors.py
index 2891d84f..703e2486 100644
--- a/narwhals/_dask/selectors.py
+++ b/narwhals/_dask/selectors.py
@@ -8,7 +8,10 @@ from narwhals._dask.expr import DaskExpr
 from narwhals.utils import import_dtypes_module
 
 if TYPE_CHECKING:
-    import dask_expr
+    try:
+        import dask.dataframe.dask_expr as dx
+    except ModuleNotFoundError:
+        import dask_expr as dx
     from typing_extensions import Self
 
     from narwhals._dask.dataframe import DaskLazyFrame
@@ -135,7 +138,7 @@ class DaskSelector(DaskExpr):
     def __or__(self: Self, other: DaskSelector | Any) -> DaskSelector | Any:
         if isinstance(other, DaskSelector):
 
-            def call(df: DaskLazyFrame) -> list[dask_expr.Series]:
+            def call(df: DaskLazyFrame) -> list[dx.Series]:
                 lhs = self._call(df)
                 rhs = other._call(df)
                 return [*(x for x in lhs if x.name not in {x.name for x in rhs}), *rhs]
diff --git a/narwhals/_dask/utils.py b/narwhals/_dask/utils.py
index 4f2952d0..cd303d8e 100644
--- a/narwhals/_dask/utils.py
+++ b/narwhals/_dask/utils.py
@@ -14,7 +14,11 @@ from narwhals.utils import parse_version
 
 if TYPE_CHECKING:
     import dask.dataframe as dd
-    import dask_expr
+
+    try:
+        import dask.dataframe.dask_expr as dx
+    except ModuleNotFoundError:
+        import dask_expr as dx
 
     from narwhals._dask.dataframe import DaskLazyFrame
     from narwhals._dask.expr import DaskExpr
@@ -42,7 +46,7 @@ def maybe_evaluate(df: DaskLazyFrame, obj: Any) -> Any:
 
 def parse_exprs_and_named_exprs(
     df: DaskLazyFrame, *exprs: Any, **named_exprs: Any
-) -> dict[str, dask_expr.Series]:
+) -> dict[str, dx.Series]:
     results = {}
     for expr in exprs:
         if hasattr(expr, "__narwhals_expr__"):
@@ -82,10 +86,13 @@ def add_row_index(
     )
 
 
-def validate_comparand(lhs: dask_expr.Series, rhs: dask_expr.Series) -> None:
-    import dask_expr
+def validate_comparand(lhs: dx.Series, rhs: dx.Series) -> None:
+    try:
+        import dask.dataframe.dask_expr as dx
+    except ModuleNotFoundError:
+        import dask_expr as dx
 
-    if not dask_expr._expr.are_co_aligned(lhs._expr, rhs._expr):  # pragma: no cover
+    if not dx._expr.are_co_aligned(lhs._expr, rhs._expr):  # pragma: no cover
         # are_co_aligned is a method which cheaply checks if two Dask expressions
         # have the same index, and therefore don't require index alignment.
         # If someone only operates on a Dask DataFrame via expressions, then this
@@ -154,11 +161,11 @@ def narwhals_to_native_dtype(dtype: DType | type[DType], version: Version) -> An
     raise AssertionError(msg)
 
 
-def name_preserving_sum(s1: dask_expr.Series, s2: dask_expr.Series) -> dask_expr.Series:
+def name_preserving_sum(s1: dx.Series, s2: dx.Series) -> dx.Series:
     return (s1 + s2).rename(s1.name)
 
 
-def name_preserving_div(s1: dask_expr.Series, s2: dask_expr.Series) -> dask_expr.Series:
+def name_preserving_div(s1: dx.Series, s2: dx.Series) -> dx.Series:
     return (s1 / s2).rename(s1.name)
 
 
diff --git a/narwhals/_duckdb/dataframe.py b/narwhals/_duckdb/dataframe.py
index 76ff68ae..e1c0f994 100644
--- a/narwhals/_duckdb/dataframe.py
+++ b/narwhals/_duckdb/dataframe.py
@@ -224,30 +224,43 @@ class DuckDBLazyFrame:
             left_on = [left_on]
         if isinstance(right_on, str):
             right_on = [right_on]
+        original_alias = self._native_frame.alias
 
-        if how not in ("inner", "left"):
+        if how not in ("inner", "left", "semi", "cross"):
             msg = "Only inner and left join is implemented for DuckDB"
             raise NotImplementedError(msg)
 
-        # help mypy
-        assert left_on is not None  # noqa: S101
-        assert right_on is not None  # noqa: S101
-
-        conditions = [
-            f"lhs.{left} = rhs.{right}" for left, right in zip(left_on, right_on)
-        ]
-        original_alias = self._native_frame.alias
-        condition = " and ".join(conditions)
-        rel = self._native_frame.set_alias("lhs").join(
-            other._native_frame.set_alias("rhs"), condition=condition, how=how
-        )
+        if how == "cross":
+            if self._backend_version < (1, 1, 4):
+                msg = f"DuckDB>=1.1.4 is required for cross-join, found version: {self._backend_version}"
+                raise NotImplementedError(msg)
+            rel = self._native_frame.set_alias("lhs").cross(  # pragma: no cover
+                other._native_frame.set_alias("rhs")
+            )
+        else:
+            # help mypy
+            assert left_on is not None  # noqa: S101
+            assert right_on is not None  # noqa: S101
+
+            conditions = [
+                f"lhs.{left} = rhs.{right}" for left, right in zip(left_on, right_on)
+            ]
+            condition = " and ".join(conditions)
+            rel = self._native_frame.set_alias("lhs").join(
+                other._native_frame.set_alias("rhs"), condition=condition, how=how
+            )
 
-        select = [f"lhs.{x}" for x in self._native_frame.columns]
-        for col in other._native_frame.columns:
-            if col in self._native_frame.columns and col not in right_on:
-                select.append(f"rhs.{col} as {col}{suffix}")
-            elif col not in right_on:
-                select.append(col)
+        if how in ("inner", "left", "cross"):
+            select = [f"lhs.{x}" for x in self._native_frame.columns]
+            for col in other._native_frame.columns:
+                if col in self._native_frame.columns and (
+                    right_on is None or col not in right_on
+                ):
+                    select.append(f"rhs.{col} as {col}{suffix}")
+                elif right_on is None or col not in right_on:
+                    select.append(col)
+        else:  # semi
+            select = [f"lhs.{x}" for x in self._native_frame.columns]
 
         res = rel.select(", ".join(select)).set_alias(original_alias)
         return self._from_native_frame(res)
diff --git a/narwhals/_duckdb/expr.py b/narwhals/_duckdb/expr.py
index 4515cbba..e5e61208 100644
--- a/narwhals/_duckdb/expr.py
+++ b/narwhals/_duckdb/expr.py
@@ -408,6 +408,28 @@ class DuckDBExpr(CompliantExpr["duckdb.Expression"]):
             lambda _input: FunctionExpression("sum", _input), "sum", returns_scalar=True
         )
 
+    def n_unique(self) -> Self:
+        from duckdb import CaseExpression
+        from duckdb import ConstantExpression
+        from duckdb import FunctionExpression
+
+        def func(_input: duckdb.Expression) -> duckdb.Expression:
+            # https://stackoverflow.com/a/79338887/4451315
+            return FunctionExpression(
+                "array_unique", FunctionExpression("array_agg", _input)
+            ) + FunctionExpression(
+                "max",
+                CaseExpression(
+                    condition=_input.isnotnull(), value=ConstantExpression(0)
+                ).otherwise(ConstantExpression(1)),
+            )
+
+        return self._from_call(
+            func,
+            "n_unique",
+            returns_scalar=True,
+        )
+
     def count(self) -> Self:
         from duckdb import FunctionExpression
 
diff --git a/narwhals/_spark_like/dataframe.py b/narwhals/_spark_like/dataframe.py
index e04da7f5..e54a0599 100644
--- a/narwhals/_spark_like/dataframe.py
+++ b/narwhals/_spark_like/dataframe.py
@@ -50,7 +50,7 @@ class SparkLikeLazyFrame:
     def __narwhals_namespace__(self) -> SparkLikeNamespace:
         from narwhals._spark_like.namespace import SparkLikeNamespace
 
-        return SparkLikeNamespace(  # type: ignore[abstract]
+        return SparkLikeNamespace(
             backend_version=self._backend_version, version=self._version
         )
 
diff --git a/narwhals/_spark_like/expr.py b/narwhals/_spark_like/expr.py
index 03529ca9..10fb7622 100644
--- a/narwhals/_spark_like/expr.py
+++ b/narwhals/_spark_like/expr.py
@@ -59,7 +59,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
         # Unused, just for compatibility with PandasLikeExpr
         from narwhals._spark_like.namespace import SparkLikeNamespace
 
-        return SparkLikeNamespace(  # type: ignore[abstract]
+        return SparkLikeNamespace(
             backend_version=self._backend_version, version=self._version
         )
 
@@ -123,7 +123,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
 
     def __add__(self, other: SparkLikeExpr) -> Self:
         return self._from_call(
-            lambda _input, other: _input + other,
+            lambda _input, other: _input.__add__(other),
             "__add__",
             other=other,
             returns_scalar=False,
@@ -131,7 +131,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
 
     def __sub__(self, other: SparkLikeExpr) -> Self:
         return self._from_call(
-            lambda _input, other: _input - other,
+            lambda _input, other: _input.__sub__(other),
             "__sub__",
             other=other,
             returns_scalar=False,
@@ -139,16 +139,66 @@ class SparkLikeExpr(CompliantExpr["Column"]):
 
     def __mul__(self, other: SparkLikeExpr) -> Self:
         return self._from_call(
-            lambda _input, other: _input * other,
+            lambda _input, other: _input.__mul__(other),
             "__mul__",
             other=other,
             returns_scalar=False,
         )
 
-    def __lt__(self, other: SparkLikeExpr) -> Self:
+    def __truediv__(self, other: SparkLikeExpr) -> Self:
         return self._from_call(
-            lambda _input, other: _input < other,
-            "__lt__",
+            lambda _input, other: _input.__truediv__(other),
+            "__truediv__",
+            other=other,
+            returns_scalar=False,
+        )
+
+    def __floordiv__(self, other: SparkLikeExpr) -> Self:
+        def _floordiv(_input: Column, other: Column) -> Column:
+            from pyspark.sql import functions as F  # noqa: N812
+
+            return F.floor(_input / other)
+
+        return self._from_call(
+            _floordiv, "__floordiv__", other=other, returns_scalar=False
+        )
+
+    def __pow__(self, other: SparkLikeExpr) -> Self:
+        return self._from_call(
+            lambda _input, other: _input.__pow__(other),
+            "__pow__",
+            other=other,
+            returns_scalar=False,
+        )
+
+    def __mod__(self, other: SparkLikeExpr) -> Self:
+        return self._from_call(
+            lambda _input, other: _input.__mod__(other),
+            "__mod__",
+            other=other,
+            returns_scalar=False,
+        )
+
+    def __eq__(self, other: SparkLikeExpr) -> Self:  # type: ignore[override]
+        return self._from_call(
+            lambda _input, other: _input.__eq__(other),
+            "__eq__",
+            other=other,
+            returns_scalar=False,
+        )
+
+    def __ne__(self, other: SparkLikeExpr) -> Self:  # type: ignore[override]
+        return self._from_call(
+            lambda _input, other: _input.__ne__(other),
+            "__ne__",
+            other=other,
+            returns_scalar=False,
+        )
+
+    def __ge__(self, other: SparkLikeExpr) -> Self:
+        return self._from_call(
+            lambda _input, other: _input.__ge__(other),
+            "__ge__",
             other=other,
             returns_scalar=False,
         )
@@ -161,6 +211,45 @@ class SparkLikeExpr(CompliantExpr["Column"]):
             returns_scalar=False,
         )
 
+    def __le__(self, other: SparkLikeExpr) -> Self:
+        return self._from_call(
+            lambda _input, other: _input.__le__(other),
+            "__le__",
+            other=other,
+            returns_scalar=False,
+        )
+
+    def __lt__(self, other: SparkLikeExpr) -> Self:
+        return self._from_call(
+            lambda _input, other: _input.__lt__(other),
+            "__lt__",
+            other=other,
+            returns_scalar=False,
+        )
+
+    def __and__(self, other: SparkLikeExpr) -> Self:
+        return self._from_call(
+            lambda _input, other: _input.__and__(other),
+            "__and__",
+            other=other,
+            returns_scalar=False,
+        )
+
+    def __or__(self, other: SparkLikeExpr) -> Self:
+        return self._from_call(
+            lambda _input, other: _input.__or__(other),
+            "__or__",
+            other=other,
+            returns_scalar=False,
+        )
+
+    def __invert__(self) -> Self:
+        return self._from_call(
+            lambda _input: _input.__invert__(),
+            "__invert__",
+            returns_scalar=self._returns_scalar,
+        )
+
     def abs(self) -> Self:
         from pyspark.sql import functions as F  # noqa: N812
 
diff --git a/narwhals/_spark_like/namespace.py b/narwhals/_spark_like/namespace.py
index d34867b0..56cc4d27 100644
--- a/narwhals/_spark_like/namespace.py
+++ b/narwhals/_spark_like/namespace.py
@@ -16,6 +16,7 @@ if TYPE_CHECKING:
 
     from narwhals._spark_like.dataframe import SparkLikeLazyFrame
     from narwhals._spark_like.typing import IntoSparkLikeExpr
+    from narwhals.dtypes import DType
     from narwhals.utils import Version
 
 
@@ -67,6 +68,28 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
             *column_names, backend_version=self._backend_version, version=self._version
         )
 
+    def lit(self, value: object, dtype: DType | None) -> SparkLikeExpr:
+        if dtype is not None:
+            msg = "todo"
+            raise NotImplementedError(msg)
+
+        def _lit(_: SparkLikeLazyFrame) -> list[Column]:
+            import pyspark.sql.functions as F  # noqa: N812
+
+            return [F.lit(value).alias("literal")]
+
+        return SparkLikeExpr(  # type: ignore[abstract]
+            call=_lit,
+            depth=0,
+            function_name="lit",
+            root_names=None,
+            output_names=["literal"],
+            returns_scalar=True,
+            backend_version=self._backend_version,
+            version=self._version,
+            kwargs={},
+        )
+
     def sum_horizontal(self, *exprs: IntoSparkLikeExpr) -> SparkLikeExpr:
         parsed_exprs = parse_into_exprs(*exprs, namespace=self)
 
diff --git a/narwhals/translate.py b/narwhals/translate.py
index 8d0805a2..9ad86801 100644
--- a/narwhals/translate.py
+++ b/narwhals/translate.py
@@ -684,7 +684,10 @@ def _from_native_impl(  # noqa: PLR0915
                 msg = "Cannot only use `eager_only` or `eager_or_interchange_only` with dask DataFrame"
                 raise TypeError(msg)
             return native_object
-        if get_dask_expr() is None:  # pragma: no cover
+        if (
+            parse_version(get_dask().__version__) <= (2024, 12, 1)
+            and get_dask_expr() is None
+        ):  # pragma: no cover
             msg = "Please install dask-expr"
             raise ImportError(msg)
         return LazyFrame(
diff --git a/narwhals/utils.py b/narwhals/utils.py
index c03642c9..509a0e36 100644
--- a/narwhals/utils.py
+++ b/narwhals/utils.py
@@ -372,7 +372,7 @@ def _is_iterable(arg: Any | Iterable[Any]) -> bool:
     return isinstance(arg, Iterable) and not isinstance(arg, (str, bytes, Series))
 
 
-def parse_version(version: Sequence[str | int]) -> tuple[int, ...]:
+def parse_version(version: str) -> tuple[int, ...]:
     """Simple version parser; split into a tuple of ints for comparison.
 
     Arguments:
@@ -382,9 +382,10 @@ def parse_version(version: Sequence[str | int]) -> tuple[int, ...]:
         Parsed version number.
     """
     # lifted from Polars
-    if isinstance(version, str):  # pragma: no cover
-        version = version.split(".")
-    return tuple(int(re.sub(r"\D", "", str(v))) for v in version)
+    # [marco]: Take care of DuckDB pre-releases which end with e.g. `-dev4108`
+    # and pandas pre-releases which end with e.g. .dev0+618.gb552dc95c9
+    version = re.sub(r"(\D?dev.*$)", "", version)
+    return tuple(int(re.sub(r"\D", "", str(v))) for v in version.split("."))
 
 
 def isinstance_or_issubclass(obj: Any, cls: Any) -> bool:
diff --git a/tpch/execute.py b/tpch/execute.py
index 1f3823ce..5209ad48 100644
--- a/tpch/execute.py
+++ b/tpch/execute.py
@@ -40,7 +40,7 @@ BACKEND_COLLECT_FUNC_MAP = {
     "dask": lambda x: x.compute(),
 }
 
-DUCKDB_XFAILS = ["q11", "q14", "q15", "q16", "q18", "q22"]
+DUCKDB_SKIPS = ["q14", "q15"]
 
 QUERY_DATA_PATH_MAP = {
     "q1": (LINEITEM_PATH,),
@@ -95,7 +95,7 @@ def execute_query(query_id: str) -> None:
     data_paths = QUERY_DATA_PATH_MAP[query_id]
 
     for backend, (native_namespace, kwargs) in BACKEND_NAMESPACE_KWARGS_MAP.items():
-        if backend == "duckdb" and query_id in DUCKDB_XFAILS:
+        if backend == "duckdb" and query_id in DUCKDB_SKIPS:
             print(f"\nSkipping {query_id} for DuckDB")  # noqa: T201
             continue
 


Test patch:
diff --git a/tests/expr_and_series/n_unique_test.py b/tests/expr_and_series/n_unique_test.py
index d8e4d9b7..90bffb04 100644
--- a/tests/expr_and_series/n_unique_test.py
+++ b/tests/expr_and_series/n_unique_test.py
@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-import pytest
-
 import narwhals.stable.v1 as nw
 from tests.utils import Constructor
 from tests.utils import ConstructorEager
@@ -13,9 +11,7 @@ data = {
 }
 
 
-def test_n_unique(constructor: Constructor, request: pytest.FixtureRequest) -> None:
-    if "duckdb" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
+def test_n_unique(constructor: Constructor) -> None:
     df = nw.from_native(constructor(data))
     result = df.select(nw.all().n_unique())
     expected = {"a": [3], "b": [4]}
diff --git a/tests/expr_and_series/unary_test.py b/tests/expr_and_series/unary_test.py
index 9ee38a23..f3e01d80 100644
--- a/tests/expr_and_series/unary_test.py
+++ b/tests/expr_and_series/unary_test.py
@@ -126,11 +126,7 @@ def test_unary_two_elements_series(constructor_eager: ConstructorEager) -> None:
     assert_equal_data(result, expected)
 
 
-def test_unary_one_element(
-    constructor: Constructor, request: pytest.FixtureRequest
-) -> None:
-    if "duckdb" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
+def test_unary_one_element(constructor: Constructor) -> None:
     data = {"a": [1], "b": [2], "c": [None]}
     # Dask runs into a divide by zero RuntimeWarning for 1 element skew.
     context = (
diff --git a/tests/frame/join_test.py b/tests/frame/join_test.py
index 7332cb25..5bf5c91f 100644
--- a/tests/frame/join_test.py
+++ b/tests/frame/join_test.py
@@ -10,6 +10,7 @@ import pytest
 
 import narwhals.stable.v1 as nw
 from narwhals.utils import Implementation
+from tests.utils import DUCKDB_VERSION
 from tests.utils import PANDAS_VERSION
 from tests.utils import Constructor
 from tests.utils import assert_equal_data
@@ -75,7 +76,7 @@ def test_inner_join_single_key(constructor: Constructor) -> None:
 
 
 def test_cross_join(constructor: Constructor, request: pytest.FixtureRequest) -> None:
-    if "duckdb" in str(constructor):
+    if "duckdb" in str(constructor) and DUCKDB_VERSION < (1, 1, 4):
         request.applymarker(pytest.mark.xfail)
     data = {"antananarivo": [1, 3, 2]}
     df = nw.from_native(constructor(data))
@@ -117,7 +118,7 @@ def test_suffix(constructor: Constructor, how: str, suffix: str) -> None:
 def test_cross_join_suffix(
     constructor: Constructor, suffix: str, request: pytest.FixtureRequest
 ) -> None:
-    if "duckdb" in str(constructor):
+    if "duckdb" in str(constructor) and DUCKDB_VERSION < (1, 1, 4):
         request.applymarker(pytest.mark.xfail)
     data = {"antananarivo": [1, 3, 2]}
     df = nw.from_native(constructor(data))
@@ -206,10 +207,7 @@ def test_semi_join(
     join_key: list[str],
     filter_expr: nw.Expr,
     expected: dict[str, list[Any]],
-    request: pytest.FixtureRequest,
 ) -> None:
-    if "duckdb" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
     data = {"antananarivo": [1, 3, 2], "bob": [4, 4, 6], "zorro": [7.0, 8, 9]}
     df = nw.from_native(constructor(data))
     other = df.filter(filter_expr)
diff --git a/tests/group_by_test.py b/tests/group_by_test.py
index 0dd6d8a1..c854da45 100644
--- a/tests/group_by_test.py
+++ b/tests/group_by_test.py
@@ -115,8 +115,6 @@ def test_group_by_depth_1_agg(
     expected: dict[str, list[int | float]],
     request: pytest.FixtureRequest,
 ) -> None:
-    if "duckdb" in str(constructor) and attr == "n_unique":
-        request.applymarker(pytest.mark.xfail)
     if "pandas_pyarrow" in str(constructor) and attr == "var" and PANDAS_VERSION < (2, 1):
         # Known issue with variance calculation in pandas 2.0.x with pyarrow backend in groupby operations"
         request.applymarker(pytest.mark.xfail)
@@ -166,11 +164,7 @@ def test_group_by_median(constructor: Constructor) -> None:
     assert_equal_data(result, expected)
 
 
-def test_group_by_n_unique_w_missing(
-    constructor: Constructor, request: pytest.FixtureRequest
-) -> None:
-    if "duckdb" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
+def test_group_by_n_unique_w_missing(constructor: Constructor) -> None:
     data = {"a": [1, 1, 2], "b": [4, None, 5], "c": [None, None, 7], "d": [1, 1, 3]}
     result = (
         nw.from_native(constructor(data))
diff --git a/tests/spark_like_test.py b/tests/spark_like_test.py
index 30610be4..f7cd9e6a 100644
--- a/tests/spark_like_test.py
+++ b/tests/spark_like_test.py
@@ -21,6 +21,7 @@ from tests.utils import assert_equal_data
 if TYPE_CHECKING:
     from pyspark.sql import SparkSession
 
+    from narwhals.dtypes import DType
     from narwhals.typing import IntoFrame
     from tests.utils import Constructor
 
@@ -954,6 +955,53 @@ def test_left_join_overlapping_column(pyspark_constructor: Constructor) -> None:
     assert_equal_data(result, expected)
 
 
+# copied from tests/expr_and_series/arithmetic_test.py
+@pytest.mark.parametrize(
+    ("attr", "rhs", "expected"),
+    [
+        ("__add__", 1, [2, 3, 4]),
+        ("__sub__", 1, [0, 1, 2]),
+        ("__mul__", 2, [2, 4, 6]),
+        ("__truediv__", 2.0, [0.5, 1.0, 1.5]),
+        ("__truediv__", 1, [1, 2, 3]),
+        ("__floordiv__", 2, [0, 1, 1]),
+        ("__mod__", 2, [1, 0, 1]),
+        ("__pow__", 2, [1, 4, 9]),
+    ],
+)
+def test_arithmetic_expr(
+    attr: str, rhs: Any, expected: list[Any], pyspark_constructor: Constructor
+) -> None:
+    data = {"a": [1.0, 2, 3]}
+    df = nw.from_native(pyspark_constructor(data))
+    result = df.select(getattr(nw.col("a"), attr)(rhs))
+    assert_equal_data(result, {"a": expected})
+
+
+@pytest.mark.parametrize(
+    ("attr", "rhs", "expected"),
+    [
+        ("__radd__", 1, [2, 3, 4]),
+        ("__rsub__", 1, [0, -1, -2]),
+        ("__rmul__", 2, [2, 4, 6]),
+        ("__rtruediv__", 2.0, [2, 1, 2 / 3]),
+        ("__rfloordiv__", 2, [2, 1, 0]),
+        ("__rmod__", 2, [0, 0, 2]),
+        ("__rpow__", 2, [2, 4, 8]),
+    ],
+)
+def test_right_arithmetic_expr(
+    attr: str,
+    rhs: Any,
+    expected: list[Any],
+    pyspark_constructor: Constructor,
+) -> None:
+    data = {"a": [1, 2, 3]}
+    df = nw.from_native(pyspark_constructor(data))
+    result = df.select(getattr(nw.col("a"), attr)(rhs))
+    assert_equal_data(result, {"literal": expected})
+
+
 # Copied from tests/expr_and_series/median_test.py
 def test_median(pyspark_constructor: Constructor) -> None:
     data = {"a": [3, 8, 2, None], "b": [5, 5, None, 7], "z": [7.0, 8, 9, None]}
@@ -1099,3 +1147,65 @@ def test_skew(
     df = nw.from_native(pyspark_constructor({"a": data}))
     result = df.select(skew=nw.col("a").skew())
     assert_equal_data(result, {"skew": [expected]})
+
+
+# copied from tests/expr_and_series/list_test.py
+@pytest.mark.parametrize(
+    ("dtype", "expected_lit"),
+    [(None, [2, 2, 2]), (nw.String, ["2", "2", "2"]), (nw.Float32, [2.0, 2.0, 2.0])],
+)
+def test_lit(
+    pyspark_constructor: Constructor,
+    dtype: DType | None,
+    expected_lit: list[Any],
+    request: pytest.FixtureRequest,
+) -> None:
+    if dtype is not None:
+        request.applymarker(pytest.mark.xfail)
+    data = {"a": [1, 3, 2], "b": [4, 4, 6], "z": [7.0, 8, 9]}
+    df_raw = pyspark_constructor(data)
+    df = nw.from_native(df_raw).lazy()
+    result = df.with_columns(nw.lit(2, dtype).alias("lit"))
+    expected = {
+        "a": [1, 3, 2],
+        "b": [4, 4, 6],
+        "z": [7.0, 8.0, 9.0],
+        "lit": expected_lit,
+    }
+    assert_equal_data(result, expected)
+
+
+@pytest.mark.parametrize(
+    ("col_name", "expr", "expected_result"),
+    [
+        ("left_lit", nw.lit(1) + nw.col("a"), [2, 4, 3]),
+        ("right_lit", nw.col("a") + nw.lit(1), [2, 4, 3]),
+        ("left_lit_with_agg", nw.lit(1) + nw.col("a").mean(), [3]),
+        ("right_lit_with_agg", nw.col("a").mean() - nw.lit(1), [1]),
+        ("left_scalar", 1 + nw.col("a"), [2, 4, 3]),
+        ("right_scalar", nw.col("a") + 1, [2, 4, 3]),
+        ("left_scalar_with_agg", 1 + nw.col("a").mean(), [3]),
+        ("right_scalar_with_agg", nw.col("a").mean() - 1, [1]),
+    ],
+)
+def test_lit_operation(
+    pyspark_constructor: Constructor,
+    col_name: str,
+    expr: nw.Expr,
+    expected_result: list[int],
+    request: pytest.FixtureRequest,
+) -> None:
+    if col_name in (
+        "left_scalar_with_agg",
+        "left_lit_with_agg",
+        "right_lit",
+        "right_lit_with_agg",
+    ):
+        request.applymarker(pytest.mark.xfail)
+
+    data = {"a": [1, 3, 2]}
+    df_raw = pyspark_constructor(data)
+    df = nw.from_native(df_raw).lazy()
+    result = df.select(expr.alias(col_name))
+    expected = {col_name: expected_result}
+    assert_equal_data(result, expected)
diff --git a/tests/utils.py b/tests/utils.py
index 005b4eee..2d41d678 100644
--- a/tests/utils.py
+++ b/tests/utils.py
@@ -33,6 +33,7 @@ def get_module_version_as_tuple(module_name: str) -> tuple[int, ...]:
 IBIS_VERSION: tuple[int, ...] = get_module_version_as_tuple("ibis")
 NUMPY_VERSION: tuple[int, ...] = get_module_version_as_tuple("numpy")
 PANDAS_VERSION: tuple[int, ...] = get_module_version_as_tuple("pandas")
+DUCKDB_VERSION: tuple[int, ...] = get_module_version_as_tuple("duckdb")
 POLARS_VERSION: tuple[int, ...] = get_module_version_as_tuple("polars")
 DASK_VERSION: tuple[int, ...] = get_module_version_as_tuple("dask")
 PYARROW_VERSION: tuple[int, ...] = get_module_version_as_tuple("pyarrow")
diff --git a/tests/utils_test.py b/tests/utils_test.py
index 26bd2ecf..e999696d 100644
--- a/tests/utils_test.py
+++ b/tests/utils_test.py
@@ -13,6 +13,7 @@ from pandas.testing import assert_index_equal
 from pandas.testing import assert_series_equal
 
 import narwhals.stable.v1 as nw
+from narwhals.utils import parse_version
 from tests.utils import PANDAS_VERSION
 from tests.utils import get_module_version_as_tuple
 
@@ -271,3 +272,15 @@ def test_generate_temporary_column_name_raise() -> None:
         match="Internal Error: Narwhals was not able to generate a column name with ",
     ):
         nw.generate_temporary_column_name(n_bytes=1, columns=columns)
+
+
+@pytest.mark.parametrize(
+    ("version", "expected"),
+    [
+        ("2020.1.2", (2020, 1, 2)),
+        ("2020.1.2-dev123", (2020, 1, 2)),
+        ("3.0.0.dev0+618.gb552dc95c9", (3, 0, 0)),
+    ],
+)
+def test_parse_version(version: str, expected: tuple[int, ...]) -> None:
+    assert parse_version(version) == expected