feat: Add support for full join

narwhals/_arrow/dataframe.py

@@ -416,7 +416,7 @@ def join(
         self: Self,
         other: Self,
         *,
-        how: Literal["left", "inner", "cross", "anti", "semi"],
+        how: Literal["inner", "left", "full", "cross", "semi", "anti"],
         left_on: Sequence[str] | None,
         right_on: Sequence[str] | None,
         suffix: str,
@@ -426,6 +426,7 @@ def join(
             "semi": "left semi",
             "inner": "inner",
             "left": "left outer",
+            "full": "full outer",
         }
 
         if how == "cross":
@@ -450,13 +451,15 @@ def join(
                 .drop([key_token])
             )
 
+        coalesce_keys = how != "full"  # polars full join does not coalesce keys
         return self._from_native_frame(
             self.native.join(
                 other.native,
                 keys=left_on or [],  # type: ignore[arg-type]
                 right_keys=right_on,  # type: ignore[arg-type]
                 join_type=how_to_join_map[how],
                 right_suffix=suffix,
+                coalesce_keys=coalesce_keys,
             ),
         )
 

narwhals/_compliant/dataframe.py

@@ -107,7 +107,7 @@ def join(
         self: Self,
         other: Self,
         *,
-        how: Literal["left", "inner", "cross", "anti", "semi"],
+        how: Literal["inner", "left", "full", "cross", "semi", "anti"],
         left_on: Sequence[str] | None,
         right_on: Sequence[str] | None,
         suffix: str,

narwhals/_dask/dataframe.py

@@ -17,7 +17,9 @@
 from narwhals.typing import CompliantDataFrame
 from narwhals.typing import CompliantLazyFrame
 from narwhals.utils import Implementation
+from narwhals.utils import _remap_full_join_keys
 from narwhals.utils import check_column_exists
+from narwhals.utils import check_column_names_are_unique
 from narwhals.utils import generate_temporary_column_name
 from narwhals.utils import not_implemented
 from narwhals.utils import parse_columns_to_drop
@@ -255,7 +257,7 @@ def join(
         self: Self,
         other: Self,
         *,
-        how: Literal["left", "inner", "cross", "anti", "semi"],
+        how: Literal["inner", "left", "full", "cross", "semi", "anti"],
         left_on: Sequence[str] | None,
         right_on: Sequence[str] | None,
         suffix: str,
@@ -350,6 +352,30 @@ def join(
                     extra.append(f"{right_key}_right")
             return self._from_native_frame(result_native.drop(columns=extra))
 
+        if how == "full":
+            # dask does not retain keys post-join
+            # we must append the suffix to each key before-hand
+
+            # help mypy
+            assert left_on is not None  # noqa: S101
+            assert right_on is not None  # noqa: S101
+
+            right_on_mapper = _remap_full_join_keys(left_on, right_on, suffix)
+
+            other_native = other._native_frame
+            other_native = other_native.rename(columns=right_on_mapper)
+            check_column_names_are_unique(other_native.columns)
+            right_on = list(right_on_mapper.values())  # we now have the suffixed keys
+            return self._from_native_frame(
+                self._native_frame.merge(
+                    other_native,
+                    left_on=left_on,
+                    right_on=right_on,
+                    how="outer",
+                    suffixes=("", suffix),
+                ),
+            )
+
         return self._from_native_frame(
             self._native_frame.merge(
                 other._native_frame,

narwhals/_duckdb/dataframe.py

@@ -252,14 +252,16 @@ def join(
         self: Self,
         other: Self,
         *,
-        how: Literal["left", "inner", "cross", "anti", "semi"],
+        how: Literal["inner", "left", "full", "cross", "semi", "anti"],
         left_on: Sequence[str] | None,
         right_on: Sequence[str] | None,
         suffix: str,
     ) -> Self:
         original_alias = self._native_frame.alias
 
-        if how == "cross":
+        native_how = "outer" if how == "full" else how
+
+        if native_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)
@@ -274,16 +276,20 @@ def join(
             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
+                other._native_frame.set_alias("rhs"), condition=condition, how=native_how
             )
 
-        if how in {"inner", "left", "cross"}:
+        if native_how in {"inner", "left", "cross", "outer"}:
             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
+                col_in_lhs: bool = col in self._native_frame.columns
+                if native_how == "outer" and not col_in_lhs:
+                    select.append(f'rhs."{col}"')
+                elif (native_how == "outer") or (
+                    col_in_lhs 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:

narwhals/_pandas_like/dataframe.py

@@ -28,6 +28,7 @@
 from narwhals.exceptions import InvalidOperationError
 from narwhals.exceptions import ShapeError
 from narwhals.utils import Implementation
+from narwhals.utils import _remap_full_join_keys
 from narwhals.utils import check_column_exists
 from narwhals.utils import generate_temporary_column_name
 from narwhals.utils import import_dtypes_module
@@ -600,7 +601,7 @@ def join(
         self: Self,
         other: Self,
         *,
-        how: Literal["left", "inner", "cross", "anti", "semi"],
+        how: Literal["inner", "left", "full", "cross", "semi", "anti"],
         left_on: Sequence[str] | None,
         right_on: Sequence[str] | None,
         suffix: str,
@@ -723,6 +724,29 @@ def join(
                     extra.append(f"{right_key}{suffix}")
             return self._from_native_frame(result_native.drop(columns=extra))
 
+        if how == "full":
+            # Pandas coalesces keys in full joins unless there's no collision
+
+            # help mypy
+            assert left_on is not None  # noqa: S101
+            assert right_on is not None  # noqa: S101
+
+            right_on_mapper = _remap_full_join_keys(left_on, right_on, suffix)
+
+            other_native = other._native_frame
+            other_native = other_native.rename(columns=right_on_mapper)
+            check_column_names_are_unique(other_native.columns)
+            right_on = list(right_on_mapper.values())  # we now have the suffixed keys
+            return self._from_native_frame(
+                self._native_frame.merge(
+                    other_native,
+                    left_on=left_on,
+                    right_on=right_on,
+                    how="outer",
+                    suffixes=("", suffix),
+                ),
+            )
+
         return self._from_native_frame(
             self.native.merge(
                 other._native_frame,

narwhals/_polars/dataframe.py

@@ -62,7 +62,6 @@ class PolarsDataFrame:
     item: Method[Any]
     iter_rows: Method[Iterator[tuple[Any, ...]] | Iterator[Mapping[str, Any]]]
     is_unique: Method[PolarsSeries]
-    join: Method[Self]
     join_asof: Method[Self]
     rename: Method[Self]
     row: Method[tuple[Any, ...]]
@@ -417,14 +416,36 @@ def pivot(
     def to_polars(self: Self) -> pl.DataFrame:
         return self.native
 
+    def join(
+        self: Self,
+        other: Self,
+        *,
+        how: Literal["inner", "left", "full", "cross", "semi", "anti"],
+        left_on: Sequence[str] | None,
+        right_on: Sequence[str] | None,
+        suffix: str,
+    ) -> Self:
+        how_native = (
+            "outer" if (self._backend_version < (1, 0, 0) and how == "full") else how
+        )
+
+        return self._from_native_frame(
+            self._native_frame.join(
+                other=other._native_frame,
+                how=how_native,  # type: ignore[arg-type]
+                left_on=left_on,
+                right_on=right_on,
+                suffix=suffix,
+            )
+        )
+
 
 class PolarsLazyFrame:
     drop_nulls: Method[Self]
     explode: Method[Self]
     filter: Method[Self]
     gather_every: Method[Self]
     head: Method[Self]
-    join: Method[Self]
     join_asof: Method[Self]
     rename: Method[Self]
     select: Method[Self]
@@ -616,3 +637,26 @@ def simple_select(self, *column_names: str) -> Self:
 
     def aggregate(self: Self, *exprs: Any) -> Self:
         return self.select(*exprs)
+
+    def join(
+        self: Self,
+        other: Self,
+        *,
+        how: Literal["inner", "left", "full", "cross", "semi", "anti"],
+        left_on: Sequence[str] | None,
+        right_on: Sequence[str] | None,
+        suffix: str,
+    ) -> Self:
+        how_native = (
+            "outer" if (self._backend_version < (1, 0, 0) and how == "full") else how
+        )
+
+        return self._from_native_frame(
+            self._native_frame.join(
+                other=other._native_frame,
+                how=how_native,  # type: ignore[arg-type]
+                left_on=left_on,
+                right_on=right_on,
+                suffix=suffix,
+            )
+        )

narwhals/_spark_like/dataframe.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import warnings
+from functools import reduce
+from operator import and_
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Iterator
@@ -328,7 +330,7 @@ def unique(
     def join(
         self: Self,
         other: Self,
-        how: Literal["inner", "left", "cross", "semi", "anti"],
+        how: Literal["inner", "left", "full", "cross", "semi", "anti"],
         left_on: Sequence[str] | None,
         right_on: Sequence[str] | None,
         suffix: str,
@@ -339,14 +341,23 @@ def join(
         left_columns = self.columns
         right_columns = other.columns
 
+        right_on_: list[str] = list(right_on) if right_on is not None else []
+        left_on_: list[str] = list(left_on) if left_on is not None else []
+
         # create a mapping for columns on other
         # `right_on` columns will be renamed as `left_on`
         # the remaining columns will be either added the suffix or left unchanged.
+        right_cols_to_rename = (
+            [c for c in right_columns if c not in right_on_]
+            if how != "full"
+            else right_columns
+        )
+
         rename_mapping = {
-            **dict(zip(right_on or [], left_on or [])),
+            **dict(zip(right_on_, left_on_)),
             **{
                 colname: f"{colname}{suffix}" if colname in left_columns else colname
-                for colname in list(set(right_columns).difference(set(right_on or [])))
+                for colname in right_cols_to_rename
             },
         }
         other_native = other_native.select(
@@ -363,12 +374,30 @@ def join(
                 [
                     rename_mapping[colname]
                     for colname in right_columns
-                    if colname not in (right_on or [])
+                    if colname not in right_on_
                 ]
             )
-        on = list(left_on) if left_on else None
+        elif how == "full":
+            col_order.extend(rename_mapping.values())
+
+        right_on_remapped = [rename_mapping[c] for c in right_on_]
+        on_ = (
+            reduce(
+                and_,
+                (
+                    getattr(self_native, left_key) == getattr(other_native, right_key)
+                    for left_key, right_key in zip(left_on_, right_on_remapped)
+                ),
+            )
+            if how == "full"
+            else None
+            if how == "cross"
+            else left_on_
+        )
+        how_native = "full_outer" if how == "full" else how
+
         return self._from_native_frame(
-            self_native.join(other_native, on=on, how=how).select(col_order)
+            self_native.join(other_native, on=on_, how=how_native).select(col_order)
         )
 
     def explode(self: Self, columns: Sequence[str]) -> Self:

narwhals/dataframe.py

@@ -237,7 +237,7 @@ def join(
         self: Self,
         other: Self,
         on: str | list[str] | None = None,
-        how: Literal["inner", "left", "cross", "semi", "anti"] = "inner",
+        how: Literal["inner", "left", "full", "cross", "semi", "anti"] = "inner",
         *,
         left_on: str | list[str] | None = None,
         right_on: str | list[str] | None = None,
@@ -247,7 +247,9 @@ def join(
         left_on = [left_on] if isinstance(left_on, str) else left_on
         right_on = [right_on] if isinstance(right_on, str) else right_on
 
-        if how not in (_supported_joins := ("inner", "left", "cross", "anti", "semi")):
+        if how not in (
+            _supported_joins := ("inner", "left", "full", "cross", "anti", "semi")
+        ):
             msg = f"Only the following join strategies are supported: {_supported_joins}; found '{how}'."
             raise NotImplementedError(msg)
 
@@ -1622,7 +1624,7 @@ def join(
         self: Self,
         other: Self,
         on: str | list[str] | None = None,
-        how: Literal["inner", "left", "cross", "semi", "anti"] = "inner",
+        how: Literal["inner", "left", "full", "cross", "semi", "anti"] = "inner",
         *,
         left_on: str | list[str] | None = None,
         right_on: str | list[str] | None = None,
@@ -1638,6 +1640,7 @@ def join(
 
                   * *inner*: Returns rows that have matching values in both tables.
                   * *left*: Returns all rows from the left table, and the matched rows from the right table.
+                  * *full*: Returns all rows in both dataframes, with the suffix appended to the right join keys.
                   * *cross*: Returns the Cartesian product of rows from both tables.
                   * *semi*: Filter rows that have a match in the right table.
                   * *anti*: Filter rows that do not have a match in the right table.
@@ -2909,7 +2912,7 @@ def join(
         self: Self,
         other: Self,
         on: str | list[str] | None = None,
-        how: Literal["inner", "left", "cross", "semi", "anti"] = "inner",
+        how: Literal["inner", "left", "full", "cross", "semi", "anti"] = "inner",
         *,
         left_on: str | list[str] | None = None,
         right_on: str | list[str] | None = None,
@@ -2925,6 +2928,7 @@ def join(
 
                   * *inner*: Returns rows that have matching values in both tables.
                   * *left*: Returns all rows from the left table, and the matched rows from the right table.
+                  * *full*: Returns all rows in both dataframes, with the suffix appended to the right join keys.
                   * *cross*: Returns the Cartesian product of rows from both tables.
                   * *semi*: Filter rows that have a match in the right table.
                   * *anti*: Filter rows that do not have a match in the right table.