From 23683c30de2bc08f8b662cbd39f350a312bc98d3 Mon Sep 17 00:00:00 2001 From: Konstantinos Stefanidis Vozikis Date: Tue, 8 Sep 2026 18:11:27 +0200 Subject: [PATCH] fix: become compatible with pyiceberg 0.12 --- INSTALL-AND-REFERENCE.md | 16 ++++- src/tower/_tables.py | 30 ++++++-- tests/tower/test_table_retries.py | 6 +- tests/tower/test_table_schemas.py | 22 +++++- tests/tower/test_tables.py | 115 ++++++++++++------------------ 5 files changed, 107 insertions(+), 82 deletions(-) diff --git a/INSTALL-AND-REFERENCE.md b/INSTALL-AND-REFERENCE.md index 0cccf67e..e5ab37b1 100644 --- a/INSTALL-AND-REFERENCE.md +++ b/INSTALL-AND-REFERENCE.md @@ -146,9 +146,19 @@ PyIceberg, which assigns field IDs and preserves nested nullability and `b"doc"` metadata. Timestamp units from seconds through microseconds, UTC-zoned microsecond timestamps, `time64[us]`, `date32`, and Decimal128 values up to precision 38 are supported. Nanosecond timestamps are rejected by default instead of being silently downcast, as are -`time32`, `time64[ns]`, `date64`, Float16, Decimal256, and non-UTC zoned timestamps. Convert -those fields explicitly before creating the table when the loss is acceptable. PyIceberg's -native validation exceptions propagate unchanged. +`time32`, `time64[ns]`, `date64`, Decimal256, and non-UTC zoned timestamps. Float16 handling +follows the installed PyIceberg version: PyIceberg 0.12 and newer accept it and widen it +losslessly to Iceberg `float` (Arrow float32 on read), while PyIceberg 0.11 rejects it. +Convert unsupported fields explicitly before creating the table when the loss is acceptable. +PyIceberg's native validation exceptions propagate unchanged. + +With PyIceberg 0.12 and newer, compatible commit races are retried internally. When an +upsert or delete was planned from stale data and conflicts with a concurrent change, +PyIceberg raises `pyiceberg.exceptions.ValidationException`; Tower does not automatically +re-run the complete operation with last-writer-wins behavior. The failed operation does not +commit. Reload, reconcile the newer data, and explicitly retry only when that is appropriate +for the application. Tower's per-call retry options still handle a surfaced +`CommitFailedException`, including for supported PyIceberg 0.11 installations. ### dbt Core support diff --git a/src/tower/_tables.py b/src/tower/_tables.py index 40d5b53b..78fd73ce 100644 --- a/src/tower/_tables.py +++ b/src/tower/_tables.py @@ -294,6 +294,8 @@ def _commit_with_retry( initial_retry_ceiling_seconds, _MAX_COMMIT_RETRY_DELAY_SECONDS ) + # A PyIceberg ValidationException means that data changed incompatibly. + # It deliberately propagates instead of becoming last-writer-wins. for attempt in range(max_retries + 1): try: return operation() @@ -369,12 +371,17 @@ def upsert( retry_delay_seconds: float = 0.5, ) -> TTable: """ - Performs an upsert operation (update or insert) on the Iceberg table. In case of commit conflicts, reloads the metadata and retries. + Performs an upsert operation (update or insert) on the Iceberg table. + + PyIceberg 0.12 and newer retry compatible commit races internally. If + PyIceberg detects that a concurrent write changed data relevant to this + upsert, its ``ValidationException`` propagates instead of Tower retrying + the complete operation with last-writer-wins behavior. This method will: - Update existing rows if they match the join columns - Insert new rows if no match is found - - Retry for max_retries if commits fail + - Preserve PyIceberg's concurrency validation semantics All operations are case-sensitive by default. Args: @@ -382,8 +389,9 @@ def upsert( must match the schema of the target table. join_cols (Optional[list[str]]): The columns that form the key to match rows on. If not provided, all columns will be used for matching. - max_retries (int): Maximum number of retry attempts on commit conflicts. - Defaults to 5. + max_retries (int): Maximum number of Tower retry attempts when PyIceberg + surfaces a ``CommitFailedException``. This does not apply to validated + data conflicts. Defaults to 5. retry_delay_seconds (float): Maximum randomized wait before the first retry, in seconds. The maximum doubles after each conflict but never exceeds 30 seconds; values above 30 are treated as 30. Defaults to 0.5 seconds. @@ -393,6 +401,8 @@ def upsert( Raises: CommitFailedException: If all retry attempts are exhausted. + pyiceberg.exceptions.ValidationException: If PyIceberg detects an + incompatible concurrent write. Note: - The operation is always case-sensitive @@ -445,7 +455,10 @@ def delete( ) -> TTable: """ Deletes rows from the Iceberg table that match the specified filter conditions. - In case of commit conflicts, reloads the metadata and retries. + PyIceberg 0.12 and newer retry compatible commit races internally, but reject + incompatible concurrent data changes with ``ValidationException``. Tower lets + that conflict propagate rather than retrying the complete delete against newer + data. This method removes rows from the table based on the provided filter expressions. The operation is always case-sensitive. Note that the number of deleted rows @@ -454,8 +467,9 @@ def delete( Args: filters (str | BooleanExpression): A SQL-like string or a PyIceberg boolean expression. Use ``Table.column()`` to construct expressions. - max_retries (int): Maximum number of retry attempts on commit conflicts. - Defaults to 5. + max_retries (int): Maximum number of Tower retry attempts when PyIceberg + surfaces a ``CommitFailedException``. This does not apply to validated + data conflicts. Defaults to 5. retry_delay_seconds (float): Maximum randomized wait before the first retry, in seconds. The maximum doubles after each conflict but never exceeds 30 seconds; values above 30 are treated as 30. Defaults to 0.5 seconds. @@ -465,6 +479,8 @@ def delete( Raises: CommitFailedException: If all retry attempts are exhausted. + pyiceberg.exceptions.ValidationException: If PyIceberg detects an + incompatible concurrent write. Note: - The operation is always case-sensitive diff --git a/tests/tower/test_table_retries.py b/tests/tower/test_table_retries.py index 6de9a76e..0df1be57 100644 --- a/tests/tower/test_table_retries.py +++ b/tests/tower/test_table_retries.py @@ -13,6 +13,7 @@ ServerError, ServiceUnavailableError, UnauthorizedError, + ValidationException, WaitingForLockException, ) @@ -217,14 +218,13 @@ def test_commit_retry_exhaustion_preserves_final_exception(monkeypatch, max_retr ServerError, BadRequestError, WaitingForLockException, + ValidationException, httpx.TimeoutException, httpx.ConnectError, ValueError, ], ) -def test_mutation_errors_other_than_commit_conflicts_are_not_retried( - monkeypatch, exception_type -): +def test_non_retryable_mutation_errors_are_not_retried(monkeypatch, exception_type): failure = exception_type("not retryable") iceberg_table = FakeMutationTable([failure]) table = make_table(iceberg_table) diff --git a/tests/tower/test_table_schemas.py b/tests/tower/test_table_schemas.py index 2683f05e..7bf5751e 100644 --- a/tests/tower/test_table_schemas.py +++ b/tests/tower/test_table_schemas.py @@ -1,3 +1,5 @@ +from importlib.metadata import version + import pyarrow as pa import pytest from pyiceberg import types as iceberg_types @@ -9,6 +11,11 @@ from tower._context import TowerContext +_PYICEBERG_SUPPORTS_FLOAT16 = tuple( + int(component) for component in version("pyiceberg").split(".")[:2] +) >= (0, 12) + + class RecordingCatalog: def __init__(self): self.schemas = [] @@ -201,6 +208,20 @@ def test_pyiceberg_accepts_supported_arrow_precision( assert table.schema().find_field("value").field_type == iceberg_type +def test_float16_schema_follows_pyiceberg_version(in_memory_schema_catalog): + schema = pa.schema([pa.field("value", pa.float16())]) + + if not _PYICEBERG_SUPPORTS_FLOAT16: + with pytest.raises(UnsupportedPyArrowTypeException): + make_reference(in_memory_schema_catalog, "float16").create(schema) + return + + make_reference(in_memory_schema_catalog, "float16").create(schema) + + table = in_memory_schema_catalog.load_table("default.float16") + assert table.schema().find_field("value").field_type == iceberg_types.FloatType() + + @pytest.mark.parametrize( ("name", "arrow_type"), [ @@ -208,7 +229,6 @@ def test_pyiceberg_accepts_supported_arrow_precision( ("timestamp_non_utc", pa.timestamp("us", tz="Europe/Berlin")), ("time32", pa.time32("s")), ("time_ns", pa.time64("ns")), - ("float16", pa.float16()), ("date64", pa.date64()), ("decimal256", pa.decimal256(38, 10)), ], diff --git a/tests/tower/test_tables.py b/tests/tower/test_tables.py index acd428f7..3d574988 100644 --- a/tests/tower/test_tables.py +++ b/tests/tower/test_tables.py @@ -3,6 +3,7 @@ import datetime import tempfile import pathlib +from importlib.metadata import version from urllib.parse import urljoin from urllib.request import pathname2url import threading @@ -12,7 +13,7 @@ import pyarrow as pa from pyiceberg.catalog.memory import InMemoryCatalog from pyiceberg.catalog.sql import SqlCatalog -from pyiceberg.exceptions import CommitFailedException +from pyiceberg.exceptions import CommitFailedException, ValidationException import concurrent.futures @@ -28,6 +29,11 @@ ) +_PYICEBERG_HAS_WRITE_CONFLICT_VALIDATION = tuple( + int(component) for component in version("pyiceberg").split(".")[:2] +) >= (0, 12) + + class FakeLoadedTable: def __init__(self, mode: str, identifier: str): self.mode = mode @@ -622,8 +628,9 @@ def test_upsert_to_tables(in_memory_catalog): assert res["age"].item() == 26 -def test_upsert_concurrent_writes_with_retry(sql_catalog): - """Test that concurrent upserts succeed with retry logic handling conflicts.""" +def test_stale_upserts_to_different_rows_follow_pyiceberg_conflict_semantics( + sql_catalog, +): schema = pa.schema( [ pa.field("ticker", pa.string()), @@ -645,40 +652,28 @@ def test_upsert_concurrent_writes_with_retry(sql_catalog): ) table.insert(initial_data) - retry_count = {"value": 0} - retry_lock = threading.Lock() - - def upsert_ticker(ticker: str, new_price: float): - t = tower.tables("concurrent_test", catalog=sql_catalog).load() - - original_refresh = t._table.refresh + first_writer = tower.tables("concurrent_test", catalog=sql_catalog).load() + stale_writer = tower.tables("concurrent_test", catalog=sql_catalog).load() - def tracked_refresh(): - with retry_lock: - retry_count["value"] += 1 - return original_refresh() - - t._table.refresh = tracked_refresh - - data = pa.Table.from_pylist( - [{"ticker": ticker, "date": "2024-01-01", "price": new_price}], + first_writer.upsert( + pa.Table.from_pylist( + [{"ticker": "AAPL", "date": "2024-01-01", "price": 150.0}], schema=schema, - ) - t.upsert(data, join_cols=["ticker", "date"]) - return ticker - - with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor: - futures = [ - executor.submit(upsert_ticker, "AAPL", 150.0), - executor.submit(upsert_ticker, "GOOGL", 250.0), - executor.submit(upsert_ticker, "MSFT", 350.0), - ] - results = [f.result() for f in concurrent.futures.as_completed(futures)] + ), + join_cols=["ticker", "date"], + ) - assert len(results) == 3 - assert ( - retry_count["value"] > 0 - ), "Expected at least one retry due to concurrent conflicts" + stale_upsert = pa.Table.from_pylist( + [{"ticker": "GOOGL", "date": "2024-01-01", "price": 250.0}], + schema=schema, + ) + if _PYICEBERG_HAS_WRITE_CONFLICT_VALIDATION: + with pytest.raises(ValidationException): + stale_writer.upsert(stale_upsert, join_cols=["ticker", "date"]) + else: + # PyIceberg 0.11 surfaces CommitFailedException, so Tower refreshes and + # replans the whole operation for supported older installations. + stale_writer.upsert(stale_upsert, join_cols=["ticker", "date"]) final_table = tower.tables("concurrent_test", catalog=sql_catalog).load() df = final_table.read() @@ -688,12 +683,13 @@ def tracked_refresh(): ticker_prices = {row["ticker"]: row["price"] for row in df.iter_rows(named=True)} assert ticker_prices["AAPL"] == 150.0 - assert ticker_prices["GOOGL"] == 250.0 - assert ticker_prices["MSFT"] == 350.0 + assert ticker_prices["GOOGL"] == ( + 200.0 if _PYICEBERG_HAS_WRITE_CONFLICT_VALIDATION else 250.0 + ) + assert ticker_prices["MSFT"] == 300.0 -def test_upsert_concurrent_writes_same_row(sql_catalog): - """Test concurrent upserts to the SAME row - last write wins.""" +def test_stale_upserts_to_same_row_follow_pyiceberg_conflict_semantics(sql_catalog): schema = pa.schema( [ pa.field("id", pa.int64()), @@ -710,37 +706,20 @@ def test_upsert_concurrent_writes_same_row(sql_catalog): ) table.insert(initial_data) - retry_count = {"value": 0} - retry_lock = threading.Lock() - - def upsert_counter(value: int): - t = tower.tables("concurrent_same_row_test", catalog=sql_catalog).load() - - original_refresh = t._table.refresh - - def tracked_refresh(): - with retry_lock: - retry_count["value"] += 1 - return original_refresh() - - t._table.refresh = tracked_refresh + first_writer = tower.tables("concurrent_same_row_test", catalog=sql_catalog).load() + stale_writer = tower.tables("concurrent_same_row_test", catalog=sql_catalog).load() - data = pa.Table.from_pylist( - [{"id": 1, "counter": value}], - schema=schema, - ) - t.upsert(data, join_cols=["id"]) - return value - - with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: - futures = [executor.submit(upsert_counter, i) for i in range(1, 6)] - results = [f.result() for f in concurrent.futures.as_completed(futures)] - - assert len(results) == 5 + first_writer.upsert( + pa.Table.from_pylist([{"id": 1, "counter": 1}], schema=schema), + join_cols=["id"], + ) - assert ( - retry_count["value"] > 0 - ), "Expected at least one retry due to concurrent conflicts" + stale_upsert = pa.Table.from_pylist([{"id": 1, "counter": 2}], schema=schema) + if _PYICEBERG_HAS_WRITE_CONFLICT_VALIDATION: + with pytest.raises(ValidationException): + stale_writer.upsert(stale_upsert, join_cols=["id"]) + else: + stale_writer.upsert(stale_upsert, join_cols=["id"]) final_table = tower.tables("concurrent_same_row_test", catalog=sql_catalog).load() df = final_table.read() @@ -748,7 +727,7 @@ def tracked_refresh(): assert len(df) == 1 final_counter = df.select("counter").item() - assert final_counter in [1, 2, 3, 4, 5] + assert final_counter == (1 if _PYICEBERG_HAS_WRITE_CONFLICT_VALIDATION else 2) def test_insert_concurrent_writes_with_retry(sql_catalog):