Skip to content

Commit 8c59067

Browse files
kotwal-itproclaude
andcommitted
Support atomic multi-table commits through the REST catalog
Add Catalog.commit_transactions(*transactions), which commits the staged changes of several table transactions in one request to the REST catalog's transactions/commit endpoint: either every table is updated, or none of them. Each table is refreshed after a successful commit. Other catalogs raise NotImplementedError. A transaction can be given only once, each table at most once, and creating a table is not supported. Conflicts are not retried. On a definite failure (409, 404) the files written for the transactions are deleted; on an unknown outcome (500, 502, 504) they are kept, as in commit_transaction. The endpoint is assumed supported by servers that don't list their endpoints, as in the Java client. Closes #4095 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011ZBQVsXRmKT8WXDr2qSfbd
1 parent eb30d8a commit 8c59067

7 files changed

Lines changed: 407 additions & 1 deletion

File tree

‎mkdocs/docs/api.md‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1408,6 +1408,30 @@ assert tbl.metadata.snapshots[-1].summary["abc"] == "def"
14081408
New snapshot summaries automatically include `engine-name` (`pyiceberg`) and `engine-version`
14091409
(the installed PyIceberg version). These values override same-named entries in `snapshot_properties`.
14101410

1411+
## Multi-table commits
1412+
1413+
With the REST catalog, changes to several tables can be committed atomically: either every table
1414+
is updated, or none of them is. Stage the changes in one transaction per table, then pass the
1415+
transactions to `commit_transactions`:
1416+
1417+
```python
1418+
orders_tx = orders.transaction()
1419+
orders_tx.append(new_orders)
1420+
items_tx = items.transaction()
1421+
items_tx.append(new_items)
1422+
1423+
catalog.commit_transactions(orders_tx, items_tx)
1424+
```
1425+
1426+
The commit uses the `POST /v1/{prefix}/transactions/commit` endpoint of the REST catalog
1427+
specification, and each table is refreshed afterwards. A few things differ from committing one
1428+
transaction:
1429+
1430+
- A conflict raises `CommitFailedException` and isn't retried: nothing was applied, so stage the
1431+
changes again and commit again.
1432+
- Each table can appear in only one transaction, and creating a table isn't supported.
1433+
- Other catalogs raise `NotImplementedError`.
1434+
14111435
## Snapshot Management
14121436

14131437
Manage snapshots with operations through the `Table` API:

‎pyiceberg/catalog/__init__.py‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,7 @@
5252
StagedTable,
5353
Table,
5454
TableProperties,
55+
Transaction,
5556
)
5657
from pyiceberg.table.locations import load_location_provider
5758
from pyiceberg.table.metadata import TableMetadata, TableMetadataV1, new_table_metadata
@@ -570,6 +571,23 @@ def commit_table(
570571
CommitStateUnknownException: Failed due to an internal exception on the side of the catalog.
571572
"""
572573

574+
def commit_transactions(self, *transactions: Transaction) -> None:
575+
"""Commit the changes of several table transactions atomically.
576+
577+
Either the changes of every transaction are applied, or none of them. Each table is
578+
refreshed after a successful commit.
579+
580+
Args:
581+
transactions: The transactions to commit, at most one per table.
582+
583+
Raises:
584+
NotImplementedError: If the catalog doesn't support multi-table commits.
585+
CommitFailedException: A requirement wasn't met, or a concurrent commit conflicted.
586+
CommitStateUnknownException: The catalog failed with an internal error; the commit may
587+
or may not have been applied.
588+
"""
589+
raise NotImplementedError(f"{type(self).__name__} does not support multi-table commits")
590+
573591
@abstractmethod
574592
def create_namespace(self, namespace: str | Identifier, properties: Properties = EMPTY_DICT) -> None:
575593
"""Create a namespace in the catalog.

‎pyiceberg/catalog/rest/__init__.py‎

Lines changed: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -81,12 +81,14 @@
8181
from pyiceberg.table import (
8282
CommitTableRequest,
8383
CommitTableResponse,
84+
CommitTransactionRequest,
8485
CreateTableTransaction,
8586
FileScanTask,
8687
StagedTable,
8788
Table,
8889
TableIdentifier,
8990
TableProperties,
91+
Transaction,
9092
)
9193
from pyiceberg.table.metadata import TableMetadata
9294
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder, assign_fresh_sort_order_ids
@@ -181,6 +183,7 @@ class Endpoints:
181183
fetch_planning_result: str = "namespaces/{namespace}/tables/{table}/plan/{plan_id}"
182184
cancel_planning: str = "namespaces/{namespace}/tables/{table}/plan/{plan_id}"
183185
fetch_scan_tasks: str = "namespaces/{namespace}/tables/{table}/tasks"
186+
commit_transaction: str = "transactions/commit"
184187

185188

186189
API_PREFIX = "/v1/{prefix}"
@@ -220,6 +223,7 @@ class Capability:
220223
http_method=HttpMethod.DELETE, path=f"{API_PREFIX}/namespaces/{{namespace}}/tables/{{table}}/plan/{{plan-id}}"
221224
)
222225
V1_TABLE_SCAN_PLAN_TASKS = Endpoint(http_method=HttpMethod.POST, path=f"{API_PREFIX}/{Endpoints.fetch_scan_tasks}")
226+
V1_COMMIT_TRANSACTION = Endpoint(http_method=HttpMethod.POST, path=f"{API_PREFIX}/{Endpoints.commit_transaction}")
223227

