Expire snapshot mutability issue (#2430)

<!--
Thanks for opening a pull request!
-->

<!-- In the case this PR will resolve an issue, please replace
${GITHUB_ISSUE_ID} below with the actual Github issue id. -->
Closes #2409, and partially closes #2427

# Rationale for this change

This PR fixes a critical thread safety issue in the `ExpireSnapshots`
class where concurrent snapshot expiration operations on different
tables would share snapshot IDs, causing operations to fail with
"snapshot does not exist" errors.

**Root Cause:** 
The `ExpireSnapshots` class had class-level attributes
(`_snapshot_ids_to_expire`, `_updates`, `_requirements`) that were
shared across all instances. When multiple threads created different
`ExpireSnapshots` instances, they all shared the same underlying `set()`
object for tracking snapshot IDs.

**Impact:**
- Thread 1: `table1.expire_snapshots().by_id(1001)` adds `1001` to
shared set
- Thread 2: `table2.expire_snapshots().by_id(2001)` adds `2001` to same
shared set
- Result: Both threads have `{1001, 2001}` and try to expire snapshot
`1001` from `table2`, causing failure

**Solution:**
Moved the shared class-level attributes to instance-level attributes in
the `__init__` method, ensuring each `ExpireSnapshots` instance has its
own isolated state.

## Are these changes tested?

> 📢 🔥 Big shout-out to @QlikFrederic, as the testing methodology was
largely derived from the testing and analysis done by the user! 🔥 📢

Yes, comprehensive test coverage has been added:

- **`test_thread_safety_fix()`** - Verifies that different
ExpireSnapshots instances have separate snapshot sets
- **`test_concurrent_operations()`** - Tests concurrent operations don't
contaminate each other
- **`test_concurrent_different_tables_expiration()`** - Reproduces the
exact scenario from GitHub issue #2409
- **`test_concurrent_same_table_different_snapshots()`** - Tests
concurrent operations on the same table
- **`test_cross_table_snapshot_id_isolation()`** - Validates no
cross-contamination of snapshot IDs between tables
- **`test_batch_expire_snapshots()`** - Tests batch expiration
operations in threaded environments

All existing tests continue to pass, ensuring no regression in
functionality.

## Are there any user-facing changes?

**No breaking changes.** The public API remains identical:
- All existing `ExpireSnapshots` methods work the same way
- Method signatures are unchanged
- Behavior is identical except for the thread safety fix

**Behavioral improvement:**
- Concurrent `expire_snapshots()` operations on different tables now
work correctly
- No more "snapshot does not exist" errors when using ExpireSnapshots in
multi-threaded environments

This is a pure bug fix with no user-facing API changes.

---------

Co-authored-by: Fokko Driesprong <[email protected]>

Author: ForeverAngry

pyiceberg/table/update/snapshot.py

@@ -924,9 +924,15 @@ class ExpireSnapshots(UpdateTableMetadata["ExpireSnapshots"]):
     Pending changes are applied on commit.
     """
 
-    _snapshot_ids_to_expire: Set[int] = set()
-    _updates: Tuple[TableUpdate, ...] = ()
-    _requirements: Tuple[TableRequirement, ...] = ()
+    _updates: Tuple[TableUpdate, ...]
+    _requirements: Tuple[TableRequirement, ...]
+    _snapshot_ids_to_expire: Set[int]
+
+    def __init__(self, transaction: Transaction) -> None:
+        super().__init__(transaction)
+        self._updates = ()
+        self._requirements = ()
+        self._snapshot_ids_to_expire = set()
 
     def _commit(self) -> UpdatesAndRequirements:
         """

tests/table/test_expire_snapshots.py

@@ -14,13 +14,16 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-import datetime
-from unittest.mock import MagicMock
+import threading
+from datetime import datetime, timedelta
+from typing import Dict
+from unittest.mock import MagicMock, Mock
 from uuid import uuid4
 
 import pytest
 
 from pyiceberg.table import CommitTableResponse, Table
+from pyiceberg.table.update.snapshot import ExpireSnapshots
 
 
 def test_cannot_expire_protected_head_snapshot(table_v2: Table) -> None:
@@ -143,7 +146,7 @@ def test_expire_snapshots_by_timestamp_skips_protected(table_v2: Table) -> None:
     table_v2.catalog = MagicMock()
 
     # Attempt to expire all snapshots before a future timestamp (so both are candidates)
-    future_datetime = datetime.datetime.now() + datetime.timedelta(days=1)
+    future_datetime = datetime.now() + timedelta(days=1)
 
     # Mock the catalog's commit_table to return the current metadata (simulate no change)
     mock_response = CommitTableResponse(
@@ -223,3 +226,57 @@ def test_expire_snapshots_by_ids(table_v2: Table) -> None:
     assert EXPIRE_SNAPSHOT_1 not in remaining_snapshots
     assert EXPIRE_SNAPSHOT_2 not in remaining_snapshots
     assert len(table_v2.metadata.snapshots) == 1
+
+
+def test_thread_safety_fix() -> None:
+    """Test that ExpireSnapshots instances have isolated state."""
+    # Create two ExpireSnapshots instances
+    expire1 = ExpireSnapshots(Mock())
+    expire2 = ExpireSnapshots(Mock())
+
+    # Verify they have separate snapshot sets (this was the bug!)
+    # Before fix: both would have the same id (shared class attribute)
+    # After fix: they should have different ids (separate instance attributes)
+    assert id(expire1._snapshot_ids_to_expire) != id(expire2._snapshot_ids_to_expire), (
+        "ExpireSnapshots instances are sharing the same snapshot set - thread safety bug still exists"
+    )
+
+    # Test that modifications to one don't affect the other
+    expire1._snapshot_ids_to_expire.add(1001)
+    expire2._snapshot_ids_to_expire.add(2001)
+
+    # Verify no cross-contamination of snapshot IDs
+    assert 2001 not in expire1._snapshot_ids_to_expire, "Snapshot IDs are leaking between instances"
+    assert 1001 not in expire2._snapshot_ids_to_expire, "Snapshot IDs are leaking between instances"
+
+
+def test_concurrent_operations() -> None:
+    """Test concurrent operations with separate ExpireSnapshots instances."""
+    results: Dict[str, set[int]] = {"expire1_snapshots": set(), "expire2_snapshots": set()}
+
+    def worker1() -> None:
+        expire1 = ExpireSnapshots(Mock())
+        expire1._snapshot_ids_to_expire.update([1001, 1002, 1003])
+        results["expire1_snapshots"] = expire1._snapshot_ids_to_expire.copy()
+
+    def worker2() -> None:
+        expire2 = ExpireSnapshots(Mock())
+        expire2._snapshot_ids_to_expire.update([2001, 2002, 2003])
+        results["expire2_snapshots"] = expire2._snapshot_ids_to_expire.copy()
+
+    # Run both workers concurrently
+    thread1 = threading.Thread(target=worker1)
+    thread2 = threading.Thread(target=worker2)
+
+    thread1.start()
+    thread2.start()
+
+    thread1.join()
+    thread2.join()
+
+    # Check for cross-contamination
+    expected_1 = {1001, 1002, 1003}
+    expected_2 = {2001, 2002, 2003}
+
+    assert results["expire1_snapshots"] == expected_1, "Worker 1 snapshots contaminated"
+    assert results["expire2_snapshots"] == expected_2, "Worker 2 snapshots contaminated"