abnobdoss commented on code in PR #3320:
URL: https://github.com/apache/iceberg-python/pull/3320#discussion_r3607157722
##########
pyiceberg/table/__init__.py:
##########
@@ -1043,17 +1084,82 @@ def commit_transaction(self) -> Table:
The table with the updates applied.
"""
if len(self._updates) > 0:
- self._requirements +=
(AssertTableUUID(uuid=self.table_metadata.table_uuid),)
- self._table._do_commit( # pylint: disable=W0212
- updates=self._updates,
- requirements=self._requirements,
+ properties = self._table.metadata.properties
+ num_retries: int = property_as_int( # type: ignore # The default
is set with non-None value.
+ properties, TableProperties.COMMIT_NUM_RETRIES,
TableProperties.COMMIT_NUM_RETRIES_DEFAULT
)
+ min_wait_ms: int = property_as_int( # type: ignore # The default
is set with non-None value.
+ properties, TableProperties.COMMIT_MIN_RETRY_WAIT_MS,
TableProperties.COMMIT_MIN_RETRY_WAIT_MS_DEFAULT
+ )
+ max_wait_ms: int = property_as_int( # type: ignore # The default
is set with non-None value.
+ properties, TableProperties.COMMIT_MAX_RETRY_WAIT_MS,
TableProperties.COMMIT_MAX_RETRY_WAIT_MS_DEFAULT
+ )
+ total_timeout_ms: int = property_as_int( # type: ignore # The
default is set with non-None value.
+ properties, TableProperties.COMMIT_TOTAL_RETRY_TIME_MS,
TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT
+ )
+ start_time = time.monotonic()
+ self._requirements +=
(AssertTableUUID(uuid=self.table_metadata.table_uuid),)
+
+ try:
+ for attempt in range(num_retries + 1):
+ try:
+ self._table._do_commit( # pylint: disable=W0212
+ updates=self._updates,
+ requirements=self._requirements,
+ )
+ self._cleanup_uncommitted_manifests()
+ break
+ except CommitFailedException:
+ elapsed_ms = (time.monotonic() - start_time) * 1000
+ if attempt == num_retries or not
self._snapshot_producers or elapsed_ms >= total_timeout_ms:
+ raise
+
+ wait = min(min_wait_ms * (2**attempt), max_wait_ms)
+ jitter = random.uniform(0, 0.1 * wait)
+ logger.warning(
+ "Commit failed due to a concurrent update,
retrying (%s/%s) in %s ms",
+ attempt + 1,
+ num_retries,
+ round(wait + jitter),
+ )
+ time.sleep((wait + jitter) / 1000.0)
+
+ self._table.refresh()
+ self._rebuild_snapshot_updates()
+ except Exception:
+ for producer in self._snapshot_producers:
+ producer._clean_all_uncommitted()
+ raise
self._updates = ()
Review Comment:
Is a Transaction object meant to be safe to use after `commit_transaction`?
If it is then I think clearing `_snapshot_producers` on success and resetting
or invalidating the state on failure is missing.
Here are some test scenarios:
**Scenario 1:** a commit fails once for a transient reason (the kinds traced
in my `except Exception` comment), and the caller retries the same transaction.
The staged `AddSnapshotUpdate` survived the failure, its manifest list did not,
and the catalog accepts it:
```python
import pyarrow as pa
import pytest
from unittest.mock import patch
from pyiceberg.catalog.memory import InMemoryCatalog
from pyiceberg.exceptions import CommitFailedException
from pyiceberg.schema import Schema
from pyiceberg.types import LongType, NestedField
def test_reusing_a_failed_transaction_cannot_publish_deleted_files(tmp_path):
catalog = InMemoryCatalog("test", warehouse=tmp_path.as_uri())
catalog.create_namespace("default")
table = catalog.create_table(
"default.t",
schema=Schema(NestedField(1, "x", LongType(), required=False)),
properties={"commit.retry.num-retries": "0"},
)
table.append(pa.table({"x": [1]}))
tx = table.transaction()
tx.append(pa.table({"x": [2]}))
# The commit fails once for a transient reason.
with patch.object(catalog, "commit_table",
side_effect=CommitFailedException("transient")):
with pytest.raises(CommitFailedException):
tx.commit_transaction()
# The caller retries the same transaction, which was safe before this PR.
try:
tx.commit_transaction()
except Exception:
# Refusing reuse would be fine, as long as the table is untouched.
assert catalog.load_table("default.t").scan().to_arrow().to_pylist()
== [{"x": 1}]
return
# If the re-commit is accepted, the table must still be readable.
rows =
sorted(catalog.load_table("default.t").scan().to_arrow()["x"].to_pylist())
assert rows == [1, 2]
```
Fails with `FileNotFoundError` scanning the table: the catalog head points
at a manifest list the cleanup deleted.
**Scenario 2:** no failure, but reuse after a successful commit plus one
concurrent writer:
```python
def
test_reusing_a_committed_transaction_does_not_damage_the_first_commit(tmp_path):
catalog = InMemoryCatalog("test", warehouse=tmp_path.as_uri())
catalog.create_namespace("default")
table = catalog.create_table(
"default.t",
schema=Schema(NestedField(1, "x", LongType(), required=False)),
properties={"commit.retry.min-wait-ms": "1",
"commit.retry.max-wait-ms": "2"},
)
tx = table.transaction()
tx.append(pa.table({"x": [1, 2, 3]}))
tx.commit_transaction()
tx.append(pa.table({"x": [10, 20]}))
catalog.load_table("default.t").append(pa.table({"x": [100]})) # forces
a retry
tx.commit_transaction()
rows =
sorted(catalog.load_table("default.t").scan().to_arrow()["x"].to_pylist())
assert rows == [1, 2, 3, 10, 20, 100]
```
Also fails with `FileNotFoundError`: the retry replays the first producer,
which is still registered, so its batch is committed a second time and its
already-committed manifests get moved into the uncommitted list and deleted by
the post-success cleanup.
##########
pyiceberg/table/update/snapshot.py:
##########
@@ -499,8 +637,19 @@ def files_affected(self) -> bool:
"""Indicate if any manifest-entries can be dropped."""
return len(self._deleted_entries()) > 0
+ def _refresh_for_retry(self) -> None:
Review Comment:
Is it safe to clear the computed deletes on retry? Here's a scenario: a
writer plans `delete("x == 1")`, and before it commits, someone else appends a
file whose rows all match that predicate. The commit conflicts, the retry
replans the delete against the refreshed head, and the concurrent file is now a
delete target, so it gets dropped whole. The writer deletes rows it never saw,
under the isolation setting whose promise is that concurrent appends are
preserved:
```python
import pyarrow as pa
from pyiceberg.catalog.memory import InMemoryCatalog
from pyiceberg.schema import Schema
from pyiceberg.types import LongType, NestedField
def
test_snapshot_isolation_delete_does_not_remove_rows_it_never_saw(tmp_path):
catalog = InMemoryCatalog("test", warehouse=tmp_path.as_uri())
catalog.create_namespace("default")
table = catalog.create_table(
"default.t",
schema=Schema(NestedField(1, "x", LongType(), required=False)),
properties={
"write.delete.isolation-level": "snapshot",
"commit.retry.min-wait-ms": "1",
"commit.retry.max-wait-ms": "2",
},
)
table.append(pa.table({"x": [0, 1]}))
stale = catalog.load_table("default.t")
tx = stale.transaction()
tx.delete("x == 1")
# Lands after the delete was planned. The stale writer never saw this
row.
catalog.load_table("default.t").append(pa.table({"x": [1]}))
tx.commit_transaction()
rows =
sorted(catalog.load_table("default.t").scan().to_arrow()["x"].to_pylist())
assert rows == [0, 1] # fails: the concurrent row is gone
```
Maybe that is the intent, and a retried delete is supposed to re-execute
against the new table state and remove the concurrent row too. That reading
seems defensible as well, but I do not think the current behavior lands on it
either: only whole-file drops are replanned, while partial-file rewrites are
planned once in `Transaction.delete` and not recomputed. So a single concurrent
commit can end up half incorporated:
```python
import uuid
from pyiceberg.io.pyarrow import _dataframe_to_data_files
def test_retried_delete_treats_a_concurrent_commit_atomically(tmp_path):
catalog = InMemoryCatalog("test", warehouse=tmp_path.as_uri())
catalog.create_namespace("default")
table = catalog.create_table(
"default.t",
schema=Schema(NestedField(1, "x", LongType(), required=False)),
properties={
"write.delete.isolation-level": "snapshot",
"commit.retry.min-wait-ms": "1",
"commit.retry.max-wait-ms": "2",
},
)
table.append(pa.table({"x": [0, 1]}))
stale = catalog.load_table("default.t")
tx = stale.transaction()
tx.delete("x == 1")
# One concurrent commit carrying two files: [1] wholly matches, [1, 5]
partially.
b = catalog.load_table("default.t")
btx = b.transaction()
with btx.update_snapshot().fast_append() as append:
for df in (pa.table({"x": [1]}), pa.table({"x": [1, 5]})):
for f in _dataframe_to_data_files(
table_metadata=btx.table_metadata, io=b.io,
write_uuid=uuid.uuid4(), df=df
):
append.append_data_file(f)
btx.commit_transaction()
tx.commit_transaction()
final =
sorted(catalog.load_table("default.t").scan().to_arrow()["x"].to_pylist())
# The concurrent commit is atomic, so the delete may apply to all of it
or none of it.
assert final in ([0, 1, 1, 5], [0, 5]), f"half of the concurrent commit
was deleted: {final}"
```
The second test fails with `[0, 1, 5]`: one of the commit's matching rows
deleted, the other preserved, which I do not think either reading intends.
Since users cannot see or control which file a row lands in, the outcome of the
race is effectively random.
(`test_snapshot_isolation_allows_concurrent_append_delete` in this PR is the
partial-match instance of the same race and expects the row to survive.)
For reference, Java fixes the set of files a delete operates on at planning
time; retries revalidate but never replan, so concurrent commits are
consistently preserved under snapshot isolation. Would freezing the planned
file set here be an option? As far as I can tell it would make all of these
tests pass, including the existing one.
##########
pyiceberg/table/update/snapshot.py:
##########
@@ -91,19 +97,50 @@ def _new_manifest_list_file_name(snapshot_id: int, attempt:
int, commit_uuid: uu
return f"snap-{snapshot_id}-{attempt}-{commit_uuid}.avro"
+@dataclass(frozen=True)
+class CommitWindow:
+ """Tracks the commit range to validate against during retry.
+
+ base: The snapshot when the operation began (exclusive lower bound, fixed
across retries).
+ head: The branch HEAD snapshot after metadata refresh (inclusive upper
bound).
+ """
+
+ base: Snapshot | None
+ head: Snapshot | None
+
+ @classmethod
+ def resolve(cls, metadata: TableMetadata, base_id: int | None, branch: str
| None) -> CommitWindow:
+ """Resolve a CommitWindow from metadata, starting snapshot ID, and
target branch."""
+ head = metadata.snapshot_by_name(branch)
+ base = metadata.snapshot_by_id(base_id) if base_id is not None else
None
+ if base_id is not None and base is None:
+ raise ValidationException(f"Cannot find starting snapshot
{base_id}")
+ return cls(base=base, head=head)
+
+ def is_empty(self) -> bool:
Review Comment:
Is `base is None` safe to treat as an empty window? It also means the writer
started on a table with no snapshots, and if another writer lands the first
snapshot in the meantime, the retry skips validation entirely. Two jobs racing
to initialize the same table is a common bootstrap pattern:
```python
import pyarrow as pa
import pytest
from pyiceberg.catalog.memory import InMemoryCatalog
from pyiceberg.exceptions import ValidationException
from pyiceberg.schema import Schema
from pyiceberg.types import LongType, NestedField
def test_writer_that_started_on_an_empty_table_still_validates(tmp_path):
catalog = InMemoryCatalog("test", warehouse=tmp_path.as_uri())
catalog.create_namespace("default")
table = catalog.create_table(
"default.t",
schema=Schema(NestedField(1, "x", LongType(), required=False)),
properties={"commit.retry.min-wait-ms": "1",
"commit.retry.max-wait-ms": "2"},
)
stale = catalog.load_table("default.t")
tx = stale.transaction()
with pytest.warns(UserWarning): # the delete matches nothing on an
empty table
tx.overwrite(pa.table({"x": [2]}), overwrite_filter="x == 1")
# The first ever snapshot lands concurrently, with a row matching the
filter.
catalog.load_table("default.t").append(pa.table({"x": [1]}))
with pytest.raises(ValidationException):
tx.commit_transaction()
```
Fails because no exception is raised: the commit goes through unvalidated
and the table ends up as just `[2]`. The concurrent writer's row matched the
filter and was deleted, with no error on either side. Java treats a null
starting snapshot as validate against the entire history
(MergingSnapshotProducer walks every ancestor of the head), and treating a None
base the same way here would close this.
##########
pyiceberg/table/__init__.py:
##########
@@ -1043,17 +1084,82 @@ def commit_transaction(self) -> Table:
The table with the updates applied.
"""
if len(self._updates) > 0:
- self._requirements +=
(AssertTableUUID(uuid=self.table_metadata.table_uuid),)
- self._table._do_commit( # pylint: disable=W0212
- updates=self._updates,
- requirements=self._requirements,
+ properties = self._table.metadata.properties
+ num_retries: int = property_as_int( # type: ignore # The default
is set with non-None value.
+ properties, TableProperties.COMMIT_NUM_RETRIES,
TableProperties.COMMIT_NUM_RETRIES_DEFAULT
)
+ min_wait_ms: int = property_as_int( # type: ignore # The default
is set with non-None value.
+ properties, TableProperties.COMMIT_MIN_RETRY_WAIT_MS,
TableProperties.COMMIT_MIN_RETRY_WAIT_MS_DEFAULT
+ )
+ max_wait_ms: int = property_as_int( # type: ignore # The default
is set with non-None value.
+ properties, TableProperties.COMMIT_MAX_RETRY_WAIT_MS,
TableProperties.COMMIT_MAX_RETRY_WAIT_MS_DEFAULT
+ )
+ total_timeout_ms: int = property_as_int( # type: ignore # The
default is set with non-None value.
+ properties, TableProperties.COMMIT_TOTAL_RETRY_TIME_MS,
TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT
+ )
+ start_time = time.monotonic()
+ self._requirements +=
(AssertTableUUID(uuid=self.table_metadata.table_uuid),)
+
+ try:
+ for attempt in range(num_retries + 1):
+ try:
+ self._table._do_commit( # pylint: disable=W0212
+ updates=self._updates,
+ requirements=self._requirements,
+ )
+ self._cleanup_uncommitted_manifests()
+ break
+ except CommitFailedException:
+ elapsed_ms = (time.monotonic() - start_time) * 1000
+ if attempt == num_retries or not
self._snapshot_producers or elapsed_ms >= total_timeout_ms:
+ raise
+
+ wait = min(min_wait_ms * (2**attempt), max_wait_ms)
+ jitter = random.uniform(0, 0.1 * wait)
+ logger.warning(
+ "Commit failed due to a concurrent update,
retrying (%s/%s) in %s ms",
+ attempt + 1,
+ num_retries,
+ round(wait + jitter),
+ )
+ time.sleep((wait + jitter) / 1000.0)
+
+ self._table.refresh()
+ self._rebuild_snapshot_updates()
+ except Exception:
Review Comment:
I'm a bit wary of the bare `except Exception` here, so I traced what can
actually reach it. I count three groups with very different meanings:
1. The commit definitely did not happen: `ValidationException` from the
conflict checks during rebuild, IO errors while writing the next attempt's
manifests.
2. The outcome is unknown: `CommitStateUnknownException` (REST raises it for
500/502/504, which is what a gateway timeout looks like when the backend
finished the commit), raw `requests` connection errors and timeouts (nothing in
the client catches these), and SQLAlchemy `OperationalError`, which covers both
"database locked" and "connection dropped after COMMIT".
3. The commit definitely happened: a pydantic `ValidationError` while
parsing the 200 commit response, and the `ValueError` from `_check_uuid`, both
raised after `commit_table` already succeeded.
The handler treats all three the same way and deletes every written manifest
and manifest list. For groups 2 and 3 those files can already be referenced by
the catalog's current snapshot, and deleting them makes the table unreadable
for every reader, not just this writer.
```python
import pyarrow as pa
import pytest
from unittest.mock import patch
from pyiceberg.catalog.memory import InMemoryCatalog
from pyiceberg.exceptions import CommitStateUnknownException
from pyiceberg.schema import Schema
from pyiceberg.types import LongType, NestedField
def test_unknown_commit_outcome_keeps_the_committed_files(tmp_path):
catalog = InMemoryCatalog("test", warehouse=tmp_path.as_uri())
catalog.create_namespace("default")
table = catalog.create_table(
"default.t", schema=Schema(NestedField(1, "x", LongType(),
required=False))
)
real_commit = catalog.commit_table
def commit_then_lose_response(*args, **kwargs):
real_commit(*args, **kwargs)
raise CommitStateUnknownException("response lost after the commit
landed")
with patch.object(catalog, "commit_table",
side_effect=commit_then_lose_response):
with pytest.raises(CommitStateUnknownException):
table.append(pa.table({"x": [1]}))
# The commit landed, so the table must still be readable.
assert catalog.load_table("default.t").scan().to_arrow().to_pylist() ==
[{"x": 1}]
```
Fails with `FileNotFoundError` on the committed snapshot's manifest list.
I think the safe rule is to invert the default: only delete files for
exceptions that guarantee the commit did not happen (`CommitFailedException`,
`ValidationException`) and keep them for everything else. Orphan files are
recoverable, a deleted manifest list behind a live snapshot is not. That
matches Java, which rethrows `CommitStateUnknownException` before any cleanup
runs (SnapshotProducer.java).
Longer term, group 2 can only be classified truthfully by asking the catalog
whether the commit landed, the way Java's checkCommitStatus polls for its own
metadata location. That needs a stable snapshot id across attempts, which I
have raised separately.
##########
pyiceberg/table/__init__.py:
##########
@@ -1043,17 +1084,82 @@ def commit_transaction(self) -> Table:
The table with the updates applied.
"""
if len(self._updates) > 0:
- self._requirements +=
(AssertTableUUID(uuid=self.table_metadata.table_uuid),)
- self._table._do_commit( # pylint: disable=W0212
- updates=self._updates,
- requirements=self._requirements,
+ properties = self._table.metadata.properties
+ num_retries: int = property_as_int( # type: ignore # The default
is set with non-None value.
+ properties, TableProperties.COMMIT_NUM_RETRIES,
TableProperties.COMMIT_NUM_RETRIES_DEFAULT
)
+ min_wait_ms: int = property_as_int( # type: ignore # The default
is set with non-None value.
+ properties, TableProperties.COMMIT_MIN_RETRY_WAIT_MS,
TableProperties.COMMIT_MIN_RETRY_WAIT_MS_DEFAULT
+ )
+ max_wait_ms: int = property_as_int( # type: ignore # The default
is set with non-None value.
+ properties, TableProperties.COMMIT_MAX_RETRY_WAIT_MS,
TableProperties.COMMIT_MAX_RETRY_WAIT_MS_DEFAULT
+ )
+ total_timeout_ms: int = property_as_int( # type: ignore # The
default is set with non-None value.
+ properties, TableProperties.COMMIT_TOTAL_RETRY_TIME_MS,
TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT
+ )
+ start_time = time.monotonic()
+ self._requirements +=
(AssertTableUUID(uuid=self.table_metadata.table_uuid),)
+
+ try:
+ for attempt in range(num_retries + 1):
Review Comment:
With `commit.retry.num-retries=-1` this is `range(0)`: no commit is
attempted, no error is raised, the staged updates are cleared after the loop,
and the call returns as success. Since this is a table property, one writer
setting it turns every writer's commits into silent no-ops. Some tools use -1
to mean retry forever, so it is a plausible value.
```python
import pyarrow as pa
from pyiceberg.catalog.memory import InMemoryCatalog
from pyiceberg.schema import Schema
from pyiceberg.types import LongType, NestedField
def test_invalid_retry_count_does_not_silently_skip_the_commit(tmp_path):
catalog = InMemoryCatalog("test", warehouse=tmp_path.as_uri())
catalog.create_namespace("default")
table = catalog.create_table(
"default.t",
schema=Schema(NestedField(1, "x", LongType(), required=False)),
properties={"commit.retry.num-retries": "-1"},
)
table.append(pa.table({"x": [1]}))
assert catalog.load_table("default.t").scan().to_arrow().to_pylist() ==
[{"x": 1}]
```
Fails with an empty table. Validating the property when read (raise on
negatives, or clamp to zero) makes this a loud config error instead of silent
data loss. Java runs one attempt even for a negative value.
##########
pyiceberg/table/update/snapshot.py:
##########
@@ -353,11 +400,101 @@ def new_manifest_output(self) -> OutputFile:
location_provider = self._transaction._table.location_provider()
file_name =
_new_manifest_file_name(num=next(self._manifest_num_counter),
commit_uuid=self.commit_uuid)
file_path = location_provider.new_metadata_location(file_name)
+ self._written_manifests.append(file_path)
return self._io.new_output(file_path)
def fetch_manifest_entry(self, manifest: ManifestFile, discard_deleted:
bool = True) -> list[ManifestEntry]:
return manifest.fetch_manifest_entry(io=self._io,
discard_deleted=discard_deleted)
+ def commit(self) -> None:
+ self._transaction._register_snapshot_producer(self)
+ super().commit()
+
+ def _cleanup_uncommitted(self) -> None:
+ """Delete manifest files and manifest lists from failed retry
attempts."""
+ for path in self._uncommitted_manifests:
+ try:
+ self._io.delete(path)
+ except Exception:
+ logger.warning("Failed to delete uncommitted manifest: %s",
path, exc_info=True)
+ self._uncommitted_manifests.clear()
+ # Delete all manifest lists except the last one (which the committed
snapshot references)
+ if len(self._written_manifest_lists) > 1:
+ for path in self._written_manifest_lists[:-1]:
+ try:
+ self._io.delete(path)
+ except Exception:
+ logger.warning("Failed to delete uncommitted manifest
list: %s", path, exc_info=True)
+ self._written_manifest_lists = self._written_manifest_lists[-1:]
+
+ def _clean_all_uncommitted(self) -> None:
+ """Clean up all manifests and manifest lists on abort."""
+ for path in itertools.chain(self._uncommitted_manifests,
self._written_manifests):
+ try:
+ self._io.delete(path)
+ except Exception:
+ logger.warning("Failed to delete uncommitted manifest: %s",
path, exc_info=True)
+ for path in self._written_manifest_lists:
+ try:
+ self._io.delete(path)
+ except Exception:
+ logger.warning("Failed to delete uncommitted manifest list:
%s", path, exc_info=True)
+ self._uncommitted_manifests.clear()
+ self._written_manifests.clear()
+ self._written_manifest_lists.clear()
+
+ def _refresh_for_retry(self) -> None:
+ """Reset state for a retry attempt with refreshed metadata."""
+ self._uncommitted_manifests.extend(self._written_manifests)
+ self._written_manifests.clear()
+ self._parent_snapshot_id = self._current_branch_head_id()
+ self._snapshot_id = self._transaction.table_metadata.new_snapshot_id()
Review Comment:
On a similar note to my comment on the bare `except Exception` in
`Transaction.commit_transaction`:
A commit call over a network has three possible outcomes: it applied and we
got the response, it did not apply and we got an error, or it applied and we
still got an error because the response was lost. The retry loop only models
the first two. In the third case it refreshes and commits the same data again,
like retrying a payment without an idempotency key.
The snapshot id could be that idempotency key, but this line mints a new one
on every attempt, so the client has no way to recognize its own commit in the
refreshed metadata. The third case is not hypothetical: Glue's boto client
silently resends `UpdateTable` on connection errors and the resend's version
conflict is reported as `CommitFailedException`, and a retrying proxy in front
of a REST catalog does the same via a 409.
```python
import pyarrow as pa
from unittest.mock import patch
from pyiceberg.catalog.memory import InMemoryCatalog
from pyiceberg.exceptions import CommitFailedException
from pyiceberg.schema import Schema
from pyiceberg.types import LongType, NestedField
def
test_commit_that_landed_but_was_reported_failed_is_not_committed_twice(tmp_path):
catalog = InMemoryCatalog("test", warehouse=tmp_path.as_uri())
catalog.create_namespace("default")
table = catalog.create_table(
"default.t",
schema=Schema(NestedField(1, "x", LongType(), required=False)),
properties={"commit.retry.min-wait-ms": "1",
"commit.retry.max-wait-ms": "2"},
)
real_commit = catalog.commit_table
calls = []
def commit_then_report_conflict(*args, **kwargs):
result = real_commit(*args, **kwargs)
calls.append(1)
if len(calls) == 1:
raise CommitFailedException("transport layer retried; first
response was lost")
return result
with patch.object(catalog, "commit_table",
side_effect=commit_then_report_conflict):
table.append(pa.table({"x": [1]}))
assert catalog.load_table("default.t").scan().to_arrow().to_pylist() ==
[{"x": 1}]
```
Fails with `FileNotFoundError`: the data is committed twice, and the
post-success cleanup then deletes manifests the first committed snapshot
references.
Java generates the snapshot id once, reuses it for every attempt, and checks
the refreshed metadata for it before re-committing (SnapshotProducer.java).
Doing the same here would close this for every catalog at once.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
