fix: raise when join introduces duplicates

narwhals/_arrow/dataframe.py

@@ -21,6 +21,7 @@
 from narwhals.utils import Implementation
 from narwhals.utils import Version
 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 is_sequence_but_not_str
 from narwhals.utils import parse_columns_to_drop
@@ -90,10 +91,15 @@ def _change_version(self: Self, version: Version) -> Self:
             self._native_frame, backend_version=self._backend_version, version=version
         )
 
-    def _from_native_frame(self: Self, df: pa.Table) -> Self:
-        return self.__class__(
+    def _from_native_frame(
+        self: Self, df: pa.Table, *, validate_column_names: bool = False
+    ) -> Self:
+        result = self.__class__(
             df, backend_version=self._backend_version, version=self._version
         )
+        if validate_column_names:
+            check_column_names_are_unique(result.columns)
+        return result
 
     @property
     def shape(self: Self) -> tuple[int, int]:
@@ -367,6 +373,7 @@ def join(
                     right_suffix=suffix,
                 )
                 .drop([key_token]),
+                validate_column_names=True,
             )
 
         return self._from_native_frame(
@@ -377,6 +384,7 @@ def join(
                 join_type=how_to_join_map[how],
                 right_suffix=suffix,
             ),
+            validate_column_names=True,
         )
 
     def join_asof(
@@ -472,7 +480,8 @@ def with_row_index(self: Self, name: str) -> Self:
 
         row_indices = pa.array(range(df.num_rows))
         return self._from_native_frame(
-            df.append_column(name, row_indices).select([name, *cols])
+            df.append_column(name, row_indices).select([name, *cols]),
+            validate_column_names=True,
         )
 
     def filter(self: Self, *predicates: IntoArrowExpr, **constraints: Any) -> Self:
@@ -634,7 +643,9 @@ def item(self: Self, row: int | None, column: int | str | None) -> Any:
     def rename(self: Self, mapping: dict[str, str]) -> Self:
         df = self._native_frame
         new_cols = [mapping.get(c, c) for c in df.column_names]
-        return self._from_native_frame(df.rename_columns(new_cols))
+        return self._from_native_frame(
+            df.rename_columns(new_cols), validate_column_names=True
+        )
 
     def write_parquet(self: Self, file: str | Path | BytesIO) -> None:
         import pyarrow.parquet as pp
@@ -802,7 +813,8 @@ def unpivot(
                     for on_col in on_
                 ],
                 **promote_kwargs,
-            )
+            ),
+            validate_column_names=True,
         )
         # TODO(Unassigned): Even with promote_options="permissive", pyarrow does not
         # upcast numeric to non-numeric (e.g. string) datatypes

narwhals/_dask/dataframe.py

@@ -17,6 +17,7 @@
 from narwhals.typing import CompliantLazyFrame
 from narwhals.utils import Implementation
 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 parse_columns_to_drop
 from narwhals.utils import parse_version
@@ -68,10 +69,15 @@ def _change_version(self: Self, version: Version) -> Self:
             self._native_frame, backend_version=self._backend_version, version=version
         )
 
-    def _from_native_frame(self: Self, df: Any) -> Self:
-        return self.__class__(
+    def _from_native_frame(
+        self: Self, df: Any, *, validate_column_names: bool = False
+    ) -> Self:
+        result = self.__class__(
             df, backend_version=self._backend_version, version=self._version
         )
+        if validate_column_names:
+            check_column_names_are_unique(result.columns)
+        return result
 
     def with_columns(self: Self, *exprs: DaskExpr, **named_exprs: DaskExpr) -> Self:
         df = self._native_frame
@@ -278,6 +284,7 @@ def join(
                     suffixes=("", suffix),
                 )
                 .drop(columns=key_token),
+                validate_column_names=True,
             )
 
         if how == "anti":
@@ -308,7 +315,8 @@ def join(
                 right_on=left_on,
             )
             return self._from_native_frame(
-                df[df[indicator_token] == "left_only"].drop(columns=[indicator_token])
+                df[df[indicator_token] == "left_only"].drop(columns=[indicator_token]),
+                validate_column_names=True,
             )
 
         if how == "semi":
@@ -333,7 +341,8 @@ def join(
                     how="inner",
                     left_on=left_on,
                     right_on=left_on,
-                )
+                ),
+                validate_column_names=True,
             )
 
         if how == "left":
