fix(data-imports): retry delta vacuum on commit conflict

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

@@ -64,8 +64,13 @@ def __init__(self, table: "DeltaTableHelper") -> None:
 
     async def _vacuum(self, table: deltalake.DeltaTable) -> None:
         await self._logger.adebug("Vacuuming table...")
-        vacuum_stats = await asyncio.to_thread(
-            table.vacuum, retention_hours=24, enforce_retention_duration=False, dry_run=False
+        # vacuum() commits a REMOVE of tombstoned files, so it's just as subject to delta-rs's
+        # conflict checker as merge/optimize.compact — see execute_with_conflict_retry.
+        vacuum_stats = await execute_with_conflict_retry(
+            table,
+            lambda: table.vacuum(retention_hours=24, enforce_retention_duration=False, dry_run=False),
+            "vacuum_table",
+            self._logger,
         )
         await self._logger.adebug(json.dumps(vacuum_stats))
 

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

@@ -1,12 +1,15 @@
 import asyncio
 from collections.abc import Callable
+from typing import TypeVar
 
 from django.conf import settings
 
 import deltalake
 import deltalake.exceptions
 from structlog.types import FilteringBoundLogger
 
+T = TypeVar("T")
+
 
 def delta_merge_spill_kwargs() -> dict[str, int]:
     """delta-rs `merge` kwargs that let DataFusion spill to disk instead of OOMing on large merges.
@@ -27,8 +30,9 @@ def delta_merge_spill_kwargs() -> dict[str, int]:
 
 
 # Delta's conflict checker raises CommitFailedError the moment a concurrent commit invalidates what
-# a committing operation read — a merge predicate, or optimize.compact's file-rewrite plan — unlike a
-# plain version-bump race, delta-rs does not consume max_commit_retries or retry this itself (see
+# a committing operation read — a merge predicate, optimize.compact's file-rewrite plan, or vacuum's
+# tombstone list — unlike a plain version-bump race, delta-rs does not consume max_commit_retries or
+# retry this itself (see
 # delta-rs kernel/transaction/conflict_checker.rs), because resolving it safely requires re-reading
 # the table and re-running the operation, which is exactly what its "must be rerun" error message
 # asks the caller to do.
@@ -37,12 +41,12 @@ def delta_merge_spill_kwargs() -> dict[str, int]:
 
 async def execute_with_conflict_retry(
     table: deltalake.DeltaTable,
-    operation_fn: Callable[[], dict],
+    operation_fn: Callable[[], T],
     operation_name: str,
     logger: FilteringBoundLogger,
-) -> dict:
-    """Run a Delta operation that commits (merge, optimize.compact, ...), refreshing the table
-    and re-running it on a commit conflict.
+) -> T:
+    """Run a Delta operation that commits (merge, optimize.compact, vacuum, ...), refreshing the
+    table and re-running it on a commit conflict.
 
     See DELTA_MERGE_CONFLICT_RETRIES for why this can't rely on delta-rs's own retry budget.
     """

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

@@ -171,6 +171,29 @@ async def test_retries_compact_on_commit_conflict_then_succeeds(self):
         mock_delta.update_incremental.assert_called_once()
 
 
+class TestVacuum:
+    @pytest.mark.asyncio
+    async def test_retries_on_commit_conflict_then_succeeds(self):
+        # vacuum() commits a REMOVE of tombstoned files — the same commit-conflict shape as
+        # optimize.compact() (see TestCompactTable.test_retries_compact_on_commit_conflict_then_succeeds).
+        # Regression coverage for a CommitFailedError propagating straight out of _vacuum on the first
+        # conflict instead of retrying with a refreshed table, like compact_table already does.
+        mock_delta = MagicMock()
+        mock_delta.vacuum = MagicMock(
+            side_effect=[
+                deltalake.exceptions.CommitFailedError(
+                    "Commit failed: a concurrent transaction deleted data this operation read."
+                ),
+                [],
+            ]
+        )
+
+        await _make_maintenance(mock_delta)._vacuum(mock_delta)
+
+        assert mock_delta.vacuum.call_count == 2
+        mock_delta.update_incremental.assert_called_once()
+
+
 class TestVacuumIfStale:
     @parameterized.expand(
         [