From 383de709485dd2ec045d7fb0772b85c86e0e98df Mon Sep 17 00:00:00 2001 From: Tom Owers Date: Wed, 29 Jul 2026 15:40:05 +0100 Subject: [PATCH] fix(data-imports): treat a racy optimize/vacuum scan failure as transient MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pre-write delta maintenance (`run_pre_write_defensive_compact`) can hit a `DeltaError` when `optimize.compact`'s scan reads a file a concurrent maintenance pass has already vacuumed away. The scan fails before anything commits, so the table is left exactly as it was — just still fragmented — making this safe to skip and retry on the next pass, the same way an existing transient object-store error is already handled. Classify this specific delta-rs error signature the same way and log a warning instead of capturing it to error tracking. Generated-By: PostHog Code Task-Id: ee9ef592-a2cd-471f-a4a4-1520a5e26b1b --- .../data_imports/pipelines/common/extract.py | 12 ++++++---- .../pipelines/common/test/test_extract.py | 22 ++++++++++++++++++ .../pipelines/core/delta_table_helper.py | 15 ++++++++++++ .../core/test/test_delta_table_helper.py | 23 +++++++++++++++++++ 4 files changed, 67 insertions(+), 5 deletions(-) diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/extract.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/extract.py index 71903ab38c67..9ef2d758212e 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/extract.py +++ b/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/extract.py @@ -21,6 +21,7 @@ from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.cdp_producer import CDPProducer from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.delta_table_helper import ( DeltaTableHelper, + is_transient_delta_maintenance_error, is_transient_object_store_error, ) from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.person_property_row_sink import ( @@ -704,9 +705,10 @@ async def run_pre_write_defensive_compact( the CDC post-load path in `common/load.py` writes the same watermark, and both merge via `update_sync_type_config_keys` under a row lock. Wrapped in try/except so a maintenance failure never blocks the actual sync; the original error path is unaffected. - A transient object-store error (see `is_transient_object_store_error`) is logged at - warning instead of captured — the next sync's maintenance pass retries it from scratch, - so it isn't a bug in this function, just a temporary blip talking to our own S3 bucket. + A transient object-store error (see `is_transient_object_store_error`) or a racy + concurrent-maintenance DeltaError (see `is_transient_delta_maintenance_error`) is + logged at warning instead of captured — the next sync's maintenance pass retries it + from scratch, so neither is a bug in this function. Used by both `PipelineNonDLT.run` (v2) and `PipelineV3.run` to keep the behaviour identical across pipelines without each having to know how to look up `partition_count` @@ -730,8 +732,8 @@ async def run_pre_write_defensive_compact( schema.id, schema.team_id, updates={"last_vacuum_version": new_version} ) except Exception as e: - if is_transient_object_store_error(e): - await logger.awarning(f"Pre-write maintenance skipped: transient object-store error: {e}") + if is_transient_object_store_error(e) or is_transient_delta_maintenance_error(e): + await logger.awarning(f"Pre-write maintenance skipped: transient error: {e}") return capture_exception(e) await logger.aexception(f"Pre-write maintenance failed: {e}", exc_info=e) diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/test/test_extract.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/test/test_extract.py index 6f21faa3d827..b769cd8ee556 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/test/test_extract.py +++ b/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/test/test_extract.py @@ -5,6 +5,7 @@ from posthog.test.base import BaseTest from unittest.mock import AsyncMock, MagicMock, patch +import deltalake from asgiref.sync import async_to_sync from parameterized import parameterized @@ -199,6 +200,27 @@ async def test_logs_transient_object_store_error_without_capturing(self, _name: logger.awarning.assert_awaited_once() logger.aexception.assert_not_awaited() + @pytest.mark.asyncio + async def test_logs_transient_delta_maintenance_race_without_capturing(self): + # Regression: a concurrent optimize/vacuum pass on the same table (e.g. a zombie Temporal + # attempt racing its own retry) can have `optimize.compact` scan a file the other attempt + # already vacuumed away. Nothing gets committed when the scan fails, so the table isn't + # corrupted — this must be treated the same as the object-store blips above, not captured. + error = deltalake.exceptions.DeltaError( + "Failed to parse parquet: Optimize selected-file scan failed while scanning data: " + "Object at location .../part-0.parquet not found: 404 Not Found" + ) + helper = MagicMock(run_maintenance=AsyncMock(side_effect=error)) + logger = MagicMock(aexception=AsyncMock(), awarning=AsyncMock()) + + schema = MagicMock(partition_count=5, sync_type_config={}) + with patch(f"{_EXTRACT_MODULE}.capture_exception") as mock_capture: + await run_pre_write_defensive_compact(helper, schema, MagicMock(partition_count=None), logger) + + mock_capture.assert_not_called() + logger.awarning.assert_awaited_once() + logger.aexception.assert_not_awaited() + class TestReportHeartbeatTimeoutRecording(BaseTest): def _schema(self) -> ExternalDataSchema: diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta_table_helper.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta_table_helper.py index 4176f587bcc9..7ee920e61f2b 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta_table_helper.py +++ b/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta_table_helper.py @@ -78,6 +78,21 @@ def is_transient_object_store_error(error: BaseException) -> bool: # exactly what its "must be rerun" error message asks the caller to do. DELTA_MERGE_CONFLICT_RETRIES = 3 +# `optimize.compact` plans its rewrite against the file list at the start of its scan, then reads +# those files. A concurrent maintenance pass on the same table (e.g. a Temporal activity attempt that +# heartbeat-timed-out but keeps running as a zombie — see this package's README on the equivalent +# unfenced race for repartition) can vacuum one of those files out from under the scan before it gets +# 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",) + + +def is_transient_delta_maintenance_error(error: BaseException) -> bool: + return isinstance(error, deltalake.exceptions.DeltaError) and any( + needle in str(error) for needle in TRANSIENT_DELTA_MAINTENANCE_ERRORS + ) + def _delta_merge_spill_kwargs() -> dict[str, int]: """delta-rs `merge` kwargs that let DataFusion spill to disk instead of OOMing on large merges. diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/test/test_delta_table_helper.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/test/test_delta_table_helper.py index ed155fae90d2..6a450e35a42e 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/test/test_delta_table_helper.py +++ b/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/test/test_delta_table_helper.py @@ -25,6 +25,7 @@ _delta_merge_spill_kwargs, _first_per_pk_table, _realign_decimal_buffers, + is_transient_delta_maintenance_error, ) @@ -1151,3 +1152,25 @@ async def test_is_table_corrupted(self, _name: str, is_delta: bool, open_exc: Ex result = await helper.is_table_corrupted() assert result is expected + + +class TestIsTransientDeltaMaintenanceError: + @parameterized.expand( + [ + # A concurrent optimize/vacuum losing the race on a file scan: safe to skip and retry. + ( + "optimize_scan_file_not_found", + deltalake.exceptions.DeltaError( + "Failed to parse parquet: Optimize selected-file scan failed while scanning data: " + "Object at location .../part-0.parquet not found: 404 Not Found" + ), + True, + ), + # 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): + assert is_transient_delta_maintenance_error(error) is expected