Fix case-insensitive delete regression and improve retry loop

Fix _build_delete_files_partition_predicate overwriting _case_sensitive
to True by passing the current value to delete_by_predicate. This
caused case-insensitive deletes to fail when _OverwriteFiles was used
with a user-specified predicate.

Move import random/time to file top level. Add total timeout
(commit.retry.total-timeout-ms) to the retry loop. Add comments for
intentional validation duplication and cached_property clearing.
Stabilize test_commit_retry_on_commit_failed by removing flaky
patch.object assertion.

Signed-off-by: Sotaro Hikita <bering1814@gmail.com>

Author: lawofcycles

pyiceberg/table/__init__.py

@@ -18,6 +18,8 @@
 
 import itertools
 import os
+import random
+import time
 import uuid
 import warnings
 from abc import ABC, abstractmethod
@@ -988,6 +990,13 @@ def commit_transaction(self) -> Table:
                 properties, TableProperties.COMMIT_MAX_RETRY_WAIT_MS, TableProperties.COMMIT_MAX_RETRY_WAIT_MS_DEFAULT
             )
             max_wait_ms = max_wait_val if max_wait_val is not None else TableProperties.COMMIT_MAX_RETRY_WAIT_MS_DEFAULT
+            total_timeout_val = property_as_int(
+                properties, TableProperties.COMMIT_TOTAL_RETRY_TIME_MS, TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT
+            )
+            total_timeout_ms = (
+                total_timeout_val if total_timeout_val is not None else TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT
+            )
+            start_time = time.monotonic()
 
             for attempt in range(num_retries + 1):
                 try:
@@ -999,10 +1008,9 @@ def commit_transaction(self) -> Table:
                     self._cleanup_uncommitted_manifests()
                     break
                 except CommitFailedException:
-                    if attempt == num_retries or not self._snapshot_producers:
+                    elapsed_ms = (time.monotonic() - start_time) * 1000
+                    if attempt == num_retries or not self._snapshot_producers or elapsed_ms >= total_timeout_ms:
                         raise
-                    import random
-                    import time
 
                     wait = min(min_wait_ms * (2**attempt), max_wait_ms)
                     jitter = random.uniform(0, 0.25 * wait)

pyiceberg/table/update/snapshot.py

@@ -418,7 +418,8 @@ def _build_delete_files_partition_predicate(self) -> None:
             self.delete_by_predicate(
                 self._transaction._build_partition_predicate(
                     partition_records=partition_records, schema=self.schema(), spec=self.spec(spec_id)
-                )
+                ),
+                self._case_sensitive,
             )
 
 
@@ -534,11 +535,19 @@ def files_affected(self) -> bool:
     def _refresh_for_retry(self) -> None:
         """Reset state for a retry attempt, clearing the cached delete computation."""
         super()._refresh_for_retry()
+        # Clear @cached_property by removing it from the instance __dict__.
+        # _compute_deletes depends on _parent_snapshot_id which changes on retry.
         if "_compute_deletes" in self.__dict__:
             del self.__dict__["_compute_deletes"]
 
     def _validate_concurrency(self) -> None:
-        """Validate that concurrent changes do not conflict with this delete."""
+        """Validate that concurrent changes do not conflict with this delete.
+
+        Note: This method is intentionally duplicated in _OverwriteFiles rather than
+        extracted to the base class. While the logic is currently identical, Java Iceberg's
+        BaseOverwriteFiles and BaseRowDelta have divergent validation. Keeping them separate
+        makes it easier to add RowDelta-specific validation in the future.
+        """
         from pyiceberg.table import TableProperties
         from pyiceberg.table.snapshots import IsolationLevel
         from pyiceberg.table.update.validate import (
@@ -746,7 +755,11 @@ def _get_entries(manifest: ManifestFile) -> list[ManifestEntry]:
             return []
 
     def _validate_concurrency(self) -> None:
-        """Validate that concurrent changes do not conflict with this overwrite."""
+        """Validate that concurrent changes do not conflict with this overwrite.
+
+        Note: See _DeleteFiles._validate_concurrency() for why this is intentionally
+        duplicated rather than extracted to the base class.
+        """
         from pyiceberg.table import TableProperties
         from pyiceberg.table.snapshots import IsolationLevel
         from pyiceberg.table.update.validate import (