224228

225229
# Default endpoints for backwards compatibility with legacy servers that don't return endpoints
@@ -238,6 +242,7 @@ class Capability:
238242
Capability.V1_DELETE_TABLE,
239243
Capability.V1_RENAME_TABLE,
240244
Capability.V1_REGISTER_TABLE,
245+
Capability.V1_COMMIT_TRANSACTION,
241246
)
242247
)
243248

@@ -1634,6 +1639,97 @@ def commit_table(
16341639
)
16351640
return CommitTableResponse.model_validate_json(response.text)
16361641

1642+
@override
1643+
def commit_transactions(self, *transactions: Transaction) -> None:
1644+
"""Commit the changes of several table transactions atomically.
1645+
1646+
Sends the staged changes of every transaction in one request to the
1647+
``transactions/commit`` endpoint: either all of them are applied, or none. Each
1648+
table is refreshed after a successful commit. Unlike ``Transaction.commit_transaction``,
1649+
a conflict is not retried: re-stage the changes and commit again.
1650+
1651+
Args:
1652+
transactions: The transactions to commit, at most one per table. Transactions
1653+
without staged changes are skipped.
1654+
1655+
Raises:
1656+
ValueError: If no transaction was given, a transaction is given twice, two
1657+
transactions change the same table, a transaction belongs to another catalog,
1658+
or a transaction creates a table.
1659+
NotImplementedError: If the server does not support multi-table commits.
1660+
NoSuchTableError: If one of the tables does not exist.
1661+
CommitFailedException: A requirement wasn't met, or a concurrent commit conflicted.
1662+
Nothing was applied.
1663+
CommitStateUnknownException: The catalog failed with an internal error; the commit
1664+
may or may not have been applied.
1665+
"""
1666+
if not transactions:
1667+
raise ValueError("At least one transaction is required")
1668+
if len({id(transaction) for transaction in transactions}) != len(transactions):
1669+
raise ValueError("A transaction can be committed only once per call")
1670+
identifiers: set[Identifier] = set()
1671+
for transaction in transactions:
1672+
if isinstance(transaction, CreateTableTransaction):
1673+
raise ValueError("Creating a table in a multi-table commit is not supported")
1674+
table = transaction._table # pylint: disable=W0212
1675+
if table.catalog is not self:
1676+
raise ValueError(f"Table {table.name()} belongs to a different catalog")
1677+
if table.name() in identifiers:
1678+
raise ValueError(f"More than one transaction changes table {table.name()}")
1679+
identifiers.add(table.name())
1680+
1681+
staged = [transaction for transaction in transactions if transaction._updates] # pylint: disable=W0212
1682+
if not staged:
1683+
return
1684+
self._check_endpoint(Capability.V1_COMMIT_TRANSACTION)
1685+
1686+
table_changes = []
1687+
for transaction in staged:
1688+
requirements, updates = transaction._changes_for_multi_table_commit() # pylint: disable=W0212
1689+
identifier = transaction._table.name() # pylint: disable=W0212
1690+
table_changes.append(
1691+
CommitTableRequest(
1692+
identifier=TableIdentifier(namespace=identifier[:-1], name=identifier[-1]),
1693+
requirements=requirements,
1694+
updates=updates,
1695+
)
1696+
)
1697+
request = CommitTransactionRequest(table_changes=table_changes)
1698+
1699+
try:
1700+
self._post_commit_transaction(request)
1701+
except (CommitFailedException, NoSuchTableError):
1702+
for transaction in staged:
1703+
transaction._multi_table_commit_failed(outcome_known=True) # pylint: disable=W0212
1704+
raise
1705+
except Exception:
1706+
for transaction in staged:
1707+
transaction._multi_table_commit_failed(outcome_known=False) # pylint: disable=W0212
1708+
raise
1709+
1710+
for transaction in staged:
1711+
transaction._multi_table_commit_succeeded() # pylint: disable=W0212
1712+
1713+
@retry(**_RETRY_ARGS)
1714+
def _post_commit_transaction(self, request: CommitTransactionRequest) -> None:
1715+
response = self._session.post(
1716+
self.url(Endpoints.commit_transaction, prefixed=True),
1717+
data=request.model_dump_json().encode(UTF8),
1718+
)
1719+
try:
1720+
response.raise_for_status()
1721+
except HTTPError as exc:
1722+
_handle_non_200_response(
1723+
exc,
1724+
{
1725+
404: NoSuchTableError,
1726+
409: CommitFailedException,
1727+
500: CommitStateUnknownException,
1728+
502: CommitStateUnknownException,
1729+
504: CommitStateUnknownException,
1730+
},
1731+
)
1732+
16371733
@retry(**_RETRY_ARGS)
16381734
@override
16391735
def create_namespace(self, namespace: str | Identifier, properties: Properties = EMPTY_DICT) -> None:

