fix(upsert): early rejection of unsupported join column types

mkdocs/docs/api.md

@@ -517,6 +517,21 @@ long: [[2.349014],[4.896029],[6.0989],[-122.431297]]
 
 PyIceberg supports upsert operations, meaning that it is able to merge an Arrow table into an Iceberg table. Rows are considered the same based on the [identifier field](https://iceberg.apache.org/spec/?column-projection#identifier-field-ids). If a row is already in the table, it will update that row. If a row cannot be found, it will insert that new row.
 
+<!-- prettier-ignore-start -->
+
+<!-- markdownlint-disable MD046 -- Allowing indented multi-line formatting in admonition-->
+
+!!! note "Join columns"
+    Use `join_cols` to pick the columns to match on; when omitted, the table's identifier fields are used. Unsupported join columns are rejected with a descriptive error before anything is written.
+
+    - **Supported**: boolean, integer, long, decimal, date, time, timestamp, string, and binary columns.
+    - **Not supported**: `float` and `double` columns, because floating-point equality is unreliable; nested columns (structs, lists, and maps), including identifier fields nested inside a struct; and UUID columns.
+    - **Input requirements**: each join column must be present, must not contain null values, and the input rows must be unique on the join columns. Dictionary-encoded, `string_view`, `binary_view`, extension-type, and `pa.null()` columns must be cast or decoded to a plain type first.
+
+<!-- markdownlint-enable MD046 -->
+
+<!-- prettier-ignore-end -->
+
 Consider the following table, with some data:
 
 ```python
@@ -1081,11 +1096,16 @@ Expert Iceberg users may choose to commit existing parquet files to the Iceberg
 
 <!-- prettier-ignore-start -->
 
+<!-- markdownlint-disable MD046 -- Allowing indented multi-line formatting in admonition-->
+
 !!! note "Name Mapping and Field IDs"
     `add_files` can work with Parquet files both with and without field IDs in their metadata:
+
     - **Files with field IDs**: When field IDs are present in the Parquet metadata, they must match the corresponding field IDs in the Iceberg table schema. This is common for files generated by tools like Spark or when using or other libraries with explicit field ID metadata.
     - **Files without field IDs**: When field IDs are absent, the table must have a [Name Mapping](https://iceberg.apache.org/spec/?h=name+mapping#name-mapping-serialization) to map field names to Iceberg field IDs. `add_files` will automatically create a Name Mapping based on the table's current schema if one doesn't already exist.
 
+<!-- markdownlint-enable MD046 -->
+
 !!! note "Partitions"
     `add_files` only requires the client to read the existing parquet files' metadata footer to infer the partition value of each file. This implementation also supports adding files to Iceberg tables with partition transforms like `MonthTransform`, and `TruncateTransform` which preserve the order of the values after the transformation (Any Transform that has the `preserves_order` property set to True is supported). Please note that if the column statistics of the `PartitionField`'s source column are not present in the parquet metadata, the partition value is inferred as `None`.
 

pyiceberg/table/__init__.py

@@ -914,10 +914,10 @@ def upsert(
         if not when_matched_update_all and not when_not_matched_insert_all:
             raise ValueError("no upsert options selected...exiting")
 
-        if upsert_util.has_duplicate_rows(df, join_cols):
-            raise ValueError("Duplicate rows found in source dataset based on the key columns. No upsert executed")
+        from pyiceberg.io.pyarrow import _check_pyarrow_schema_compatible, schema_to_pyarrow
 
-        from pyiceberg.io.pyarrow import _check_pyarrow_schema_compatible
+        table_arrow_schema = schema_to_pyarrow(self.table_metadata.schema(), include_field_ids=False)
+        upsert_util.validate_join_cols(df, join_cols, table_arrow_schema)
 
         downcast_ns_timestamp_to_us = Config().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False
         _check_pyarrow_schema_compatible(
@@ -927,6 +927,10 @@ def upsert(
             format_version=self.table_metadata.format_version,
         )
 
+        # Validate uniqueness after type checks to avoid comparing/hashing unsupported types.
+        if upsert_util.has_duplicate_rows(df, join_cols):
+            raise ValueError("Duplicate rows found in source dataset based on the key columns. No upsert executed")
+
         # get list of rows that exist so we don't have to load the entire target table
         matched_predicate = upsert_util.create_match_filter(df, join_cols)
 

pyiceberg/table/upsert_util.py

@@ -16,6 +16,7 @@
 # under the License.
 import functools
 import operator
+from collections import Counter
 
 import pyarrow as pa
 from pyarrow import Table as pyarrow_table
@@ -30,6 +31,75 @@
 )
 
 
+def validate_join_cols(df: pyarrow_table, join_cols: list[str], table_schema: pa.Schema) -> None:
+    """Validate join-key presence and types before Arrow comparison or hashing."""
+    if not isinstance(join_cols, (list, tuple)):
+        raise ValueError(f"join_cols must be a list of column names, got {type(join_cols).__name__}.")
+
+    duplicates = sorted(col for col, count in Counter(join_cols).items() if count > 1)
+    if duplicates:
+        raise ValueError(f"join_cols contains duplicates: {', '.join(duplicates)}.")
+
+    for col in join_cols:
+        _validate_table_join_col(col, table_schema)
+        _validate_input_join_col(col, df)
+
+
+def _validate_table_join_col(col: str, table_schema: pa.Schema) -> None:
+    """Reject types that are unreliable or unsupported as join keys regardless of the input."""
+    if col not in table_schema.names:
+        parent = col.split(".", 1)[0]
+        if "." in col and parent in table_schema.names and pa.types.is_nested(table_schema.field(parent).type):
+            raise ValueError(f"Join column '{col}' is a field inside struct '{parent}' and cannot be used as a join key.")
+        raise ValueError(f"Join column '{col}' does not exist in the table. Available columns: {', '.join(table_schema.names)}.")
+
+    field_type = table_schema.field(col).type
+
+    if pa.types.is_floating(field_type):
+        raise ValueError(
+            f"Join column '{col}' is floating point and cannot be used as a join key "
+            "because floating point equality is unreliable."
+        )
+
+    if pa.types.is_nested(field_type):
+        raise ValueError(f"Join column '{col}' has nested type '{field_type}'; only primitive columns can be join keys.")
+
+    if isinstance(field_type, pa.BaseExtensionType):
+        raise NotImplementedError(f"Join column '{col}' has type '{field_type}', which is not yet supported as a join key.")
+
+
+def _validate_input_join_col(col: str, df: pyarrow_table) -> None:
+    """Reject input representations that are unsupported even when the table type is valid."""
+    # Schema compatibility permits missing optional fields, but upsert needs every join key.
+    if col not in df.schema.names:
+        raise ValueError(f"Join column '{col}' is missing from the input.")
+
+    arr = df.column(col)
+
+    if pa.types.is_dictionary(arr.type):
+        raise NotImplementedError(
+            f"Input column '{col}' is dictionary-encoded, which is not yet supported for join keys. Decode it first."
+        )
+
+    if pa.types.is_null(arr.type):
+        raise ValueError(f"Input column '{col}' has the null type and cannot be used as a join key.")
+
+    if pa.types.is_string_view(arr.type) or pa.types.is_binary_view(arr.type):
+        plain = "string" if pa.types.is_string_view(arr.type) else "binary"
+        raise NotImplementedError(
+            f"Input column '{col}' has type '{arr.type}', which is not yet supported for join keys. Cast it to '{plain}' first."
+        )
+
+    if isinstance(arr.type, pa.BaseExtensionType):
+        raise NotImplementedError(
+            f"Input column '{col}' has extension type '{arr.type}', which is not yet supported for join keys."
+        )
+
+    # Null keys cannot be expressed as Iceberg literals in the match filter.
+    if arr.null_count > 0:
+        raise ValueError(f"Input column '{col}' contains null values, which cannot be join keys.")
+
+
 def create_match_filter(df: pyarrow_table, join_cols: list[str]) -> BooleanExpression:
     unique_keys = df.select(join_cols).group_by(join_cols).aggregate([])