perf: cache Transaction.table_metadata between reads

`Transaction.table_metadata` replays every staged update through
`update_table_metadata`, whose last step is `model_copy(deep=True)`. The cost of
a single read therefore scales with the size of the metadata -- the snapshot
list in particular -- and callers read the property many times per operation.

Cache the result keyed on the identity of its two inputs. `_updates` is a tuple,
so every `+=` rebinds it to a new object, and `Table.metadata` is replaced
wholesale on refresh and commit; identity equality on both is therefore
sufficient for invalidation without any explicit cache-clearing at mutation
sites.

Carries forward #3302 by Ruiyang Wang, which was approved and then closed by the
stale bot. That PR predates #3301, whose
`test_snapshot_producer_bounded_metadata_access` pins the hoisted access count
with an equality assertion; the cache absorbs that access too, so the assertion
is relaxed to an upper bound.

Co-authored-by: Ruiyang Wang <rynewang@users.noreply.github.com>

Author: devseunggwan

pyiceberg/table/__init__.py

@@ -241,6 +241,7 @@ class Transaction:
     _autocommit: bool
     _updates: tuple[TableUpdate, ...]
     _requirements: tuple[TableRequirement, ...]
+    _table_metadata_cache: tuple[TableMetadata, tuple[TableUpdate, ...], TableMetadata] | None
 
     def __init__(self, table: Table, autocommit: bool = False):
         """Open a transaction to stage and commit changes to a table.
@@ -255,10 +256,21 @@ def __init__(self, table: Table, autocommit: bool = False):
         self._requirements = ()
         self._snapshot_producers: list[_SnapshotProducer[Any]] = []
         self._failed = False
+        self._table_metadata_cache = None
 
     @property
     def table_metadata(self) -> TableMetadata:
-        return update_table_metadata(self._table.metadata, self._updates)
+        base, updates = self._table.metadata, self._updates
+        # update_table_metadata replays every staged update via model_copy(deep=True);
+        # the cache is keyed on the identity of its inputs so it self-invalidates
+        # whenever _updates is reassigned (tuple += creates a new object) or the
+        # underlying table metadata is refreshed.
+        cached = self._table_metadata_cache
+        if cached is not None and cached[0] is base and cached[1] is updates:
+            return cached[2]
+        result = update_table_metadata(base, updates)
+        self._table_metadata_cache = (base, updates, result)
+        return result
 
     def __enter__(self) -> Transaction:
         """Start a transaction to update the table."""

tests/table/test_init.py

@@ -2010,3 +2010,54 @@ def _spy(*args: Any, **kwargs: Any) -> FileIO:
 
     assert seen_locations, "expected at least one load_file_io call"
     assert all(loc is not None for loc in seen_locations), f"load_file_io called without a location: {seen_locations}"
+
+
+def test_transaction_table_metadata_cached(table_v2: Table) -> None:
+    """Repeated reads of an unchanged transaction state recompute at most once.
+
+    `Transaction.table_metadata` replays every staged update through
+    `model_copy(deep=True)`, so the cost of a read scales with the size of the
+    metadata (the snapshot list in particular), not with the work being done.
+    """
+    from unittest import mock
+
+    from pyiceberg.table.update import SetPropertiesUpdate, update_table_metadata
+
+    with mock.patch("pyiceberg.table.update_table_metadata", wraps=update_table_metadata) as spy:
+        txn = table_v2.transaction()
+
+        first = txn.table_metadata
+        for _ in range(10):
+            assert txn.table_metadata is first
+        assert spy.call_count == 1, f"expected 1 recompute for repeated reads, got {spy.call_count}"
+
+        txn._stage((SetPropertiesUpdate(updates={"k": "v"}),))
+        second = txn.table_metadata
+        assert second is not first
+        assert second.properties["k"] == "v"
+        for _ in range(10):
+            assert txn.table_metadata is second
+        assert spy.call_count == 2, f"expected 2 recomputes after one staged update, got {spy.call_count}"
+
+
+def test_transaction_table_metadata_cached_with_updates_already_staged(table_v2: Table) -> None:
+    """The cache must still hold once `_updates` is non-empty.
+
+    An empty-`_updates` short circuit (`return self._table.metadata` when nothing
+    is staged) would leave this case uncovered, and it is the expensive one:
+    `CreateTableTransaction` seeds `_updates` with ~10 entries before any write,
+    so a create-then-append transaction replays all of them on every read.
+    """
+    from unittest import mock
+
+    from pyiceberg.table.update import SetPropertiesUpdate, update_table_metadata
+
+    txn = table_v2.transaction()
+    txn._stage((SetPropertiesUpdate(updates={"staged": "before"}),))
+
+    with mock.patch("pyiceberg.table.update_table_metadata", wraps=update_table_metadata) as spy:
+        first = txn.table_metadata
+        for _ in range(10):
+            assert txn.table_metadata is first
+        assert first.properties["staged"] == "before"
+        assert spy.call_count == 1, f"expected 1 recompute with updates already staged, got {spy.call_count}"

tests/table/test_snapshots.py

@@ -645,7 +645,13 @@ def summary_calls(n_files: int) -> int:
         spy.reset_mock()
         _MergeAppendFiles(operation=Operation.APPEND, transaction=txn, io=table_v2.io)
         merge_init = spy.call_count
-        assert merge_init - fast_init == 1, (
+        # Upper bound, not equality: `Transaction.table_metadata` caches on the identity of
+        # its inputs, so the second construction reads the same staged state and adds 0 calls.
+        # The trade-off is that this assertion no longer catches an un-hoisting of
+        # `_MergeAppendFiles.__init__` on its own — repeated reads of an unchanged state are
+        # free either way. What it still pins is that constructing the producer cannot start
+        # replaying updates per access again.
+        assert merge_init - fast_init <= 1, (
             f"_MergeAppendFiles.__init__ made {merge_init - fast_init} extra update_table_metadata "
-            "calls over its superclass; expected 1 (hoisted)"
+            "calls over its superclass; expected at most 1"
         )