‎pyiceberg/table/__init__.py‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1203,6 +1203,39 @@ def commit_transaction(self) -> Table:
12031203

12041204
return self._table
12051205

1206+
def _changes_for_multi_table_commit(self) -> tuple[tuple[TableRequirement, ...], tuple[TableUpdate, ...]]:
1207+
"""Return the requirements and updates to send in a multi-table commit.
1208+
1209+
Adds the same ``AssertTableUUID`` requirement that ``commit_transaction`` adds, so the
1210+
commit fails if the table was dropped and recreated under the same name.
1211+
"""
1212+
if self._failed:
1213+
raise RuntimeError("This transaction failed to commit and cannot be reused; create a new transaction.")
1214+
if not any(isinstance(requirement, AssertTableUUID) for requirement in self._requirements):
1215+
self._requirements += (AssertTableUUID(uuid=self.table_metadata.table_uuid),)
1216+
return self._requirements, self._updates
1217+
1218+
def _multi_table_commit_succeeded(self) -> None:
1219+
"""Clean up after a multi-table commit that included this transaction succeeded."""
1220+
self._cleanup_uncommitted_manifests()
1221+
self._snapshot_producers = []
1222+
self._updates = ()
1223+
self._requirements = ()
1224+
self._table.refresh()
1225+
1226+
def _multi_table_commit_failed(self, outcome_known: bool) -> None:
1227+
"""Handle a multi-table commit that included this transaction and did not succeed.
1228+
1229+
Args:
1230+
outcome_known: True if the commit is known not to have landed, so the files written
1231+
for it can be deleted. False if the outcome is unknown: the files are kept,
1232+
because the catalog may already reference them.
1233+
"""
1234+
if outcome_known:
1235+
for producer in self._snapshot_producers:
1236+
producer._clean_all_uncommitted()
1237+
self._failed = True
1238+
12061239
def _cleanup_uncommitted_manifests(self) -> None:
12071240
"""Clean up manifests from failed retry attempts after a successful commit."""
12081241
for producer in self._snapshot_producers:
@@ -1310,6 +1343,12 @@ class CommitTableRequest(IcebergBaseModel):
13101343
updates: tuple[TableUpdate, ...] = Field(default_factory=tuple)
13111344

13121345

1346+
class CommitTransactionRequest(IcebergBaseModel):
1347+
"""A pydantic BaseModel for a multi-table commit request."""
1348+
1349+
table_changes: list[CommitTableRequest] = Field(alias="table-changes")
1350+
1351+
13131352
class CommitTableResponse(IcebergBaseModel):
13141353
"""A pydantic BaseModel for a table commit response."""
13151354

‎tests/catalog/test_base.py‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -88,3 +88,8 @@ def test_in_memory_catalog_context_manager(self, catalog: InMemoryCatalog) -> No
8888

8989
# InMemoryCatalog inherits close from SqlCatalog, so engine should be disposed
9090
assert hasattr(cat, "engine")
91+
92+
93+
def test_commit_transactions_not_supported(catalog: InMemoryCatalog) -> None:
94+
with pytest.raises(NotImplementedError, match="does not support multi-table commits"):
95+
catalog.commit_transactions()

0 commit comments

Comments
 (0)