diff --git a/mkdocs/docs/api.md b/mkdocs/docs/api.md index 786223d47e..058657d568 100644 --- a/mkdocs/docs/api.md +++ b/mkdocs/docs/api.md @@ -1408,6 +1408,30 @@ assert tbl.metadata.snapshots[-1].summary["abc"] == "def" New snapshot summaries automatically include `engine-name` (`pyiceberg`) and `engine-version` (the installed PyIceberg version). These values override same-named entries in `snapshot_properties`. +## Multi-table commits + +With the REST catalog, changes to several tables can be committed atomically: either every table +is updated, or none of them is. Stage the changes in one transaction per table, then pass the +transactions to `commit_transactions`: + +```python +orders_tx = orders.transaction() +orders_tx.append(new_orders) +items_tx = items.transaction() +items_tx.append(new_items) + +catalog.commit_transactions(orders_tx, items_tx) +``` + +The commit uses the `POST /v1/{prefix}/transactions/commit` endpoint of the REST catalog +specification, and each table is refreshed afterwards. A few things differ from committing one +transaction: + +- A conflict raises `CommitFailedException` and isn't retried: nothing was applied, so stage the + changes again and commit again. +- Each table can appear in only one transaction, and creating a table isn't supported. +- Other catalogs raise `NotImplementedError`. + ## Snapshot Management Manage snapshots with operations through the `Table` API: diff --git a/pyiceberg/catalog/__init__.py b/pyiceberg/catalog/__init__.py index 0842f51cc8..0b9d751672 100644 --- a/pyiceberg/catalog/__init__.py +++ b/pyiceberg/catalog/__init__.py @@ -52,6 +52,7 @@ StagedTable, Table, TableProperties, + Transaction, ) from pyiceberg.table.locations import load_location_provider from pyiceberg.table.metadata import TableMetadata, TableMetadataV1, new_table_metadata @@ -570,6 +571,23 @@ def commit_table( CommitStateUnknownException: Failed due to an internal exception on the side of the catalog. """ + def commit_transactions(self, *transactions: Transaction) -> None: + """Commit the changes of several table transactions atomically. + + Either the changes of every transaction are applied, or none of them. Each table is + refreshed after a successful commit. + + Args: + transactions: The transactions to commit, at most one per table. + + Raises: + NotImplementedError: If the catalog doesn't support multi-table commits. + CommitFailedException: A requirement wasn't met, or a concurrent commit conflicted. + CommitStateUnknownException: The catalog failed with an internal error; the commit may + or may not have been applied. + """ + raise NotImplementedError(f"{type(self).__name__} does not support multi-table commits") + @abstractmethod def create_namespace(self, namespace: str | Identifier, properties: Properties = EMPTY_DICT) -> None: """Create a namespace in the catalog. diff --git a/pyiceberg/catalog/rest/__init__.py b/pyiceberg/catalog/rest/__init__.py index 2b8b4496f8..95771abba3 100644 --- a/pyiceberg/catalog/rest/__init__.py +++ b/pyiceberg/catalog/rest/__init__.py @@ -81,12 +81,14 @@ from pyiceberg.table import ( CommitTableRequest, CommitTableResponse, + CommitTransactionRequest, CreateTableTransaction, FileScanTask, StagedTable, Table, TableIdentifier, TableProperties, + Transaction, ) from pyiceberg.table.metadata import TableMetadata from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder, assign_fresh_sort_order_ids @@ -181,6 +183,7 @@ class Endpoints: fetch_planning_result: str = "namespaces/{namespace}/tables/{table}/plan/{plan_id}" cancel_planning: str = "namespaces/{namespace}/tables/{table}/plan/{plan_id}" fetch_scan_tasks: str = "namespaces/{namespace}/tables/{table}/tasks" + commit_transaction: str = "transactions/commit" API_PREFIX = "/v1/{prefix}" @@ -220,6 +223,7 @@ class Capability: http_method=HttpMethod.DELETE, path=f"{API_PREFIX}/namespaces/{{namespace}}/tables/{{table}}/plan/{{plan-id}}" ) V1_TABLE_SCAN_PLAN_TASKS = Endpoint(http_method=HttpMethod.POST, path=f"{API_PREFIX}/{Endpoints.fetch_scan_tasks}") + V1_COMMIT_TRANSACTION = Endpoint(http_method=HttpMethod.POST, path=f"{API_PREFIX}/{Endpoints.commit_transaction}") # Default endpoints for backwards compatibility with legacy servers that don't return endpoints @@ -238,6 +242,7 @@ class Capability: Capability.V1_DELETE_TABLE, Capability.V1_RENAME_TABLE, Capability.V1_REGISTER_TABLE, + Capability.V1_COMMIT_TRANSACTION, ) ) @@ -1634,6 +1639,103 @@ def commit_table( ) return CommitTableResponse.model_validate_json(response.text) + @override + def commit_transactions(self, *transactions: Transaction) -> None: + """Commit the changes of several table transactions atomically. + + Sends the staged changes of every transaction in one request to the + ``transactions/commit`` endpoint: either all of them are applied, or none. Each + table is refreshed after a successful commit. Unlike ``Transaction.commit_transaction``, + a conflict is not retried: re-stage the changes and commit again. + + Args: + transactions: The transactions to commit, at most one per table. Transactions + without staged changes are skipped. A snapshot operation that staged nothing, + such as a delete that matched no rows, is still sent with a requirement that + the table has not changed, so a concurrent change to it fails the whole commit. + + Raises: + ValueError: If no transaction was given, a transaction is given twice, two + transactions change the same table, a transaction belongs to another catalog, + or a transaction creates a table. + NotImplementedError: If the server does not support multi-table commits. + NoSuchTableError: If one of the tables does not exist. + CommitFailedException: A requirement wasn't met, or a concurrent commit conflicted. + Nothing was applied. + CommitStateUnknownException: The catalog failed with an internal error; the commit + may or may not have been applied. + """ + if not transactions: + raise ValueError("At least one transaction is required") + if len({id(transaction) for transaction in transactions}) != len(transactions): + raise ValueError("A transaction can be committed only once per call") + identifiers: set[Identifier] = set() + for transaction in transactions: + if isinstance(transaction, CreateTableTransaction): + raise ValueError("Creating a table in a multi-table commit is not supported") + table = transaction._table # pylint: disable=W0212 + if table.catalog is not self: + raise ValueError(f"Table {table.name()} belongs to a different catalog") + if table.name() in identifiers: + raise ValueError(f"More than one transaction changes table {table.name()}") + identifiers.add(table.name()) + + staged = [ + transaction + for transaction in transactions + if transaction._updates or transaction._snapshot_producers # pylint: disable=W0212 + ] + if not staged: + return + self._check_endpoint(Capability.V1_COMMIT_TRANSACTION) + + table_changes = [] + for transaction in staged: + requirements, updates = transaction._changes_for_multi_table_commit() # pylint: disable=W0212 + identifier = transaction._table.name() # pylint: disable=W0212 + table_changes.append( + CommitTableRequest( + identifier=TableIdentifier(namespace=identifier[:-1], name=identifier[-1]), + requirements=requirements, + updates=updates, + ) + ) + request = CommitTransactionRequest(table_changes=table_changes) + + try: + self._post_commit_transaction(request) + except (CommitFailedException, NoSuchTableError): + for transaction in staged: + transaction._multi_table_commit_failed(outcome_known=True) # pylint: disable=W0212 + raise + except Exception: + for transaction in staged: + transaction._multi_table_commit_failed(outcome_known=False) # pylint: disable=W0212 + raise + + for transaction in staged: + transaction._multi_table_commit_succeeded() # pylint: disable=W0212 + + @retry(**_RETRY_ARGS) + def _post_commit_transaction(self, request: CommitTransactionRequest) -> None: + response = self._session.post( + self.url(Endpoints.commit_transaction, prefixed=True), + data=request.model_dump_json().encode(UTF8), + ) + try: + response.raise_for_status() + except HTTPError as exc: + _handle_non_200_response( + exc, + { + 404: NoSuchTableError, + 409: CommitFailedException, + 500: CommitStateUnknownException, + 502: CommitStateUnknownException, + 504: CommitStateUnknownException, + }, + ) + @retry(**_RETRY_ARGS) @override def create_namespace(self, namespace: str | Identifier, properties: Properties = EMPTY_DICT) -> None: diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index 5332a46b0a..39c8a099a9 100644 --- a/pyiceberg/table/__init__.py +++ b/pyiceberg/table/__init__.py @@ -1211,6 +1211,52 @@ def commit_transaction(self) -> Table: return self._table + def _changes_for_multi_table_commit(self) -> tuple[tuple[TableRequirement, ...], tuple[TableUpdate, ...]]: + """Return the requirements and updates to send in a multi-table commit. + + Adds the same ``AssertTableUUID`` requirement that ``commit_transaction`` adds, so the + commit fails if the table was dropped and recreated under the same name. + + A snapshot operation that staged nothing (for example a delete whose filter matched no + rows) is sent as a requirement that the branch still points at the snapshot it started + from, with no updates. ``commit_transaction`` validates that case on the client after a + refresh; here the catalog checks it in the same atomic commit as the other tables. + """ + if self._failed: + raise RuntimeError("This transaction failed to commit and cannot be reused; create a new transaction.") + if not any(isinstance(requirement, AssertTableUUID) for requirement in self._requirements): + self._requirements += (AssertTableUUID(uuid=self.table_metadata.table_uuid),) + if not self._updates and self._snapshot_producers: + producer = self._snapshot_producers[0] + if producer._target_branch is not None and not any( + isinstance(requirement, AssertRefSnapshotId) for requirement in self._requirements + ): + self._requirements += ( + AssertRefSnapshotId(snapshot_id=producer._starting_snapshot_id, ref=producer._target_branch), + ) + return self._requirements, self._updates + + def _multi_table_commit_succeeded(self) -> None: + """Clean up after a multi-table commit that included this transaction succeeded.""" + self._cleanup_uncommitted_manifests() + self._snapshot_producers = [] + self._updates = () + self._requirements = () + self._table.refresh() + + def _multi_table_commit_failed(self, outcome_known: bool) -> None: + """Handle a multi-table commit that included this transaction and did not succeed. + + Args: + outcome_known: True if the commit is known not to have landed, so the files written + for it can be deleted. False if the outcome is unknown: the files are kept, + because the catalog may already reference them. + """ + if outcome_known: + for producer in self._snapshot_producers: + producer._clean_all_uncommitted() + self._failed = True + def _cleanup_uncommitted_manifests(self) -> None: """Clean up manifests from failed retry attempts after a successful commit.""" for producer in self._snapshot_producers: @@ -1318,6 +1364,12 @@ class CommitTableRequest(IcebergBaseModel): updates: tuple[TableUpdate, ...] = Field(default_factory=tuple) +class CommitTransactionRequest(IcebergBaseModel): + """A pydantic BaseModel for a multi-table commit request.""" + + table_changes: list[CommitTableRequest] = Field(alias="table-changes") + + class CommitTableResponse(IcebergBaseModel): """A pydantic BaseModel for a table commit response.""" diff --git a/tests/catalog/test_base.py b/tests/catalog/test_base.py index 1a47478313..b9e9380669 100644 --- a/tests/catalog/test_base.py +++ b/tests/catalog/test_base.py @@ -88,3 +88,8 @@ def test_in_memory_catalog_context_manager(self, catalog: InMemoryCatalog) -> No # InMemoryCatalog inherits close from SqlCatalog, so engine should be disposed assert hasattr(cat, "engine") + + +def test_commit_transactions_not_supported(catalog: InMemoryCatalog) -> None: + with pytest.raises(NotImplementedError, match="does not support multi-table commits"): + catalog.commit_transactions() diff --git a/tests/catalog/test_rest.py b/tests/catalog/test_rest.py index 4877ee2f73..12e11c500d 100644 --- a/tests/catalog/test_rest.py +++ b/tests/catalog/test_rest.py @@ -54,6 +54,8 @@ ) from pyiceberg.exceptions import ( AuthorizationExpiredError, + CommitFailedException, + CommitStateUnknownException, NamespaceAlreadyExistsError, NamespaceNotEmptyError, NoSuchIdentifierError, @@ -122,6 +124,7 @@ Capability.V1_DELETE_VIEW, Capability.V1_SUBMIT_TABLE_SCAN_PLAN, Capability.V1_TABLE_SCAN_PLAN_TASKS, + Capability.V1_COMMIT_TRANSACTION, ] @@ -3533,3 +3536,199 @@ def test_load_table_without_storage_credentials( ) assert actual.metadata.model_dump() == expected.metadata.model_dump() assert actual == expected + + +def _load_tables_for_multi_table_commit( + rest_mock: Mocker, table_metadata: dict[str, Any], names: tuple[str, ...] = ("orders", "items") +) -> tuple[RestCatalog, list[Table]]: + for name in names: + rest_mock.get( + f"{TEST_URI}v1/namespaces/sales/tables/{name}", + json=table_metadata, + status_code=200, + request_headers=TEST_HEADERS, + ) + catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN) + return catalog, [catalog.load_table(("sales", name)) for name in names] + + +def test_commit_transactions_sends_one_request( + rest_mock: Mocker, example_table_metadata_with_snapshot_v1_rest_json: dict[str, Any] +) -> None: + catalog, (orders, items) = _load_tables_for_multi_table_commit(rest_mock, example_table_metadata_with_snapshot_v1_rest_json) + commit = rest_mock.post(f"{TEST_URI}v1/transactions/commit", status_code=204, request_headers=TEST_HEADERS) + table_uuid = example_table_metadata_with_snapshot_v1_rest_json["metadata"]["table-uuid"] + + orders_tx = orders.transaction().set_properties(owner="sales") + items_tx = items.transaction().set_properties(owner="sales") + catalog.commit_transactions(orders_tx, items_tx) + + assert commit.call_count == 1 + assert commit.last_request.json() == { + "table-changes": [ + { + "identifier": {"namespace": ["sales"], "name": name}, + "requirements": [{"type": "assert-table-uuid", "uuid": table_uuid}], + "updates": [{"action": "set-properties", "updates": {"owner": "sales"}}], + } + for name in ("orders", "items") + ] + } + # Each table is loaded again after the commit, and the transactions are empty. + assert rest_mock.call_count == 1 + 2 + 1 + 2 + assert orders_tx._updates == () and items_tx._updates == () + + +def test_commit_transactions_with_prefix( + requests_mock: Mocker, example_table_metadata_with_snapshot_v1_rest_json: dict[str, Any] +) -> None: + requests_mock.get( + f"{TEST_URI}v1/config", + json={ + "defaults": {}, + "overrides": {"prefix": "lake"}, + "endpoints": [str(endpoint) for endpoint in TEST_SUPPORTED_ENDPOINTS], + }, + status_code=200, + ) + requests_mock.get(f"{TEST_URI}v1/lake/namespaces/sales/tables/orders", json=example_table_metadata_with_snapshot_v1_rest_json) + commit = requests_mock.post(f"{TEST_URI}v1/lake/transactions/commit", status_code=204) + catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN) + + catalog.commit_transactions(catalog.load_table(("sales", "orders")).transaction().set_properties(owner="sales")) + + assert commit.call_count == 1 + + +@pytest.mark.parametrize( + ("status_code", "exception"), + [ + (404, NoSuchTableError), + (409, CommitFailedException), + (500, CommitStateUnknownException), + (502, CommitStateUnknownException), + (504, CommitStateUnknownException), + ], +) +def test_commit_transactions_errors( + rest_mock: Mocker, + example_table_metadata_with_snapshot_v1_rest_json: dict[str, Any], + status_code: int, + exception: type[Exception], +) -> None: + catalog, (orders, items) = _load_tables_for_multi_table_commit(rest_mock, example_table_metadata_with_snapshot_v1_rest_json) + rest_mock.post( + f"{TEST_URI}v1/transactions/commit", + json={"error": {"message": "Commit failed", "type": "Error", "code": status_code}}, + status_code=status_code, + request_headers=TEST_HEADERS, + ) + orders_tx = orders.transaction().set_properties(owner="sales") + items_tx = items.transaction().set_properties(owner="sales") + + with pytest.raises(exception, match="Commit failed"): + catalog.commit_transactions(orders_tx, items_tx) + + # Neither table was reloaded, and the failed transactions can't be committed again. + assert rest_mock.call_count == 1 + 2 + 1 + with pytest.raises(RuntimeError, match="failed to commit"): + orders_tx.commit_transaction() + + +@pytest.mark.parametrize(("status_code", "files_deleted"), [(409, True), (500, False)]) +def test_commit_transactions_cleans_up_only_when_nothing_was_applied( + rest_mock: Mocker, + example_table_metadata_with_snapshot_v1_rest_json: dict[str, Any], + status_code: int, + files_deleted: bool, +) -> None: + catalog, (orders,) = _load_tables_for_multi_table_commit( + rest_mock, example_table_metadata_with_snapshot_v1_rest_json, names=("orders",) + ) + rest_mock.post( + f"{TEST_URI}v1/transactions/commit", + json={"error": {"message": "Commit failed", "type": "Error", "code": status_code}}, + status_code=status_code, + ) + orders_tx = orders.transaction().set_properties(owner="sales") + producer = mock.MagicMock() + orders_tx._snapshot_producers.append(producer) + + with pytest.raises((CommitFailedException, CommitStateUnknownException)): + catalog.commit_transactions(orders_tx) + + assert producer._clean_all_uncommitted.called is files_deleted + + +def test_commit_transactions_skips_transactions_without_changes( + rest_mock: Mocker, example_table_metadata_with_snapshot_v1_rest_json: dict[str, Any] +) -> None: + catalog, (orders, items) = _load_tables_for_multi_table_commit(rest_mock, example_table_metadata_with_snapshot_v1_rest_json) + commit = rest_mock.post(f"{TEST_URI}v1/transactions/commit", status_code=204) + + catalog.commit_transactions(orders.transaction(), items.transaction()) + assert commit.call_count == 0 + + catalog.commit_transactions(orders.transaction().set_properties(owner="sales"), items.transaction()) + assert [change["identifier"]["name"] for change in commit.last_request.json()["table-changes"]] == ["orders"] + + +def test_commit_transactions_sends_requirements_of_snapshot_operations_that_staged_nothing( + rest_mock: Mocker, example_table_metadata_with_snapshot_v1_rest_json: dict[str, Any] +) -> None: + catalog, (orders, items) = _load_tables_for_multi_table_commit(rest_mock, example_table_metadata_with_snapshot_v1_rest_json) + commit = rest_mock.post(f"{TEST_URI}v1/transactions/commit", status_code=204) + table_uuid = example_table_metadata_with_snapshot_v1_rest_json["metadata"]["table-uuid"] + snapshot_id = example_table_metadata_with_snapshot_v1_rest_json["metadata"]["current-snapshot-id"] + + orders_tx = orders.transaction().set_properties(owner="sales") + # A delete whose filter matched no rows registers its producer but stages no updates. + items_tx = items.transaction() + producer = mock.MagicMock(_target_branch="main", _starting_snapshot_id=snapshot_id) + items_tx._snapshot_producers.append(producer) + + catalog.commit_transactions(orders_tx, items_tx) + + assert commit.last_request.json()["table-changes"][1] == { + "identifier": {"namespace": ["sales"], "name": "items"}, + "requirements": [ + {"type": "assert-table-uuid", "uuid": table_uuid}, + {"type": "assert-ref-snapshot-id", "ref": "main", "snapshot-id": snapshot_id}, + ], + "updates": [], + } + assert items_tx._snapshot_producers == [] + + +def test_commit_transactions_rejects_invalid_input( + rest_mock: Mocker, example_table_metadata_with_snapshot_v1_rest_json: dict[str, Any] +) -> None: + catalog, (orders, items) = _load_tables_for_multi_table_commit(rest_mock, example_table_metadata_with_snapshot_v1_rest_json) + other_catalog = RestCatalog("other", uri=TEST_URI, token=TEST_TOKEN) + orders_tx = orders.transaction().set_properties(owner="sales") + + with pytest.raises(ValueError, match="At least one transaction"): + catalog.commit_transactions() + with pytest.raises(ValueError, match="only once"): + catalog.commit_transactions(orders_tx, orders_tx) + with pytest.raises(ValueError, match="More than one transaction changes table"): + catalog.commit_transactions(orders_tx, orders.transaction().set_properties(team="b")) + with pytest.raises(ValueError, match="belongs to a different catalog"): + other_catalog.commit_transactions(orders_tx) + assert not any(request.path.endswith("/transactions/commit") for request in rest_mock.request_history) + + +def test_commit_transactions_endpoint_not_supported( + requests_mock: Mocker, example_table_metadata_with_snapshot_v1_rest_json: dict[str, Any] +) -> None: + requests_mock.get( + f"{TEST_URI}v1/config", + json={"defaults": {}, "overrides": {}, "endpoints": [str(Capability.V1_LOAD_TABLE)]}, + status_code=200, + ) + requests_mock.get(f"{TEST_URI}v1/namespaces/sales/tables/orders", json=example_table_metadata_with_snapshot_v1_rest_json) + catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN) + orders_tx = catalog.load_table(("sales", "orders")).transaction().set_properties(owner="sales") + + with pytest.raises(NotImplementedError, match="transactions/commit"): + catalog.commit_transactions(orders_tx) diff --git a/tests/integration/test_rest_catalog.py b/tests/integration/test_rest_catalog.py index 05039a982e..af56926d89 100644 --- a/tests/integration/test_rest_catalog.py +++ b/tests/integration/test_rest_catalog.py @@ -17,12 +17,14 @@ # pylint:disable=redefined-outer-name import time +from typing import Any +import pyarrow as pa import pytest from pytest_lazy_fixtures import lf from pyiceberg.catalog.rest import RestCatalog -from pyiceberg.exceptions import NoSuchViewError +from pyiceberg.exceptions import CommitFailedException, NoSuchViewError from pyiceberg.schema import Schema from pyiceberg.view.metadata import SQLViewRepresentation, ViewVersion @@ -110,3 +112,79 @@ def test_load_view_with_table_ident( assert catalog.table_exists(table_identifier) with pytest.raises(NoSuchViewError): catalog.load_view(table_identifier) + + +def _create_tables_for_multi_table_commit(catalog: RestCatalog, database_name: str) -> tuple[pa.Table, list[Any]]: + if not catalog.namespace_exists(database_name): + catalog.create_namespace(database_name) + rows = pa.table({"id": pa.array([1, 2, 3], type=pa.int64()), "name": pa.array(["a", "b", "c"], type=pa.string())}) + tables = [] + for name in ("orders", "items"): + identifier = (database_name, name) + if catalog.table_exists(identifier): + catalog.drop_table(identifier) + tables.append(catalog.create_table(identifier, schema=rows.schema)) + return rows, tables + + +@pytest.mark.integration +@pytest.mark.parametrize("catalog", [lf("session_catalog")]) +def test_commit_transactions_appends_to_tables_atomically(catalog: RestCatalog, database_name: str) -> None: + rows, (orders, items) = _create_tables_for_multi_table_commit(catalog, database_name) + + orders_tx = orders.transaction() + orders_tx.append(rows) + items_tx = items.transaction() + items_tx.append(rows) + items_tx.set_properties(owner="sales") + catalog.commit_transactions(orders_tx, items_tx) + + for table in (orders, items): + assert table.current_snapshot() is not None + assert table.scan().to_arrow().num_rows == 3 + assert items.properties["owner"] == "sales" + + +@pytest.mark.integration +@pytest.mark.parametrize("catalog", [lf("session_catalog")]) +def test_commit_transactions_applies_nothing_on_conflict(catalog: RestCatalog, database_name: str) -> None: + rows, (orders, items) = _create_tables_for_multi_table_commit(catalog, database_name) + orders_tx = orders.transaction() + orders_tx.append(rows) + items_tx = items.transaction() + items_tx.append(rows) + + # Another writer commits to orders after the transactions were staged. + catalog.load_table(orders.name()).append(rows) + + with pytest.raises(CommitFailedException): + catalog.commit_transactions(orders_tx, items_tx) + + assert catalog.load_table(orders.name()).scan().to_arrow().num_rows == 3 + assert catalog.load_table(items.name()).current_snapshot() is None + + +@pytest.mark.integration +@pytest.mark.parametrize("catalog", [lf("session_catalog")]) +@pytest.mark.parametrize("concurrent_write", [False, True]) +@pytest.mark.filterwarnings("ignore:Delete operation did not match any records") +def test_commit_transactions_validates_a_delete_that_matched_nothing( + catalog: RestCatalog, database_name: str, concurrent_write: bool +) -> None: + rows, (orders, items) = _create_tables_for_multi_table_commit(catalog, database_name) + orders.append(rows) + orders_tx = orders.transaction() + orders_tx.delete("id = 100") + items_tx = items.transaction() + items_tx.append(rows) + assert not orders_tx._updates + + if concurrent_write: + # Another writer commits to orders after the delete was planned. + catalog.load_table(orders.name()).append(rows) + with pytest.raises(CommitFailedException): + catalog.commit_transactions(orders_tx, items_tx) + assert catalog.load_table(items.name()).current_snapshot() is None + else: + catalog.commit_transactions(orders_tx, items_tx) + assert catalog.load_table(items.name()).scan().to_arrow().num_rows == 3