fix(data-imports): treat a racy optimize/vacuum scan failure as transient

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)

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:

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.

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