feat: add nulls_equal to join

src/narwhals/_arrow/dataframe.py

@@ -421,6 +421,7 @@ def join(
         left_on: Sequence[str] | None,
         right_on: Sequence[str] | None,
         suffix: str,
+        nulls_equal: bool,
     ) -> Self:
         how_to_join_map: dict[str, JoinType] = {
             "anti": "left anti",
@@ -450,6 +451,11 @@ def join(
                 .drop([key_token])
             )
 
+        if nulls_equal and left_on and right_on:
+            return self._join_nulls_equal(
+                other, how, left_on, right_on, suffix, how_to_join_map[how]
+            )
+
         coalesce_keys = how != "full"  # polars full join does not coalesce keys
         return self._with_native(
             self.native.join(
@@ -462,6 +468,57 @@ def join(
             )
         )
 
+    def _join_nulls_equal(
+        self,
+        other: Self,
+        how: JoinStrategy,
+        left_on: Sequence[str],
+        right_on: Sequence[str],
+        suffix: str,
+        join_type: JoinType,
+        /,
+    ) -> Self:
+        # pyarrow drops null join keys, so join on a null-safe encoding per key: an
+        # `is_null` flag and the value cast to string, so a real "" stays distinct from null.
+        token = generate_temporary_column_name(8, [*self.columns, *other.columns])
+        left, right = self.native, other.native
+        keys: list[str] = []
+        for i, (left_key, right_key) in enumerate(zip(left_on, right_on, strict=True)):
+            filled, missing = f"{token}_v{i}", f"{token}_n{i}"
+            left_filled = pc.fill_null(
+                pc.cast(self.native[left_key], pa.large_string()),
+                "",  # type: ignore[type-var]  # pyright: ignore[reportArgumentType]
+            )
+            right_filled = pc.fill_null(
+                pc.cast(other.native[right_key], pa.large_string()),
+                "",  # type: ignore[type-var]  # pyright: ignore[reportArgumentType]
+            )
+            left = left.append_column(filled, left_filled).append_column(  # type: ignore[arg-type]
+                missing, pc.is_null(self.native[left_key])
+            )
+            right = right.append_column(filled, right_filled).append_column(  # type: ignore[arg-type]
+                missing, pc.is_null(other.native[right_key])
+            )
+            keys.extend((filled, missing))
+
+        joined = left.join(
+            right,
+            keys=keys,
+            right_keys=keys,
+            join_type=join_type,
+            right_suffix=suffix,
+            coalesce_keys=how != "full",
+        )
+        joined = joined.drop([c for c in joined.column_names if c.startswith(token)])
+        if how not in {"full", "anti", "semi"}:
+            # polars coalesces the key columns; drop the right key that pyarrow kept.
+            drop = dict.fromkeys(
+                f"{right_key}{suffix}" if right_key in self.columns else right_key
+                for right_key in right_on
+            )
+            joined = joined.drop(list(drop))
+        return self._with_native(joined)
+
     join_asof = not_implemented()
 
     def drop(self, columns: Sequence[str], *, strict: bool) -> Self:

src/narwhals/_arrow/expr.py

@@ -188,6 +188,7 @@ def func(df: ArrowDataFrame) -> Sequence[ArrowSeries]:  # noqa: PLR0914
                     left_on=partition_by,
                     right_on=partition_by,
                     suffix="_right",
+                    nulls_equal=False,
                 )
                 return [tmp.get_column(alias) for alias in aliases]
 
@@ -226,6 +227,7 @@ def func(df: ArrowDataFrame) -> Sequence[ArrowSeries]:  # noqa: PLR0914
                 right_on=group_keys,
                 how="inner",
                 suffix="_right",
+                nulls_equal=False,
             )
             return [ret.get_column(alias) for alias in aliases]
 

src/narwhals/_compliant/dataframe.py

@@ -138,6 +138,7 @@ def join(
         left_on: Sequence[str] | None,
         right_on: Sequence[str] | None,
         suffix: str,
+        nulls_equal: bool,
     ) -> Self: ...
     def join_asof(
         self,

src/narwhals/_dask/dataframe.py

@@ -297,9 +297,18 @@ def top_k(self, k: int, *, by: Iterable[str], reverse: bool | Sequence[bool]) ->
         )
 
     def _join_inner(
-        self, other: Self, *, left_on: Sequence[str], right_on: Sequence[str], suffix: str
+        self,
+        other: Self,
+        *,
+        left_on: Sequence[str],
+        right_on: Sequence[str],
+        suffix: str,
+        nulls_equal: bool,
     ) -> dd.DataFrame:
