diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta/errors.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta/errors.py index af5744c26f57..c050a360070d 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta/errors.py +++ b/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: diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta/test/test_delta_errors.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta/test/test_delta_errors.py index 161c6307c5b9..d60d63522dbc 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta/test/test_delta_errors.py +++ b/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), ] ) diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta/test/test_maintenance.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta/test/test_maintenance.py index a97cbbb19daf..5834a64b6ba4 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta/test/test_maintenance.py +++ b/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