fix(data-imports): classify exhausted delta commit-conflict retries as transient maintenance noise

products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta/errors.py

@@ -46,7 +46,22 @@ def is_transient_object_store_error(error: BaseException) -> bool:
 # read, which delta-rs surfaces as this DeltaError. The scan failing here means the optimize aborted
 # before committing anything — the table is left exactly as it was, just still fragmented — so this is
 # safe to skip and retry on the next maintenance pass, not a bug in our logic.
-TRANSIENT_DELTA_MAINTENANCE_ERRORS = ("Optimize selected-file scan failed",)
+#
+# The same concurrent maintenance pass can instead lose the race at commit time rather than during the
+# scan: `execute_with_conflict_retry` already refreshes and retries a `CommitFailedError`
+# (DELTA_MERGE_CONFLICT_RETRIES times), but sustained contention from another still-running pass can
+# exhaust that budget too, and the error then propagates out here. Delta's conflict checker raises this
+# for its three concurrent-writer race variants — ConcurrentAppend ("a concurrent transactions added new
+# data"), ConcurrentDeleteRead ("a concurrent transaction deleted data this operation read"), and
+# ConcurrentDeleteDelete ("a concurrent transaction deleted the same data") — all sharing the "a
+# concurrent transaction" substring below. A failed commit never partially applies, so the table is left
+# exactly as it was, same as the scan-failure case above: safe to skip and retry on the next maintenance
+# pass. Deliberately narrower than matching CommitFailedError outright — MetadataChanged/ProtocolChanged
+# commit failures aren't a same-pass race and should still be captured.
+TRANSIENT_DELTA_MAINTENANCE_ERRORS = (
+    "Optimize selected-file scan failed",
+    "Commit failed: a concurrent transaction",
+)
 
 
 def is_transient_delta_maintenance_error(error: BaseException) -> bool:

products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta/test/test_delta_errors.py

@@ -50,13 +50,43 @@ class TestIsTransientDeltaMaintenanceError:
                 ),
                 True,
             ),
+            # execute_with_conflict_retry already retries a CommitFailedError, but a still-running
+            # concurrent maintenance pass can exhaust that budget too — same race, just losing at
+            # commit time instead of during the scan. Covers all three delta-rs conflict-checker
+            # variants that share this class of race.
+            (
+                "compact_commit_conflict_concurrent_delete_read",
+                deltalake.exceptions.CommitFailedError(
+                    "Commit failed: a concurrent transaction deleted data this operation read."
+                ),
+                True,
+            ),
+            (
+                "vacuum_commit_conflict_concurrent_append",
+                deltalake.exceptions.CommitFailedError("Commit failed: a concurrent transactions added new data."),
+                True,
+            ),
+            (
+                "commit_conflict_concurrent_delete_delete",
+                deltalake.exceptions.CommitFailedError(
+                    "Commit failed: a concurrent transaction deleted the same data your transaction deletes."
+                ),
+                True,
+            ),
+            # A commit failure from a metadata/protocol change isn't the same-pass race above — a
+            # genuine conflict worth capturing, not silently retried away next pass.
+            (
+                "commit_conflict_metadata_changed",
+                deltalake.exceptions.CommitFailedError("Metadata changed since last commit."),
+                False,
+            ),
             # Other DeltaErrors are real failures (e.g. a genuinely corrupt log) and must still be captured.
             ("unrelated_delta_error", deltalake.exceptions.DeltaError("no protocol found in delta log"), False),
             # Same message shape but not the DeltaError type delta-rs actually raises for it.
             ("wrong_exception_type", RuntimeError("Optimize selected-file scan failed"), False),
         ]
     )
-    def test_matches_only_the_racy_optimize_scan_signature(self, _name: str, error: Exception, expected: bool):
+    def test_classifies_transient_delta_maintenance_errors(self, _name: str, error: Exception, expected: bool):
         assert is_transient_delta_maintenance_error(error) is expected
 
 
@@ -79,6 +109,15 @@ class TestIsTransientMaintenanceError:
                 deltalake.exceptions.DeltaError("Optimize selected-file scan failed while scanning data"),
                 True,
             ),
+            # A commit-conflict retry budget exhausted by sustained contention from another
+            # concurrent maintenance pass — see TestIsTransientDeltaMaintenanceError above.
+            (
+                "commit_conflict_retries_exhausted",
+                deltalake.exceptions.CommitFailedError(
+                    "Commit failed: a concurrent transaction deleted data this operation read."
+                ),
+                True,
+            ),
             ("genuine_bug", RuntimeError("maintenance blew up"), False),
         ]
     )

products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta/test/test_maintenance.py

@@ -352,6 +352,16 @@ async def test_watermark_persistence_gating(self, _name: str, returned_version:
             # A transient infra blip self-heals on the next scheduled pass and must not be promoted
             # into a fresh error-tracking issue.
             ("transient_blip_warned_only", OSError("Generic S3 error: Please reduce your request rate."), False),
+            # A commit conflict that exhausted execute_with_conflict_retry's budget (sustained
+            # contention from another still-running maintenance pass) is the same self-healing race,
+            # just losing at commit time instead of during compact's file scan.
+            (
+                "commit_conflict_retries_exhausted_warned_only",
+                deltalake.exceptions.CommitFailedError(
+                    "Commit failed: a concurrent transaction deleted data this operation read."
+                ),
+                False,
+            ),
         ]
     )
     @pytest.mark.asyncio