From 2abd44fa24d92b8e49a47e11599bd36329398f4a Mon Sep 17 00:00:00 2001 From: Abanoub Doss Date: Tue, 19 May 2026 14:51:11 -0500 Subject: [PATCH 01/11] refactor(upsert): early rejection of unsupported join column types --- pyiceberg/table/__init__.py | 39 +++++++++++++++++++++++++++++++++---- tests/table/test_upsert.py | 37 ++++++++++++++++++++++++++++++++++- 2 files changed, 71 insertions(+), 5 deletions(-) diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index b8d87143c9..5fbeb2248d 100644 --- a/pyiceberg/table/__init__.py +++ b/pyiceberg/table/__init__.py @@ -782,10 +782,7 @@ 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 + from pyiceberg.io.pyarrow import _check_pyarrow_schema_compatible, schema_to_pyarrow downcast_ns_timestamp_to_us = Config().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False _check_pyarrow_schema_compatible( @@ -795,6 +792,40 @@ def upsert( format_version=self.table_metadata.format_version, ) + table_arrow_schema = schema_to_pyarrow(self.table_metadata.schema(), include_field_ids=False) + + for col in join_cols: + table_field = table_arrow_schema.field(col) + # Table-level rejections: These types are fundamentally unreliable or + # unsupported as join keys regardless of the input data format. + if pa.types.is_floating(table_field.type): + raise ValueError( + f"Floating point column '{col}' cannot be used as a join key in upsert. " + "Floating point equality is unreliable; please cast to Decimal or Integer." + ) + if pa.types.is_nested(table_field.type): + raise ValueError( + f"Nested column '{col}' of type '{table_field.type}' cannot be used as a join key in upsert. " + "Only primitive types are supported." + ) + + # Dataframe-level rejections: These implementation-specific formats (e.g., + # dictionary encoding) are not yet supported by the PyArrow join engine. + arr = df.column(col) + if pa.types.is_dictionary(arr.type): + raise NotImplementedError( + f"Dictionary-encoded column '{col}' is not currently supported as a join key in upsert." + ) + if pa.types.is_null(arr.type): + raise ValueError(f"Null-type column '{col}' cannot be used as a join key in upsert.") + if isinstance(arr.type, pa.BaseExtensionType): + raise NotImplementedError( + f"Extension type '{arr.type}' for column '{col}' is not currently supported as a join key in upsert." + ) + + 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) diff --git a/tests/table/test_upsert.py b/tests/table/test_upsert.py index 08f90c6600..50b1647ba7 100644 --- a/tests/table/test_upsert.py +++ b/tests/table/test_upsert.py @@ -666,7 +666,7 @@ def test_upsert_with_struct_field_as_join_key(catalog: Catalog) -> None: ) with pytest.raises( - pa.lib.ArrowNotImplementedError, match="Keys of type struct" + ValueError, match="Nested column 'nested_type' of type 'struct' cannot be used as a join key in upsert" ): _ = tbl.upsert(update_data, join_cols=["nested_type"]) @@ -888,3 +888,38 @@ def test_upsert_snapshot_properties(catalog: Catalog) -> None: for snapshot in snapshots[initial_snapshot_count:]: assert snapshot.summary is not None assert snapshot.summary.additional_properties.get("test_prop") == "test_value" + +@pytest.mark.parametrize( + "arrow_type, expected_error, match", + [ + (pa.float32(), ValueError, "Floating point column 'k' cannot be used as a join key in upsert"), + (pa.float64(), ValueError, "Floating point column 'k' cannot be used as a join key in upsert"), + (pa.struct([("a", pa.int32())]), ValueError, "Nested column 'k' of type 'struct' cannot be used as a join key in upsert"), + (pa.list_(pa.int32()), ValueError, "Nested column 'k' of type 'list' cannot be used as a join key in upsert"), + (pa.dictionary(pa.int32(), pa.string()), NotImplementedError, "Dictionary-encoded column 'k' is not currently supported as a join key in upsert"), + (pa.null(), ValueError, "Null-type column 'k' cannot be used as a join key in upsert"), + (pa.uuid(), NotImplementedError, "is not currently supported as a join key in upsert"), + ], + ids=["float32", "float64", "struct", "list", "dictionary", "null", "uuid"], +) +def test_upsert_unsupported_join_column_types( + catalog: Catalog, arrow_type: pa.DataType, expected_error: type[Exception], match: str +) -> None: + """Upsert must clearly reject types that are unreliable (floats) or unsupported (extensions/complex) as join keys.""" + identifier = "default.test_upsert_unsupported_join_column_types" + try: + catalog.drop_table(identifier) + except NoSuchTableError: + pass + + # Create a simple table with a valid schema + table = catalog.create_table(identifier, pa.schema([("id", pa.int32()), ("payload", pa.string())])) + + # Source has the "bad" type + source = pa.Table.from_pylist( + [{"k": None, "payload": "val"}], + schema=pa.schema([("k", arrow_type), ("payload", pa.string())]), + ) + + with pytest.raises(expected_error, match=match): + table.upsert(source, join_cols=["k"]) From 12d978e24054ac27717b5d1945aa2f7dd3eb1ea3 Mon Sep 17 00:00:00 2001 From: Abanoub Doss Date: Tue, 19 May 2026 14:59:51 -0500 Subject: [PATCH 02/11] test(upsert): align test schemas with early compatibility check and update error expectations --- tests/table/test_upsert.py | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/tests/table/test_upsert.py b/tests/table/test_upsert.py index 50b1647ba7..36d7d61040 100644 --- a/tests/table/test_upsert.py +++ b/tests/table/test_upsert.py @@ -323,7 +323,7 @@ def test_key_cols_misaligned(catalog: Catalog) -> None: df_src = ctx.sql("select 1 as item_id, date '2021-05-01' as order_date, 'B' as order_type").to_arrow_table() - with pytest.raises(Exception, match=r"""Field ".*" does not exist in schema"""): + with pytest.raises(ValueError, match="PyArrow table contains more columns: item_id"): table.upsert(df=df_src, join_cols=["order_id"]) @@ -912,14 +912,21 @@ def test_upsert_unsupported_join_column_types( except NoSuchTableError: pass - # Create a simple table with a valid schema - table = catalog.create_table(identifier, pa.schema([("id", pa.int32()), ("payload", pa.string())])) + # Define the table schema to be compatible with the arrow_type but still trigger our check + if pa.types.is_dictionary(arrow_type): + table_type = pa.string() + elif pa.types.is_null(arrow_type): + table_type = pa.int32() + else: + table_type = arrow_type + + table = catalog.create_table(identifier, pa.schema([("k", table_type), ("payload", pa.string())])) # Source has the "bad" type source = pa.Table.from_pylist( [{"k": None, "payload": "val"}], schema=pa.schema([("k", arrow_type), ("payload", pa.string())]), ) - + with pytest.raises(expected_error, match=match): table.upsert(source, join_cols=["k"]) From 336dcf8c46891ab75341f8c397646fa34a12e0d9 Mon Sep 17 00:00:00 2001 From: Abanoub Doss Date: Tue, 19 May 2026 15:00:09 -0500 Subject: [PATCH 03/11] docs(upsert): add rationale for duplicate check ordering --- pyiceberg/table/__init__.py | 1 + 1 file changed, 1 insertion(+) diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index 5fbeb2248d..ef516fe4ab 100644 --- a/pyiceberg/table/__init__.py +++ b/pyiceberg/table/__init__.py @@ -823,6 +823,7 @@ def upsert( f"Extension type '{arr.type}' for column '{col}' is not currently supported as a join key in upsert." ) + # 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") From c97f724b16c37445156cae5b389e7ad8e9202c37 Mon Sep 17 00:00:00 2001 From: Abanoub Doss Date: Tue, 19 May 2026 15:13:34 -0500 Subject: [PATCH 04/11] style: fix linting issues (line length and whitespace) --- pyiceberg/table/__init__.py | 4 ++-- tests/table/test_upsert.py | 27 ++++++++++++++++++++++----- 2 files changed, 24 insertions(+), 7 deletions(-) diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index ef516fe4ab..8cf9539a07 100644 --- a/pyiceberg/table/__init__.py +++ b/pyiceberg/table/__init__.py @@ -796,7 +796,7 @@ def upsert( for col in join_cols: table_field = table_arrow_schema.field(col) - # Table-level rejections: These types are fundamentally unreliable or + # Table-level rejections: These types are fundamentally unreliable or # unsupported as join keys regardless of the input data format. if pa.types.is_floating(table_field.type): raise ValueError( @@ -809,7 +809,7 @@ def upsert( "Only primitive types are supported." ) - # Dataframe-level rejections: These implementation-specific formats (e.g., + # Dataframe-level rejections: These implementation-specific formats (e.g., # dictionary encoding) are not yet supported by the PyArrow join engine. arr = df.column(col) if pa.types.is_dictionary(arr.type): diff --git a/tests/table/test_upsert.py b/tests/table/test_upsert.py index 36d7d61040..e63482db53 100644 --- a/tests/table/test_upsert.py +++ b/tests/table/test_upsert.py @@ -666,7 +666,11 @@ def test_upsert_with_struct_field_as_join_key(catalog: Catalog) -> None: ) with pytest.raises( - ValueError, match="Nested column 'nested_type' of type 'struct' cannot be used as a join key in upsert" + ValueError, + match=( + "Nested column 'nested_type' of type 'struct' " + "cannot be used as a join key in upsert" + ), ): _ = tbl.upsert(update_data, join_cols=["nested_type"]) @@ -889,14 +893,27 @@ def test_upsert_snapshot_properties(catalog: Catalog) -> None: assert snapshot.summary is not None assert snapshot.summary.additional_properties.get("test_prop") == "test_value" + @pytest.mark.parametrize( "arrow_type, expected_error, match", [ (pa.float32(), ValueError, "Floating point column 'k' cannot be used as a join key in upsert"), (pa.float64(), ValueError, "Floating point column 'k' cannot be used as a join key in upsert"), - (pa.struct([("a", pa.int32())]), ValueError, "Nested column 'k' of type 'struct' cannot be used as a join key in upsert"), - (pa.list_(pa.int32()), ValueError, "Nested column 'k' of type 'list' cannot be used as a join key in upsert"), - (pa.dictionary(pa.int32(), pa.string()), NotImplementedError, "Dictionary-encoded column 'k' is not currently supported as a join key in upsert"), + ( + pa.struct([("a", pa.int32())]), + ValueError, + "Nested column 'k' of type 'struct' cannot be used as a join key in upsert", + ), + ( + pa.list_(pa.int32()), + ValueError, + "Nested column 'k' of type 'list' cannot be used as a join key in upsert", + ), + ( + pa.dictionary(pa.int32(), pa.string()), + NotImplementedError, + "Dictionary-encoded column 'k' is not currently supported as a join key in upsert", + ), (pa.null(), ValueError, "Null-type column 'k' cannot be used as a join key in upsert"), (pa.uuid(), NotImplementedError, "is not currently supported as a join key in upsert"), ], @@ -911,7 +928,7 @@ def test_upsert_unsupported_join_column_types( catalog.drop_table(identifier) except NoSuchTableError: pass - + # Define the table schema to be compatible with the arrow_type but still trigger our check if pa.types.is_dictionary(arrow_type): table_type = pa.string() From fe60c9790368e03e38237051119d14d459012137 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 21 May 2026 00:11:21 +0000 Subject: [PATCH 05/11] test(upsert): expect large_list in unsupported-type error for list join key The error message renders the type from the Iceberg table's pyarrow schema, and schema_to_pyarrow converts pa.list_ into pa.large_list (see pyiceberg/io/pyarrow.py). The test regex must match the rendered large_list, not the source list. --- tests/table/test_upsert.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/table/test_upsert.py b/tests/table/test_upsert.py index e63482db53..175146aa21 100644 --- a/tests/table/test_upsert.py +++ b/tests/table/test_upsert.py @@ -907,7 +907,7 @@ def test_upsert_snapshot_properties(catalog: Catalog) -> None: ( pa.list_(pa.int32()), ValueError, - "Nested column 'k' of type 'list' cannot be used as a join key in upsert", + "Nested column 'k' of type 'large_list' cannot be used as a join key in upsert", ), ( pa.dictionary(pa.int32(), pa.string()), From 9928f50d80e310a886c138e6882b00328e6c1e94 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 21 May 2026 00:30:43 +0000 Subject: [PATCH 06/11] fix(upsert): run join-column type rejection before schema compat check A pa.null() source column was being rejected by _check_pyarrow_schema_compatible (format-version=2 forbids null) before the join-column validation could surface the intended "Null-type column ... cannot be used as a join key" error. Reordering the checks lets the upsert-specific rejection fire first, giving users the actionable message. Dataframe-level checks now skip columns that are absent from the source so the pre-existing _check_pyarrow_schema_compatible path still owns the "PyArrow table contains more columns" error in test_key_cols_misaligned. --- pyiceberg/table/__init__.py | 23 +++++++++++++---------- 1 file changed, 13 insertions(+), 10 deletions(-) diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index 8cf9539a07..7fa3e8d117 100644 --- a/pyiceberg/table/__init__.py +++ b/pyiceberg/table/__init__.py @@ -784,15 +784,8 @@ def upsert( from pyiceberg.io.pyarrow import _check_pyarrow_schema_compatible, schema_to_pyarrow - downcast_ns_timestamp_to_us = Config().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False - _check_pyarrow_schema_compatible( - self.table_metadata.schema(), - provided_schema=df.schema, - downcast_ns_timestamp_to_us=downcast_ns_timestamp_to_us, - format_version=self.table_metadata.format_version, - ) - table_arrow_schema = schema_to_pyarrow(self.table_metadata.schema(), include_field_ids=False) + df_column_names = set(df.schema.names) for col in join_cols: table_field = table_arrow_schema.field(col) @@ -809,8 +802,10 @@ def upsert( "Only primitive types are supported." ) - # Dataframe-level rejections: These implementation-specific formats (e.g., - # dictionary encoding) are not yet supported by the PyArrow join engine. + # Dataframe-level rejections: only validate when the column is present in the + # source; missing columns are surfaced by _check_pyarrow_schema_compatible below. + if col not in df_column_names: + continue arr = df.column(col) if pa.types.is_dictionary(arr.type): raise NotImplementedError( @@ -823,6 +818,14 @@ def upsert( f"Extension type '{arr.type}' for column '{col}' is not currently supported as a join key in upsert." ) + downcast_ns_timestamp_to_us = Config().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False + _check_pyarrow_schema_compatible( + self.table_metadata.schema(), + provided_schema=df.schema, + downcast_ns_timestamp_to_us=downcast_ns_timestamp_to_us, + 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") From 984057e71805890ba3f7ff0f9458961ce2bfd3d9 Mon Sep 17 00:00:00 2001 From: Abanoub Doss Date: Wed, 10 Jun 2026 08:04:13 -0500 Subject: [PATCH 07/11] fix(upsert): add checks for missing table columns before type validations. --- pyiceberg/table/__init__.py | 5 +++++ tests/table/test_upsert.py | 19 +++++++++++++++++++ 2 files changed, 24 insertions(+) diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index 7fa3e8d117..c6c4be9b05 100644 --- a/pyiceberg/table/__init__.py +++ b/pyiceberg/table/__init__.py @@ -788,6 +788,11 @@ def upsert( df_column_names = set(df.schema.names) for col in join_cols: + if col not in table_arrow_schema.names: + raise ValueError( + f"Join column '{col}' does not exist in the table schema. " + f"Available columns: {', '.join(table_arrow_schema.names)}." + ) table_field = table_arrow_schema.field(col) # Table-level rejections: These types are fundamentally unreliable or # unsupported as join keys regardless of the input data format. diff --git a/tests/table/test_upsert.py b/tests/table/test_upsert.py index 175146aa21..92506d82c4 100644 --- a/tests/table/test_upsert.py +++ b/tests/table/test_upsert.py @@ -327,6 +327,25 @@ def test_key_cols_misaligned(catalog: Catalog) -> None: table.upsert(df=df_src, join_cols=["order_id"]) +def test_key_cols_source_only_join_col(catalog: Catalog) -> None: + """ + tests join column present in the source dataframe but missing from the table + """ + + identifier = "default.test_key_cols_source_only_join_col" + _drop_table(catalog, identifier) + + ctx = SessionContext() + + df = ctx.sql("select 1 as order_id, date '2021-01-01' as order_date, 'A' as order_type").to_arrow_table() + table = catalog.create_table(identifier, df.schema) + + df_src = ctx.sql("select 1 as order_id, 10 as item_id, date '2021-05-01' as order_date, 'B' as order_type").to_arrow_table() + + with pytest.raises(ValueError, match="Join column 'item_id' does not exist in the table schema"): + table.upsert(df=df_src, join_cols=["item_id"]) + + def test_upsert_with_identifier_fields(catalog: Catalog) -> None: identifier = "default.test_upsert_with_identifier_fields" _drop_table(catalog, identifier) From 60d240ae664aa195aa06bc39fd153473df919711 Mon Sep 17 00:00:00 2001 From: Abanoub Doss Date: Tue, 15 Sep 2026 17:29:02 -0500 Subject: [PATCH 08/11] fix(upsert): validate join columns up front Move join column validation from Transaction.upsert into upsert_util.validate_join_cols and reject every key that previously reached PyArrow and crashed: columns missing from the source, UUID and view types, null values, duplicate names. Add test matrices for rejected and accepted keys and document them. --- mkdocs/docs/api.md | 15 ++ pyiceberg/table/__init__.py | 38 +---- pyiceberg/table/upsert_util.py | 57 +++++++ tests/table/test_upsert.py | 283 ++++++++++++++++++++++++++------- 4 files changed, 298 insertions(+), 95 deletions(-) diff --git a/mkdocs/docs/api.md b/mkdocs/docs/api.md index 29d09e2604..c755edbcf9 100644 --- a/mkdocs/docs/api.md +++ b/mkdocs/docs/api.md @@ -506,6 +506,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. + + + + +!!! note "Join columns" + Use `join_cols` to pick the top-level 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. + + + + + Consider the following table, with some data: ```python diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index c6c4be9b05..84810f9071 100644 --- a/pyiceberg/table/__init__.py +++ b/pyiceberg/table/__init__.py @@ -785,43 +785,7 @@ def upsert( from pyiceberg.io.pyarrow import _check_pyarrow_schema_compatible, schema_to_pyarrow table_arrow_schema = schema_to_pyarrow(self.table_metadata.schema(), include_field_ids=False) - df_column_names = set(df.schema.names) - - for col in join_cols: - if col not in table_arrow_schema.names: - raise ValueError( - f"Join column '{col}' does not exist in the table schema. " - f"Available columns: {', '.join(table_arrow_schema.names)}." - ) - table_field = table_arrow_schema.field(col) - # Table-level rejections: These types are fundamentally unreliable or - # unsupported as join keys regardless of the input data format. - if pa.types.is_floating(table_field.type): - raise ValueError( - f"Floating point column '{col}' cannot be used as a join key in upsert. " - "Floating point equality is unreliable; please cast to Decimal or Integer." - ) - if pa.types.is_nested(table_field.type): - raise ValueError( - f"Nested column '{col}' of type '{table_field.type}' cannot be used as a join key in upsert. " - "Only primitive types are supported." - ) - - # Dataframe-level rejections: only validate when the column is present in the - # source; missing columns are surfaced by _check_pyarrow_schema_compatible below. - if col not in df_column_names: - continue - arr = df.column(col) - if pa.types.is_dictionary(arr.type): - raise NotImplementedError( - f"Dictionary-encoded column '{col}' is not currently supported as a join key in upsert." - ) - if pa.types.is_null(arr.type): - raise ValueError(f"Null-type column '{col}' cannot be used as a join key in upsert.") - if isinstance(arr.type, pa.BaseExtensionType): - raise NotImplementedError( - f"Extension type '{arr.type}' for column '{col}' is not currently supported as a join key in upsert." - ) + 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( diff --git a/pyiceberg/table/upsert_util.py b/pyiceberg/table/upsert_util.py index 6f32826eb0..d060a87ee3 100644 --- a/pyiceberg/table/upsert_util.py +++ b/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,62 @@ ) +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"Duplicate join columns: {', '.join(duplicates)}.") + + df_column_names = set(df.schema.names) + + for col in join_cols: + if col not in table_schema.names: + raise ValueError( + f"Join column '{col}' does not exist in the table schema. Only top-level columns can be used as join keys. " + f"Available columns: {', '.join(table_schema.names)}." + ) + table_field = table_schema.field(col) + # Table-level rejections: These types are fundamentally unreliable or + # unsupported as join keys regardless of the input data format. + if pa.types.is_floating(table_field.type): + raise ValueError( + f"Floating point column '{col}' cannot be used as a join key in upsert. " + "Floating point equality is unreliable; choose a different join column." + ) + if pa.types.is_nested(table_field.type): + raise ValueError( + f"Nested column '{col}' of type '{table_field.type}' cannot be used as a join key in upsert. " + "Only primitive types are supported." + ) + if isinstance(table_field.type, pa.BaseExtensionType): + raise NotImplementedError( + f"Column '{col}' of type '{table_field.type}' is not currently supported as a join key in upsert." + ) + + # Schema compatibility permits missing optional fields, but upsert needs every join key. + if col not in df_column_names: + raise ValueError(f"Join column '{col}' does not exist in the source schema.") + # Some source representations are unsupported even when the table type is valid. + arr = df.column(col) + if pa.types.is_dictionary(arr.type): + raise NotImplementedError(f"Dictionary-encoded column '{col}' is not currently supported as a join key in upsert.") + if pa.types.is_null(arr.type): + raise ValueError(f"Null-type column '{col}' cannot be used as a join key in upsert.") + if pa.types.is_string_view(arr.type) or pa.types.is_binary_view(arr.type): + raise NotImplementedError( + f"View-typed column '{col}' of type '{arr.type}' is not currently supported as a join key in upsert." + ) + if isinstance(arr.type, pa.BaseExtensionType): + raise NotImplementedError( + f"Extension type '{arr.type}' for column '{col}' is not currently supported as a join key in upsert." + ) + # Null keys cannot be expressed as Iceberg literals in the match filter. + if arr.null_count > 0: + raise ValueError(f"Join column '{col}' contains null values, which cannot be used as join keys in upsert.") + + def create_match_filter(df: pyarrow_table, join_cols: list[str]) -> BooleanExpression: unique_keys = df.select(join_cols).group_by(join_cols).aggregate([]) diff --git a/tests/table/test_upsert.py b/tests/table/test_upsert.py index 92506d82c4..783ff4d8a9 100644 --- a/tests/table/test_upsert.py +++ b/tests/table/test_upsert.py @@ -14,9 +14,14 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. +import uuid +from datetime import date, datetime, time, timezone +from decimal import Decimal from pathlib import PosixPath +from typing import Any import pyarrow as pa +import pyarrow.compute as pc import pytest from datafusion import SessionContext from pyarrow import Table as pa_table @@ -25,12 +30,27 @@ from pyiceberg.exceptions import NoSuchTableError from pyiceberg.expressions import AlwaysTrue, And, EqualTo, Reference from pyiceberg.expressions.literals import LongLiteral -from pyiceberg.io.pyarrow import schema_to_pyarrow +from pyiceberg.io.pyarrow import UnsupportedPyArrowTypeException, schema_to_pyarrow from pyiceberg.schema import Schema from pyiceberg.table import Table, UpsertResult from pyiceberg.table.snapshots import Operation from pyiceberg.table.upsert_util import create_match_filter -from pyiceberg.types import IntegerType, NestedField, StringType, StructType +from pyiceberg.types import ( + BinaryType, + BooleanType, + DateType, + DecimalType, + FixedType, + IntegerType, + LongType, + NestedField, + PrimitiveType, + StringType, + StructType, + TimestampType, + TimestamptzType, + TimeType, +) from tests.catalog.test_base import InMemoryCatalog @@ -323,29 +343,10 @@ def test_key_cols_misaligned(catalog: Catalog) -> None: df_src = ctx.sql("select 1 as item_id, date '2021-05-01' as order_date, 'B' as order_type").to_arrow_table() - with pytest.raises(ValueError, match="PyArrow table contains more columns: item_id"): + with pytest.raises(ValueError, match="Join column 'order_id' does not exist in the source schema"): table.upsert(df=df_src, join_cols=["order_id"]) -def test_key_cols_source_only_join_col(catalog: Catalog) -> None: - """ - tests join column present in the source dataframe but missing from the table - """ - - identifier = "default.test_key_cols_source_only_join_col" - _drop_table(catalog, identifier) - - ctx = SessionContext() - - df = ctx.sql("select 1 as order_id, date '2021-01-01' as order_date, 'A' as order_type").to_arrow_table() - table = catalog.create_table(identifier, df.schema) - - df_src = ctx.sql("select 1 as order_id, 10 as item_id, date '2021-05-01' as order_date, 'B' as order_type").to_arrow_table() - - with pytest.raises(ValueError, match="Join column 'item_id' does not exist in the table schema"): - table.upsert(df=df_src, join_cols=["item_id"]) - - def test_upsert_with_identifier_fields(catalog: Catalog) -> None: identifier = "default.test_upsert_with_identifier_fields" _drop_table(catalog, identifier) @@ -913,56 +914,222 @@ def test_upsert_snapshot_properties(catalog: Catalog) -> None: assert snapshot.summary.additional_properties.get("test_prop") == "test_value" +_UUID_BYTES = uuid.uuid4().bytes +_STRUCT = pa.struct([("a", pa.int32())]) +_MAP = pa.map_(pa.string(), pa.int32()) + + @pytest.mark.parametrize( - "arrow_type, expected_error, match", + "table_type, source_key, expected_error, match", [ - (pa.float32(), ValueError, "Floating point column 'k' cannot be used as a join key in upsert"), - (pa.float64(), ValueError, "Floating point column 'k' cannot be used as a join key in upsert"), - ( - pa.struct([("a", pa.int32())]), - ValueError, - "Nested column 'k' of type 'struct' cannot be used as a join key in upsert", + pytest.param(pa.float32(), pa.array([1.0], pa.float32()), ValueError, "Floating point column 'k'", id="float32"), + pytest.param(pa.float64(), pa.array([1.0], pa.float64()), ValueError, "Floating point column 'k'", id="float64"), + pytest.param( + _STRUCT, pa.array([{"a": 1}], _STRUCT), ValueError, "Nested column 'k' of type 'struct'", id="struct" ), - ( + pytest.param( pa.list_(pa.int32()), + pa.array([[1]], pa.list_(pa.int32())), ValueError, - "Nested column 'k' of type 'large_list' cannot be used as a join key in upsert", + "Nested column 'k' of type 'large_list", + id="list", ), - ( - pa.dictionary(pa.int32(), pa.string()), + pytest.param( + _MAP, pa.array([[("a", 1)]], _MAP), ValueError, "Nested column 'k' of type 'map'", id="map" + ), + pytest.param( + pa.uuid(), + pa.array([_UUID_BYTES], pa.uuid()), NotImplementedError, - "Dictionary-encoded column 'k' is not currently supported as a join key in upsert", + "Column 'k' of type 'extension'", + id="uuid-table", ), - (pa.null(), ValueError, "Null-type column 'k' cannot be used as a join key in upsert"), - (pa.uuid(), NotImplementedError, "is not currently supported as a join key in upsert"), + pytest.param( + pa.uuid(), + pa.array([_UUID_BYTES], pa.binary(16)), + NotImplementedError, + "Column 'k' of type 'extension'", + id="uuid-table-fixed-source", + ), + pytest.param( + pa.string(), + pa.array([_UUID_BYTES], pa.uuid()), + NotImplementedError, + "Extension type 'extension' for column 'k'", + id="uuid-extension-source", + ), + pytest.param( + pa.string(), + pa.array(["a"]).dictionary_encode(), + NotImplementedError, + "Dictionary-encoded column 'k'", + id="dictionary-string", + ), + pytest.param( + pa.int64(), + pa.array([1]).dictionary_encode(), + NotImplementedError, + "Dictionary-encoded column 'k'", + id="dictionary-int", + ), + pytest.param(pa.int32(), pa.array([None], pa.null()), ValueError, "Null-type column 'k'", id="null-type"), + pytest.param( + pa.string(), + pa.array(["a"], pa.string_view()), + NotImplementedError, + "View-typed column 'k' of type 'string_view'", + id="string-view", + ), + pytest.param( + pa.binary(), + pa.array([b"a"], pa.binary_view()), + NotImplementedError, + "View-typed column 'k' of type 'binary_view'", + id="binary-view", + ), + pytest.param( + pa.int32(), + pc.run_end_encode(pa.array([1], pa.int32())), + UnsupportedPyArrowTypeException, + "unsupported type: run_end_encoded", + id="run-end-encoded", + ), + pytest.param( + pa.int32(), pa.array([1, None], pa.int32()), ValueError, "Join column 'k' contains null values", id="null-values" + ), + pytest.param( + pa.int32(), pa.array([None], pa.int32()), ValueError, "Join column 'k' contains null values", id="all-null-values" + ), + pytest.param(pa.int32(), pa.array(["1"]), ValueError, "Mismatch in fields", id="wrong-type-source"), ], - ids=["float32", "float64", "struct", "list", "dictionary", "null", "uuid"], ) -def test_upsert_unsupported_join_column_types( - catalog: Catalog, arrow_type: pa.DataType, expected_error: type[Exception], match: str +def test_upsert_rejects_unsupported_join_key( + catalog: Catalog, table_type: pa.DataType, source_key: pa.Array, expected_error: type[Exception], match: str ) -> None: - """Upsert must clearly reject types that are unreliable (floats) or unsupported (extensions/complex) as join keys.""" - identifier = "default.test_upsert_unsupported_join_column_types" - try: - catalog.drop_table(identifier) - except NoSuchTableError: - pass + """Unreliable or unsupported join keys fail with a descriptive error before anything is written.""" + identifier = "default.test_upsert_rejects_unsupported_join_key" + _drop_table(catalog, identifier) + table = catalog.create_table(identifier, pa.schema([("k", table_type), ("payload", pa.string())])) + source = pa.table({"k": source_key, "payload": ["val"] * len(source_key)}) - # Define the table schema to be compatible with the arrow_type but still trigger our check - if pa.types.is_dictionary(arrow_type): - table_type = pa.string() - elif pa.types.is_null(arrow_type): - table_type = pa.int32() - else: - table_type = arrow_type + with pytest.raises(expected_error, match=match): + table.upsert(source, join_cols=["k"]) + + assert table.current_snapshot() is None - table = catalog.create_table(identifier, pa.schema([("k", table_type), ("payload", pa.string())])) - # Source has the "bad" type +@pytest.mark.parametrize( + "join_cols, identifier_field_ids, drop_source_columns, match", + [ + pytest.param(["missing"], [], [], "Join column 'missing' does not exist in the table schema", id="not-in-table"), + pytest.param(["K"], [], [], "Join column 'K' does not exist in the table schema", id="case-mismatch"), + pytest.param(["k"], [], ["k"], "Join column 'k' does not exist in the source schema", id="required-not-in-source"), + pytest.param(["opt"], [], ["opt"], "Join column 'opt' does not exist in the source schema", id="optional-not-in-source"), + pytest.param( + ["k", "opt"], [], ["opt"], "Join column 'opt' does not exist in the source schema", id="composite-second-missing" + ), + pytest.param(["k", "k"], [], [], "Duplicate join columns: k", id="duplicate-join-cols"), + pytest.param(["s.x"], [], [], "Only top-level columns can be used as join keys", id="nested-path"), + pytest.param(None, [4], [], "Only top-level columns can be used as join keys", id="nested-identifier-field"), + pytest.param([], [], [], "Join columns could not be found", id="empty-join-cols"), + pytest.param(None, [], [], "Join columns could not be found", id="no-identifier-fields"), + pytest.param("k", [], [], "join_cols must be a list of column names, got str", id="string-not-list"), + pytest.param({"k"}, [], [], "join_cols must be a list of column names, got set", id="set-not-list"), + ], +) +def test_upsert_rejects_invalid_join_cols( + catalog: Catalog, join_cols: Any, identifier_field_ids: list[int], drop_source_columns: list[str], match: str +) -> None: + """Join column resolution fails clearly for missing, duplicate, nested, mistyped, or unresolvable columns.""" + identifier = "default.test_upsert_rejects_invalid_join_cols" + _drop_table(catalog, identifier) + schema = Schema( + NestedField(1, "k", IntegerType(), required=True), + NestedField(2, "opt", IntegerType(), required=False), + NestedField(3, "s", StructType(NestedField(4, "x", IntegerType(), required=True)), required=True), + NestedField(5, "payload", StringType(), required=False), + identifier_field_ids=identifier_field_ids, + ) + table = catalog.create_table(identifier, schema) source = pa.Table.from_pylist( - [{"k": None, "payload": "val"}], - schema=pa.schema([("k", arrow_type), ("payload", pa.string())]), + [{"k": 1, "opt": 1, "s": {"x": 1}, "payload": "val"}], + schema=pa.schema( + [ + pa.field("k", pa.int32(), nullable=False), + pa.field("opt", pa.int32(), nullable=True), + pa.field("s", pa.struct([pa.field("x", pa.int32(), nullable=False)]), nullable=False), + pa.field("payload", pa.string(), nullable=True), + ] + ), + ).drop_columns(drop_source_columns) + + with pytest.raises(ValueError, match=match): + table.upsert(source, join_cols=join_cols) + + assert table.current_snapshot() is None + + +_UTC = timezone.utc + + +@pytest.mark.parametrize( + "iceberg_type, arrow_type, existing_key, new_key", + [ + pytest.param(BooleanType(), pa.bool_(), False, True, id="boolean"), + pytest.param(IntegerType(), pa.int32(), 1, 2, id="int"), + pytest.param(LongType(), pa.int64(), 1, 2, id="long"), + pytest.param(LongType(), pa.int32(), 1, 2, id="long-from-int32-source"), + pytest.param(DecimalType(10, 2), pa.decimal128(10, 2), Decimal("1.50"), Decimal("2.50"), id="decimal"), + pytest.param(DateType(), pa.date32(), date(2024, 1, 1), date(2024, 1, 2), id="date"), + pytest.param(TimeType(), pa.time64("us"), time(1, 2, 3), time(4, 5, 6), id="time"), + pytest.param(TimestampType(), pa.timestamp("us"), datetime(2024, 1, 1), datetime(2024, 1, 2), id="timestamp"), + pytest.param( + TimestamptzType(), + pa.timestamp("us", "UTC"), + datetime(2024, 1, 1, tzinfo=_UTC), + datetime(2024, 1, 2, tzinfo=_UTC), + id="timestamptz", + ), + pytest.param(StringType(), pa.string(), "a", "b", id="string"), + pytest.param(StringType(), pa.large_string(), "a", "b", id="string-from-large-string-source"), + pytest.param(BinaryType(), pa.binary(), b"a", b"b", id="binary"), + pytest.param(BinaryType(), pa.large_binary(), b"a", b"b", id="binary-from-large-binary-source"), + pytest.param(FixedType(2), pa.binary(2), b"aa", b"bb", id="fixed"), + ], +) +def test_upsert_accepts_supported_join_key( + catalog: Catalog, iceberg_type: PrimitiveType, arrow_type: pa.DataType, existing_key: Any, new_key: Any +) -> None: + """Every supported primitive key updates matched rows and inserts new ones, including from a multi-chunk source.""" + identifier = "default.test_upsert_accepts_supported_join_key" + _drop_table(catalog, identifier) + schema = Schema(NestedField(1, "k", iceberg_type, required=True), NestedField(2, "payload", StringType(), required=True)) + arrow_schema = pa.schema([pa.field("k", arrow_type, nullable=False), pa.field("payload", pa.string(), nullable=False)]) + table = catalog.create_table(identifier, schema) + table.append(pa.Table.from_pylist([{"k": existing_key, "payload": "old"}], schema=arrow_schema)) + + source = pa.concat_tables( + [ + pa.Table.from_pylist([{"k": existing_key, "payload": "updated"}], schema=arrow_schema), + pa.Table.from_pylist([{"k": new_key, "payload": "inserted"}], schema=arrow_schema), + ] ) + result = table.upsert(source, join_cols=["k"]) - with pytest.raises(expected_error, match=match): - table.upsert(source, join_cols=["k"]) + assert (result.rows_updated, result.rows_inserted) == (1, 1) + rows = table.scan().to_arrow().to_pylist() + assert {(row["k"], row["payload"]) for row in rows} == {(existing_key, "updated"), (new_key, "inserted")} + + +def test_upsert_allows_null_join_key_values_in_target(catalog: Catalog) -> None: + """Null key values are only rejected in the source; existing null-key rows in the table are left untouched.""" + identifier = "default.test_upsert_allows_null_join_key_values_in_target" + _drop_table(catalog, identifier) + table = catalog.create_table(identifier, pa.schema([("k", pa.int32()), ("payload", pa.string())])) + table.append(pa.table({"k": pa.array([None, 1], pa.int32()), "payload": ["orphan", "old"]})) + + result = table.upsert(pa.table({"k": pa.array([1], pa.int32()), "payload": ["new"]}), join_cols=["k"]) + + assert (result.rows_updated, result.rows_inserted) == (1, 0) + rows = table.scan().to_arrow().to_pylist() + assert {(row["k"], row["payload"]) for row in rows} == {(None, "orphan"), (1, "new")} From 530716bf0654ef0319b44e03bfac85253a4bb160 Mon Sep 17 00:00:00 2001 From: Abanoub Doss Date: Tue, 15 Sep 2026 17:29:02 -0500 Subject: [PATCH 09/11] docs: render add_files note bullets as a list --- mkdocs/docs/api.md | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/mkdocs/docs/api.md b/mkdocs/docs/api.md index c755edbcf9..52dbc6218c 100644 --- a/mkdocs/docs/api.md +++ b/mkdocs/docs/api.md @@ -1085,11 +1085,16 @@ Expert Iceberg users may choose to commit existing parquet files to the Iceberg + + !!! 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. + + !!! 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`. From ca979640611cd2bef2f0a50a687d7176eb5b1751 Mon Sep 17 00:00:00 2001 From: Abanoub Doss Date: Tue, 15 Sep 2026 17:53:32 -0500 Subject: [PATCH 10/11] fix(upsert): clarify join column error messages Name the Iceberg table "the table" and the Arrow argument "the input" consistently, point nested paths at the struct they live in instead of saying "top-level", and finish each message with what to do about it. --- mkdocs/docs/api.md | 2 +- pyiceberg/table/upsert_util.py | 35 ++++++++++++---------- tests/table/test_upsert.py | 54 ++++++++++++++++------------------ 3 files changed, 47 insertions(+), 44 deletions(-) diff --git a/mkdocs/docs/api.md b/mkdocs/docs/api.md index d70ca6c661..6e3d35eb35 100644 --- a/mkdocs/docs/api.md +++ b/mkdocs/docs/api.md @@ -522,7 +522,7 @@ PyIceberg supports upsert operations, meaning that it is able to merge an Arrow !!! note "Join columns" - Use `join_cols` to pick the top-level 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. + 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. diff --git a/pyiceberg/table/upsert_util.py b/pyiceberg/table/upsert_util.py index d060a87ee3..2e02013449 100644 --- a/pyiceberg/table/upsert_util.py +++ b/pyiceberg/table/upsert_util.py @@ -37,54 +37,59 @@ def validate_join_cols(df: pyarrow_table, join_cols: list[str], table_schema: pa 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"Duplicate join columns: {', '.join(duplicates)}.") + raise ValueError(f"join_cols contains duplicates: {', '.join(duplicates)}.") df_column_names = set(df.schema.names) for col in join_cols: 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 schema. Only top-level columns can be used as join keys. " - f"Available columns: {', '.join(table_schema.names)}." + f"Join column '{col}' does not exist in the table. Available columns: {', '.join(table_schema.names)}." ) table_field = table_schema.field(col) # Table-level rejections: These types are fundamentally unreliable or # unsupported as join keys regardless of the input data format. if pa.types.is_floating(table_field.type): raise ValueError( - f"Floating point column '{col}' cannot be used as a join key in upsert. " - "Floating point equality is unreliable; choose a different join column." + 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(table_field.type): raise ValueError( - f"Nested column '{col}' of type '{table_field.type}' cannot be used as a join key in upsert. " - "Only primitive types are supported." + f"Join column '{col}' has nested type '{table_field.type}'; only primitive columns can be join keys." ) if isinstance(table_field.type, pa.BaseExtensionType): raise NotImplementedError( - f"Column '{col}' of type '{table_field.type}' is not currently supported as a join key in upsert." + f"Join column '{col}' has type '{table_field.type}', which is not yet supported as a join key." ) # Schema compatibility permits missing optional fields, but upsert needs every join key. if col not in df_column_names: - raise ValueError(f"Join column '{col}' does not exist in the source schema.") - # Some source representations are unsupported even when the table type is valid. + raise ValueError(f"Join column '{col}' is missing from the input.") + # Some input representations are unsupported even when the table type is valid. arr = df.column(col) if pa.types.is_dictionary(arr.type): - raise NotImplementedError(f"Dictionary-encoded column '{col}' is not currently supported as a join key in upsert.") + 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"Null-type column '{col}' cannot be used as a join key in upsert.") + 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"View-typed column '{col}' of type '{arr.type}' is not currently supported as a join key in upsert." + f"Input column '{col}' has type '{arr.type}', which is not yet supported for join keys. " + f"Cast it to '{plain}' first." ) if isinstance(arr.type, pa.BaseExtensionType): raise NotImplementedError( - f"Extension type '{arr.type}' for column '{col}' is not currently supported as a join key in upsert." + 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"Join column '{col}' contains null values, which cannot be used as join keys in upsert.") + 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: diff --git a/tests/table/test_upsert.py b/tests/table/test_upsert.py index a5812ae3bf..7e88189325 100644 --- a/tests/table/test_upsert.py +++ b/tests/table/test_upsert.py @@ -345,7 +345,7 @@ def test_key_cols_misaligned(catalog: Catalog) -> None: df_src = ctx.sql("select 1 as item_id, date '2021-05-01' as order_date, 'B' as order_type").to_arrow_table() - with pytest.raises(ValueError, match="Join column 'order_id' does not exist in the source schema"): + with pytest.raises(ValueError, match="Join column 'order_id' is missing from the input"): table.upsert(df=df_src, join_cols=["order_id"]) @@ -690,8 +690,8 @@ def test_upsert_with_struct_field_as_join_key(catalog: Catalog) -> None: with pytest.raises( ValueError, match=( - "Nested column 'nested_type' of type 'struct' " - "cannot be used as a join key in upsert" + "Join column 'nested_type' has nested type 'struct'; " + "only primitive columns can be join keys" ), ): _ = tbl.upsert(update_data, join_cols=["nested_type"]) @@ -960,69 +960,69 @@ def test_upsert_snapshot_properties(catalog: Catalog) -> None: @pytest.mark.parametrize( "table_type, source_key, expected_error, match", [ - pytest.param(pa.float32(), pa.array([1.0], pa.float32()), ValueError, "Floating point column 'k'", id="float32"), - pytest.param(pa.float64(), pa.array([1.0], pa.float64()), ValueError, "Floating point column 'k'", id="float64"), + pytest.param(pa.float32(), pa.array([1.0], pa.float32()), ValueError, "Join column 'k' is floating point", id="float32"), + pytest.param(pa.float64(), pa.array([1.0], pa.float64()), ValueError, "Join column 'k' is floating point", id="float64"), pytest.param( - _STRUCT, pa.array([{"a": 1}], _STRUCT), ValueError, "Nested column 'k' of type 'struct'", id="struct" + _STRUCT, pa.array([{"a": 1}], _STRUCT), ValueError, "Join column 'k' has nested type 'struct'", id="struct" ), pytest.param( pa.list_(pa.int32()), pa.array([[1]], pa.list_(pa.int32())), ValueError, - "Nested column 'k' of type 'large_list", + "Join column 'k' has nested type 'large_list", id="list", ), pytest.param( - _MAP, pa.array([[("a", 1)]], _MAP), ValueError, "Nested column 'k' of type 'map'", id="map" + _MAP, pa.array([[("a", 1)]], _MAP), ValueError, "Join column 'k' has nested type 'map'", id="map" ), pytest.param( pa.uuid(), pa.array([_UUID_BYTES], pa.uuid()), NotImplementedError, - "Column 'k' of type 'extension'", + "Join column 'k' has type 'extension'", id="uuid-table", ), pytest.param( pa.uuid(), pa.array([_UUID_BYTES], pa.binary(16)), NotImplementedError, - "Column 'k' of type 'extension'", + "Join column 'k' has type 'extension'", id="uuid-table-fixed-source", ), pytest.param( pa.string(), pa.array([_UUID_BYTES], pa.uuid()), NotImplementedError, - "Extension type 'extension' for column 'k'", + "Input column 'k' has extension type 'extension'", id="uuid-extension-source", ), pytest.param( pa.string(), pa.array(["a"]).dictionary_encode(), NotImplementedError, - "Dictionary-encoded column 'k'", + "Input column 'k' is dictionary-encoded", id="dictionary-string", ), pytest.param( pa.int64(), pa.array([1]).dictionary_encode(), NotImplementedError, - "Dictionary-encoded column 'k'", + "Input column 'k' is dictionary-encoded", id="dictionary-int", ), - pytest.param(pa.int32(), pa.array([None], pa.null()), ValueError, "Null-type column 'k'", id="null-type"), + pytest.param(pa.int32(), pa.array([None], pa.null()), ValueError, "Input column 'k' has the null type", id="null-type"), pytest.param( pa.string(), pa.array(["a"], pa.string_view()), NotImplementedError, - "View-typed column 'k' of type 'string_view'", + "Input column 'k' has type 'string_view'", id="string-view", ), pytest.param( pa.binary(), pa.array([b"a"], pa.binary_view()), NotImplementedError, - "View-typed column 'k' of type 'binary_view'", + "Input column 'k' has type 'binary_view'", id="binary-view", ), pytest.param( @@ -1033,10 +1033,10 @@ def test_upsert_snapshot_properties(catalog: Catalog) -> None: id="run-end-encoded", ), pytest.param( - pa.int32(), pa.array([1, None], pa.int32()), ValueError, "Join column 'k' contains null values", id="null-values" + pa.int32(), pa.array([1, None], pa.int32()), ValueError, "Input column 'k' contains null values", id="null-values" ), pytest.param( - pa.int32(), pa.array([None], pa.int32()), ValueError, "Join column 'k' contains null values", id="all-null-values" + pa.int32(), pa.array([None], pa.int32()), ValueError, "Input column 'k' contains null values", id="all-null-values" ), pytest.param(pa.int32(), pa.array(["1"]), ValueError, "Mismatch in fields", id="wrong-type-source"), ], @@ -1059,16 +1059,14 @@ def test_upsert_rejects_unsupported_join_key( @pytest.mark.parametrize( "join_cols, identifier_field_ids, drop_source_columns, match", [ - pytest.param(["missing"], [], [], "Join column 'missing' does not exist in the table schema", id="not-in-table"), - pytest.param(["K"], [], [], "Join column 'K' does not exist in the table schema", id="case-mismatch"), - pytest.param(["k"], [], ["k"], "Join column 'k' does not exist in the source schema", id="required-not-in-source"), - pytest.param(["opt"], [], ["opt"], "Join column 'opt' does not exist in the source schema", id="optional-not-in-source"), - pytest.param( - ["k", "opt"], [], ["opt"], "Join column 'opt' does not exist in the source schema", id="composite-second-missing" - ), - pytest.param(["k", "k"], [], [], "Duplicate join columns: k", id="duplicate-join-cols"), - pytest.param(["s.x"], [], [], "Only top-level columns can be used as join keys", id="nested-path"), - pytest.param(None, [4], [], "Only top-level columns can be used as join keys", id="nested-identifier-field"), + pytest.param(["missing"], [], [], "Join column 'missing' does not exist in the table", id="not-in-table"), + pytest.param(["K"], [], [], "Join column 'K' does not exist in the table", id="case-mismatch"), + pytest.param(["k"], [], ["k"], "Join column 'k' is missing from the input", id="required-not-in-source"), + pytest.param(["opt"], [], ["opt"], "Join column 'opt' is missing from the input", id="optional-not-in-source"), + pytest.param(["k", "opt"], [], ["opt"], "Join column 'opt' is missing from the input", id="composite-second-missing"), + pytest.param(["k", "k"], [], [], "join_cols contains duplicates: k", id="duplicate-join-cols"), + pytest.param(["s.x"], [], [], "Join column 's.x' is a field inside struct 's'", id="nested-path"), + pytest.param(None, [4], [], "Join column 's.x' is a field inside struct 's'", id="nested-identifier-field"), pytest.param([], [], [], "Join columns could not be found", id="empty-join-cols"), pytest.param(None, [], [], "Join columns could not be found", id="no-identifier-fields"), pytest.param("k", [], [], "join_cols must be a list of column names, got str", id="string-not-list"), From 6894d1070bbffae0876bd3ca7ae9cb3cb4f176bc Mon Sep 17 00:00:00 2001 From: Abanoub Doss Date: Tue, 15 Sep 2026 18:05:07 -0500 Subject: [PATCH 11/11] refactor(upsert): split validate_join_cols by table and input side --- pyiceberg/table/upsert_util.py | 108 ++++++++++++++++++--------------- 1 file changed, 58 insertions(+), 50 deletions(-) diff --git a/pyiceberg/table/upsert_util.py b/pyiceberg/table/upsert_util.py index 2e02013449..a9feb4691e 100644 --- a/pyiceberg/table/upsert_util.py +++ b/pyiceberg/table/upsert_util.py @@ -35,61 +35,69 @@ def validate_join_cols(df: pyarrow_table, join_cols: list[str], table_schema: pa """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)}.") - df_column_names = set(df.schema.names) - for col in join_cols: - 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)}." - ) - table_field = table_schema.field(col) - # Table-level rejections: These types are fundamentally unreliable or - # unsupported as join keys regardless of the input data format. - if pa.types.is_floating(table_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(table_field.type): - raise ValueError( - f"Join column '{col}' has nested type '{table_field.type}'; only primitive columns can be join keys." - ) - if isinstance(table_field.type, pa.BaseExtensionType): - raise NotImplementedError( - f"Join column '{col}' has type '{table_field.type}', which is not yet supported as a join key." - ) - - # Schema compatibility permits missing optional fields, but upsert needs every join key. - if col not in df_column_names: - raise ValueError(f"Join column '{col}' is missing from the input.") - # Some input representations are unsupported even when the table type is valid. - 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. " - f"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.") + _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: