From 6e28ce9f083f794b519c7132b04e7a9ddb7879ff Mon Sep 17 00:00:00 2001 From: Tom Owers Date: Sun, 2 Aug 2026 08:22:11 +0100 Subject: [PATCH 1/2] fix(data-imports): classify exhausted delta commit-conflict retries as transient maintenance noise Delta maintenance (compact/vacuum) already retries a `CommitFailedError` by refreshing the table and re-running the operation, but sustained contention from a concurrent maintenance pass on the same table can exhaust that retry budget. When it does, the error was being captured as a fresh bug instead of treated like the other known concurrent-maintenance races that self-heal on the next scheduled pass. Extend the transient-maintenance classifier to also cover the three delta-rs conflict-checker race variants (concurrent append / delete-read / delete-delete), while still capturing a real MetadataChanged/ProtocolChanged commit failure. branch: posthog-code/fix-delta-maintenance-commit-conflict-transient Generated-By: PostHog Code Task-Id: 954359e0-a2e9-4dcc-948b-de23803dec4a --- .../pipelines/core/delta/errors.py | 17 +++++++- .../core/delta/test/test_delta_errors.py | 39 +++++++++++++++++++ .../core/delta/test/test_maintenance.py | 10 +++++ 3 files changed, 65 insertions(+), 1 deletion(-) 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..3fd153b7bbd1 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,6 +50,36 @@ 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. @@ -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 From 01e4a866f47aac1f7fff53c7cde65892011ed75c Mon Sep 17 00:00:00 2001 From: Tom Owers Date: Sun, 2 Aug 2026 08:46:46 +0100 Subject: [PATCH 2/2] test(data-imports): rename delta maintenance test to reflect broader coverage The test now covers both the optimize-scan DeltaError signature and the CommitFailedError commit-conflict variants, so the old name (which only mentioned the scan case) made failures harder to interpret. Generated-By: PostHog Code Task-Id: 44141539-2628-4eee-9f52-3a60d2c54979 --- .../data_imports/pipelines/core/delta/test/test_delta_errors.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 3fd153b7bbd1..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 @@ -86,7 +86,7 @@ class TestIsTransientDeltaMaintenanceError: ("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