draft(upsert): revise exact match filter cleanup

pyiceberg/table/__init__.py

@@ -861,6 +861,34 @@ def upsert(
         if not when_matched_update_all and not when_not_matched_insert_all:
             raise ValueError("no upsert options selected...exiting")
 
+        if not case_sensitive:
+            all_names = [field.name for field in self.table_metadata.schema().fields]
+            lower_names = [n.lower() for n in all_names]
+            if len(set(lower_names)) != len(all_names):
+                import collections
+
+                collisions = [item for item, count in collections.Counter(lower_names).items() if count > 1]
+                raise ValueError(
+                    f"Case-insensitive upsert is ambiguous for this table because of name collisions: {', '.join(collisions)}. "
+                    "Please use case_sensitive=True or rename the columns."
+                )
+
+        # Ensure all top-level table columns are present in the source to avoid silent data loss.
+        if case_sensitive:
+            table_cols = {field.name for field in self.table_metadata.schema().fields}
+            source_cols = set(df.column_names)
+        else:
+            table_cols = {field.name.lower() for field in self.table_metadata.schema().fields}
+            source_cols = {name.lower() for name in df.column_names}
+
+        missing_cols = table_cols - source_cols
+        if missing_cols:
+            raise ValueError(
+                f"Partial schema updates are not yet supported. The source dataframe is missing "
+                f"the following table columns: {', '.join(sorted(missing_cols))}. "
+                "Please provide all columns to avoid accidental data loss."
+            )
+
         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")
 
@@ -917,8 +945,9 @@ def upsert(
                 expr_match_bound = bind(self.table_metadata.schema(), expr_match, case_sensitive=case_sensitive)
                 expr_match_arrow = expression_to_pyarrow(expr_match_bound)
 
-                # Filter rows per batch.
-                rows_to_insert = rows_to_insert.filter(~expr_match_arrow)
+                # Filter rows per batch. Treat null match (source key null vs non-null target key)
+                # as "no match" so the row reaches the insert path instead of being dropped.
+                rows_to_insert = rows_to_insert.filter(expr_match_arrow.is_null() | ~expr_match_arrow)
 
         update_row_cnt = 0
         insert_row_cnt = 0

pyiceberg/table/upsert_util.py

@@ -61,21 +61,59 @@ def create_file_match_filter(df: pyarrow_table, join_cols: list[str]) -> Boolean
 
 
 def create_match_filter(df: pyarrow_table, join_cols: list[str]) -> BooleanExpression:
+    """Exact predicate over the source join keys; null in a key compares with IsNull."""
     unique_keys = df.select(join_cols).group_by(join_cols).aggregate([])
+    if len(unique_keys) == 0:
+        return AlwaysFalse()
 
     if len(join_cols) == 1:
-        return In(join_cols[0], unique_keys[0].to_pylist())
-    else:
-        filters = [
-            functools.reduce(operator.and_, [EqualTo(col, row[col]) for col in join_cols]) for row in unique_keys.to_pylist()
-        ]
-
-        if len(filters) == 0:
-            return AlwaysFalse()
-        elif len(filters) == 1:
-            return filters[0]
-        else:
-            return Or(*filters)
+        col = join_cols[0]
+        vals = unique_keys[0].to_pylist()
+        non_null = [v for v in vals if v is not None]
+        if not non_null:
+            return IsNull(col)
+        in_pred: BooleanExpression = In(col, non_null)
+        return in_pred if len(non_null) == len(vals) else in_pred | IsNull(col)
+
+    row_preds = [
+        functools.reduce(
+            operator.and_,
+            [EqualTo(c, row[c]) if row[c] is not None else IsNull(c) for c in join_cols],
+        )
+        for row in unique_keys.to_pylist()
+    ]
+    return row_preds[0] if len(row_preds) == 1 else Or(*row_preds)
+
+
+def _default_scalar(arrow_type: pa.DataType) -> pa.Scalar:
+    """Return a fixed non-null scalar of the given type for use as a null sentinel."""
+    if pa.types.is_string(arrow_type) or pa.types.is_large_string(arrow_type):
+        return pa.scalar("", type=arrow_type)
+    if pa.types.is_binary(arrow_type) or pa.types.is_large_binary(arrow_type):
+        return pa.scalar(b"", type=arrow_type)
+    if pa.types.is_fixed_size_binary(arrow_type):
+        return pa.scalar(b"\x00" * arrow_type.byte_width, type=arrow_type)
+    if pa.types.is_boolean(arrow_type):
+        return pa.scalar(False, type=arrow_type)
+    return pa.scalar(0, type=arrow_type)
+
+
+def _augment_for_null_safe_join(table: pa.Table, join_cols: set[str]) -> pa.Table:
+    """Augment join columns so pyarrow inner join treats null↔null as a match.
+
+    Replaces nulls in each join col with a fixed same-type sentinel and appends an
+    `__isnull_<col>` indicator. Joining on (col, indicator) then matches null rows to
+    null rows. The sentinel is type-derived (not data-derived) so both sides agree.
+    """
+    for col in join_cols:
+        if f"__isnull_{col}" in join_cols:
+            raise ValueError(f"join column '__isnull_{col}' collides with the reserved null-indicator name")
+    out = table
+    for col in join_cols:
+        arr = table.column(col)
+        out = out.set_column(out.column_names.index(col), col, pc.fill_null(arr, _default_scalar(arr.type)))
+        out = out.append_column(f"__isnull_{col}", pc.is_null(arr))
+    return out
 
 
 def has_duplicate_rows(df: pyarrow_table, join_cols: list[str]) -> bool:
@@ -115,22 +153,26 @@ def get_rows_to_update(source_table: pa.Table, target_table: pa.Table, join_cols
             f"DataFrames, and cannot be used as column names"
         ) from None
 
-    # Step 1: Prepare source index with join keys and a marker index
-    # Cast to target table schema, so we can do the join
+    # Step 1: Prepare source index with join keys and a marker index.
+    # Select and cast using the list order to ensure schema alignment with target_key_schema.
     # See: https://github.com/apache/arrow/issues/37542
-    source_index = (
-        source_table.cast(target_table.schema)
-        .select(join_cols_set)
-        .append_column(SOURCE_INDEX_COLUMN_NAME, pa.array(range(len(source_table))))
+    target_key_schema = pa.schema([target_table.schema.field(col) for col in join_cols])
+    source_index = _augment_for_null_safe_join(
+        source_table.select(join_cols).cast(target_key_schema), join_cols_set
+    ).append_column(SOURCE_INDEX_COLUMN_NAME, pa.array(range(len(source_table))))
+
+    # Step 2: Prepare target index with join keys and a marker.
+    target_index = _augment_for_null_safe_join(target_table.select(join_cols_set), join_cols_set).append_column(
+        TARGET_INDEX_COLUMN_NAME, pa.array(range(len(target_table)))
     )
 
-    # Step 2: Prepare target index with join keys and a marker
-    target_index = target_table.select(join_cols_set).append_column(TARGET_INDEX_COLUMN_NAME, pa.array(range(len(target_table))))
-
-    # Step 3: Perform an inner join to find which rows from source exist in target
-    matching_indices = source_index.join(target_index, keys=list(join_cols_set), join_type="inner")
+    # Step 3: Inner join on (key value, is-null indicator) per col — matches null↔null.
+    join_keys = list(join_cols_set) + [f"__isnull_{c}" for c in join_cols_set]
+    matching_indices = source_index.join(target_index, keys=join_keys, join_type="inner")
 
-    # Step 4: Compare all rows using Python
+    # Step 4: Compare rows in Python to find actual changes.
+    # We use Python here because PyArrow (as of v18) cannot perform vectorized
+    # equality checks on nested types (structs/lists).
     to_update_indices = []
     for source_idx, target_idx in zip(
         matching_indices[SOURCE_INDEX_COLUMN_NAME].to_pylist(),
@@ -147,7 +189,7 @@ def get_rows_to_update(source_table: pa.Table, target_table: pa.Table, join_cols
                 to_update_indices.append(source_idx)
                 break
 
-    # Step 5: Take rows from source table using the indices and cast to target schema
+    # Step 5: Take rows from source table using the indices and return as a Table.
     if to_update_indices:
         return source_table.take(to_update_indices)
     else: