fix: become compatible with pyiceberg 0.12

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
 

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,21 +371,27 @@ 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:
             data (pa.Table): The data to upsert into the table. The schema of this table
                 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

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)

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,14 +208,27 @@ 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"),
     [
         ("timestamp_ns", pa.timestamp("ns")),
         ("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)),
     ],

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,45 +706,28 @@ 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()
 
     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):