diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta/maintenance.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta/maintenance.py index e79a14c4618c..278f5f55f571 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta/maintenance.py +++ b/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)) diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta/ops.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta/ops.py index 907eec5b2e67..7c11ee8f0424 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta/ops.py +++ b/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta/ops.py @@ -1,5 +1,6 @@ import asyncio from collections.abc import Callable +from typing import TypeVar from django.conf import settings @@ -7,6 +8,8 @@ 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. """ 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..28074347ff5e 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 @@ -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( [