-        return self.native.dropna(subset=left_on, how="any").merge(
+        left_native = (
+            self.native if nulls_equal else self.native.dropna(subset=left_on, how="any")
+        )
+        return left_native.merge(
             other.native,
             left_on=left_on,
             right_on=right_on,
@@ -308,10 +317,21 @@ def _join_inner(
         )
 
     def _join_left(
-        self, other: Self, *, left_on: Sequence[str], right_on: Sequence[str], suffix: str
+        self,
+        other: Self,
+        *,
+        left_on: Sequence[str],
+        right_on: Sequence[str],
+        suffix: str,
+        nulls_equal: bool,
     ) -> dd.DataFrame:
+        right_native = (
+            other.native
+            if nulls_equal
+            else other.native.dropna(subset=right_on, how="any")
+        )
         result_native = self.native.merge(
-            other.native.dropna(subset=right_on, how="any"),
+            right_native,
             how="left",
             left_on=left_on,
             right_on=right_on,
@@ -325,7 +345,13 @@ def _join_left(
         return result_native.drop(columns=extra)
 
     def _join_full(
-        self, other: Self, *, left_on: Sequence[str], right_on: Sequence[str], suffix: str
+        self,
+        other: Self,
+        *,
+        left_on: Sequence[str],
+        right_on: Sequence[str],
+        suffix: str,
+        nulls_equal: bool,
     ) -> dd.DataFrame:
         # dask does not retain keys post-join
         # we must append the suffix to each key before-hand
@@ -335,6 +361,16 @@ def _join_full(
         check_column_names_are_unique(other_native.columns)
         right_suffixed = list(right_on_mapper.values())
 
+        if nulls_equal:
+            # dask merges null keys natively, so a plain outer merge is enough.
+            return self_native.merge(
+                other_native,
+                left_on=left_on,
+                right_on=right_suffixed,
+                how="outer",
+                suffixes=("", suffix),
+            )
+
         left_null_mask = self_native[list(left_on)].isna().any(axis=1)
         right_null_mask = other_native[right_suffixed].isna().any(axis=1)
 
@@ -374,19 +410,32 @@ def _join_cross(self, other: Self, *, suffix: str) -> dd.DataFrame:
         )
 
     def _join_semi(
-        self, other: Self, *, left_on: Sequence[str], right_on: Sequence[str]
+        self,
+        other: Self,
+        *,
+        left_on: Sequence[str],
+        right_on: Sequence[str],
+        nulls_equal: bool,
     ) -> dd.DataFrame:
         other_native = self._join_filter_rename(
             other=other,
             columns_to_select=list(right_on),
             columns_mapping=dict(zip(right_on, left_on, strict=False)),
         )
-        return self.native.dropna(subset=left_on, how="any").merge(
+        left_native = (
+            self.native if nulls_equal else self.native.dropna(subset=left_on, how="any")
+        )
+        return left_native.merge(
             other_native, how="inner", left_on=left_on, right_on=left_on
         )
 
     def _join_anti(
-        self, other: Self, *, left_on: Sequence[str], right_on: Sequence[str]
+        self,
+        other: Self,
+        *,
+        left_on: Sequence[str],
+        right_on: Sequence[str],
+        nulls_equal: bool,
     ) -> dd.DataFrame:
         indicator_token = generate_temporary_column_name(
             n_bytes=8, columns=(*self.columns, *other.columns), prefix="join_indicator_"
@@ -396,8 +445,13 @@ def _join_anti(
             columns_to_select=list(right_on),
             columns_mapping=dict(zip(right_on, left_on, strict=False)),
         )
+        right_native = (
+            other_native
+            if nulls_equal
+            else other_native.dropna(subset=left_on, how="any")
+        )
         df = self.native.merge(
-            other_native.dropna(subset=left_on, how="any"),
+            right_native,
             how="left",
             indicator=indicator_token,  # pyright: ignore[reportArgumentType]
             left_on=left_on,
@@ -430,6 +484,7 @@ def join(
         left_on: Sequence[str] | None,
         right_on: Sequence[str] | None,
         suffix: str,
+        nulls_equal: bool,
     ) -> Self:
         if how == "cross":
             result = self._join_cross(other=other, suffix=suffix)
@@ -439,19 +494,35 @@ def join(
 
         elif how == "inner":
             result = self._join_inner(
-                other=other, left_on=left_on, right_on=right_on, suffix=suffix
+                other=other,
+                left_on=left_on,
+                right_on=right_on,
+                suffix=suffix,
+                nulls_equal=nulls_equal,
             )
         elif how == "anti":
-            result = self._join_anti(other=other, left_on=left_on, right_on=right_on)
+            result = self._join_anti(
+                other=other, left_on=left_on, right_on=right_on, nulls_equal=nulls_equal
+            )
         elif how == "semi":
-            result = self._join_semi(other=other, left_on=left_on, right_on=right_on)
+            result = self._join_semi(
+                other=other, left_on=left_on, right_on=right_on, nulls_equal=nulls_equal
+            )
         elif how == "left":
             result = self._join_left(
-                other=other, left_on=left_on, right_on=right_on, suffix=suffix
+                other=other,
+                left_on=left_on,
+                right_on=right_on,
+                suffix=suffix,
+                nulls_equal=nulls_equal,
             )
         elif how == "full":
             result = self._join_full(
-                other=other, left_on=left_on, right_on=right_on, suffix=suffix
+                other=other,
+                left_on=left_on,
+                right_on=right_on,
+                suffix=suffix,
+                nulls_equal=nulls_equal,
             )
         else:
             assert_never(how)

src/narwhals/_duckdb/dataframe.py

@@ -56,6 +56,15 @@
     from narwhals.typing import AsofJoinStrategy, JoinStrategy, UniqueKeepStrategy
 
 
+def _join_key_condition(
+    lhs: Expression, rhs: Expression, *, nulls_equal: bool
+) -> Expression:
+    # `IS NOT DISTINCT FROM` when nulls should match, plain equality otherwise.
+    if nulls_equal:
+        return (lhs == rhs) | (lhs.isnull() & rhs.isnull())  # noqa: PD003
+    return lhs == rhs
+
+
 class DuckDBLazyFrame(
     SQLLazyFrame[
         "DuckDBExpr",
@@ -296,6 +305,7 @@ def join(
         left_on: Sequence[str] | None,
         right_on: Sequence[str] | None,
         suffix: str,
+        nulls_equal: bool,
     ) -> Self:
         native_how: Literal["inner", "left", "outer", "cross", "semi", "anti"] = (
             "outer" if how == "full" else how
@@ -311,7 +321,9 @@ def join(
             assert left_on is not None  # noqa: S101
             assert right_on is not None  # noqa: S101
             it = (
-                col(f'lhs."{left}"') == col(f'rhs."{right}"')
+                _join_key_condition(
+                    col(f'lhs."{left}"'), col(f'rhs."{right}"'), nulls_equal=nulls_equal
+                )
                 for left, right in zip(left_on, right_on, strict=True)
             )
             condition: Expression = reduce(and_, it)

src/narwhals/_ibis/dataframe.py

@@ -248,6 +248,7 @@ def join(
         left_on: Sequence[str] | None,
         right_on: Sequence[str] | None,
         suffix: str,
+        nulls_equal: bool,
     ) -> Self:
         how_native = "outer" if how == "full" else how
         rname = "{name}" + suffix
@@ -260,7 +261,9 @@ def join(
         # help mypy
         assert left_on is not None  # noqa: S101
         assert right_on is not None  # noqa: S101
-        predicates = self._convert_predicates(other, left_on, right_on)
+        predicates = self._convert_predicates(
+            other, left_on, right_on, nulls_equal=nulls_equal
+        )
         joined = self.native.join(other.native, predicates, how=how_native, rname=rname)
         if how_native == "left":
             right_names = (n + suffix for n in right_on)
@@ -274,6 +277,11 @@ def join(
                     to_drop.append(right)
             if to_drop:
                 joined = joined.drop(*to_drop)
+        elif nulls_equal and how_native == "inner":
+            # `identical_to` keeps both key columns; drop the duplicate right key to
+            # coalesce, like the name-based join.
+            right_names = (n + suffix for n in right_on)
+            joined = self._join_drop_duplicate_columns(joined, right_names)
         return self._with_native(joined)
 
     def join_asof(
@@ -305,10 +313,24 @@ def join_asof(
         return self._with_native(joined)
 
     def _convert_predicates(
-        self, other: Self, left_on: Sequence[str], right_on: Sequence[str]
+        self,
+        other: Self,
+        left_on: Sequence[str],
+        right_on: Sequence[str],
+        *,
+        nulls_equal: bool = False,
     ) -> JoinPredicates:
-        if left_on == right_on:
+        if left_on == right_on and not nulls_equal:
             return left_on
+        if nulls_equal:
+            # `identical_to` is a null-safe equality (null matches null).
+            return [
+                cast(
+                    "ir.BooleanColumn",
+                    self.native[left].identical_to(other.native[right]),
+                )
+                for left, right in zip(left_on, right_on, strict=True)
+            ]
         return [
             cast("ir.BooleanColumn", (self.native[left] == other.native[right]))
             for left, right in zip(left_on, right_on, strict=True)