@@ -351,7 +360,9 @@ def join(
                     extra.append(right_key)
                 elif right_key != left_key:
                     extra.append(f"{right_key}_right")
-            return self._from_native_frame(result_native.drop(columns=extra))
+            return self._from_native_frame(
+                result_native.drop(columns=extra), validate_column_names=True
+            )
 
         return self._from_native_frame(
             self._native_frame.merge(
@@ -361,6 +372,7 @@ def join(
                 how=how,
                 suffixes=("", suffix),
             ),
+            validate_column_names=True,
         )
 
     def join_asof(
@@ -386,6 +398,7 @@ def join_asof(
                 direction=strategy,
                 suffixes=("", suffix),
             ),
+            validate_column_names=True,
         )
 
     def group_by(self: Self, *by: str, drop_null_keys: bool) -> DaskLazyGroupBy:
@@ -428,5 +441,6 @@ def unpivot(
                 value_vars=on,
                 var_name=variable_name,
                 value_name=value_name,
-            )
+            ),
+            validate_column_names=True,
         )

narwhals/_dask/group_by.py

@@ -94,14 +94,6 @@ def agg(
             self._grouped,
             exprs,
             self._keys,
-            self._from_native_frame,
-        )
-
-    def _from_native_frame(self: Self, df: DaskLazyFrame) -> DaskLazyFrame:
-        from narwhals._dask.dataframe import DaskLazyFrame
-
-        return DaskLazyFrame(
-            df, backend_version=self._df._backend_version, version=self._df._version
         )
 
 
@@ -110,7 +102,6 @@ def agg_dask(
     grouped: Any,
     exprs: Sequence[CompliantExpr[dx.Series]],
     keys: list[str],
-    from_dataframe: Callable[[Any], DaskLazyFrame],
 ) -> DaskLazyFrame:
     """This should be the fastpath, but cuDF is too far behind to use it.
 
@@ -163,7 +154,9 @@ def agg_dask(
                 }
             )
         result_simple = grouped.agg(**simple_aggregations)
-        return from_dataframe(result_simple.reset_index())
+        return df._from_native_frame(
+            result_simple.reset_index(), validate_column_names=True
+        )
 
     msg = (
         "Non-trivial complex aggregation found.\n\n"

narwhals/_duckdb/dataframe.py

@@ -20,6 +20,7 @@
 from narwhals.typing import CompliantDataFrame
 from narwhals.utils import Implementation
 from narwhals.utils import Version
+from narwhals.utils import check_column_names_are_unique
 from narwhals.utils import generate_temporary_column_name
 from narwhals.utils import import_dtypes_module
 from narwhals.utils import parse_columns_to_drop
@@ -252,10 +253,15 @@ def _change_version(self: Self, version: Version) -> Self:
             self._native_frame, version=version, backend_version=self._backend_version
         )
 
-    def _from_native_frame(self: Self, df: duckdb.DuckDBPyRelation) -> Self:
-        return self.__class__(
+    def _from_native_frame(
+        self: Self, df: duckdb.DuckDBPyRelation, *, validate_column_names: bool = False
+    ) -> Self:
+        result = self.__class__(
             df, backend_version=self._backend_version, version=self._version
         )
+        if validate_column_names:
+            check_column_names_are_unique(result.columns)
+        return result
 
     def group_by(self: Self, *keys: str, drop_null_keys: bool) -> DuckDBGroupBy:
         from narwhals._duckdb.group_by import DuckDBGroupBy
@@ -269,7 +275,9 @@ def rename(self: Self, mapping: dict[str, str]) -> Self:
         selection = [
             f"{col} as {mapping[col]}" if col in mapping else col for col in df.columns
         ]
-        return self._from_native_frame(df.select(", ".join(selection)))
+        return self._from_native_frame(
+            df.select(", ".join(selection)), validate_column_names=True
+        )
 
     def join(
         self: Self,
@@ -319,7 +327,7 @@ def join(
             select = ["lhs.*"]
 
         res = rel.select(", ".join(select)).set_alias(original_alias)
-        return self._from_native_frame(res)
+        return self._from_native_frame(res, validate_column_names=True)
 
     def join_asof(
         self: Self,
@@ -518,4 +526,4 @@ def unpivot(
             select {cols_to_select}
             from unpivot_cte;
             """  # noqa: S608
-        return self._from_native_frame(duckdb.sql(query))
+        return self._from_native_frame(duckdb.sql(query), validate_column_names=True)

narwhals/_duckdb/group_by.py

@@ -49,5 +49,6 @@ def agg(self: Self, *exprs: DuckDBExpr) -> DuckDBLazyFrame:
             )
 
         return self._compliant_frame._from_native_frame(
-            self._compliant_frame._native_frame.aggregate(agg_columns)
+            self._compliant_frame._native_frame.aggregate(agg_columns),
+            validate_column_names=True,
         )