fix(data-imports): classify a concurrent reset losing a delta log commit file as transient

products/warehouse_sources/backend/temporal/data_imports/pipelines/common/load.py

@@ -274,6 +274,7 @@ async def _run_delta_maintenance(
     logger: FilteringBoundLogger,
 ) -> None:
     from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.delta_table_helper import (  # noqa: PLC0415 — keeps the heavy deltalake dep off this module's top-level import path
+        is_transient_delta_maintenance_error,
         is_transient_object_store_error,
     )
 
@@ -314,6 +315,8 @@ async def _run_delta_maintenance(
                 # A rate-limited or connectivity blip talking to our own S3 bucket isn't a bug - the
                 # next tick's maintenance pass retries the same idempotent cleanup.
                 logger.warning(f"Delta maintenance skipped: transient object-store error: {e}")
+            elif is_transient_delta_maintenance_error(e):
+                logger.warning(f"Delta maintenance skipped: transient maintenance error: {e}")
             else:
                 capture_exception(e)
                 logger.exception(f"Delta maintenance failed: {e}", exc_info=e)
@@ -325,6 +328,8 @@ async def _run_delta_maintenance(
         except Exception as e:
             if is_transient_object_store_error(e):
                 logger.warning(f"Compaction skipped: transient object-store error: {e}")
+            elif is_transient_delta_maintenance_error(e):
+                logger.warning(f"Compaction skipped: transient maintenance error: {e}")
             else:
                 capture_exception(e)
                 logger.exception(f"Compaction failed: {e}", exc_info=e)

products/warehouse_sources/backend/temporal/data_imports/pipelines/common/test/test_load.py

@@ -4,6 +4,7 @@
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pyarrow as pa
+import deltalake.exceptions
 from parameterized import parameterized
 
 from products.warehouse_sources.backend.models.external_data_schema import ExternalDataSchema
@@ -197,6 +198,15 @@ async def test_prepares_s3_files_with_post_maintenance_file_list(self, _name: st
             # tick's maintenance retries the same idempotent cleanup) and must not be promoted
             # into a fresh error-tracking issue — the regression this guards.
             ("transient_s3_slowdown", OSError("Generic S3 error: Please reduce your request rate."), False),
+            # Same non-fatal handling for a concurrent-maintenance DeltaError (e.g. a full_refresh
+            # reset losing a `_delta_log` commit file out from under this same compact/vacuum call).
+            (
+                "transient_delta_maintenance_race",
+                deltalake.exceptions.DeltaError(
+                    "Generic error: Kernel error: File not found: table/_delta_log/00000000000000000001.json"
+                ),
+                False,
+            ),
         ]
     )
     @pytest.mark.asyncio
@@ -217,6 +227,13 @@ async def test_maintenance_failure_handling(self, _name: str, error: Exception,
         [
             ("genuine_bug", RuntimeError("compaction blew up"), True),
             ("transient_s3_slowdown", OSError("Generic S3 error: Please reduce your request rate."), False),
+            (
+                "transient_delta_maintenance_race",
+                deltalake.exceptions.DeltaError(
+                    "Generic error: Kernel error: File not found: table/_delta_log/00000000000000000001.json"
+                ),
+                False,
+            ),
         ]
     )
     @pytest.mark.asyncio

products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta_table_helper.py

@@ -108,9 +108,20 @@ def is_transient_object_store_error(error: BaseException) -> bool:
 
 
 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
-    )
+    if not isinstance(error, deltalake.exceptions.DeltaError):
+        return False
+
+    text = str(error)
+    if any(needle in text for needle in TRANSIENT_DELTA_MAINTENANCE_ERRORS):
+        return True
+
+    # A zombie's compact/vacuum can also lose a commit file from under it: `reset_table` (full_refresh)
+    # purges the whole table prefix, including `_delta_log`, out from under a still-running maintenance
+    # pass that opened the table before the purge. Neither vacuum nor optimize.compact ever delete a
+    # `_delta_log/*.json` commit file themselves, so a missing one here means something else raced the
+    # read, not a corrupt table — that's why this checks for the log directory specifically rather than
+    # matching "File not found" alone, which a genuinely missing/corrupt table can also raise.
+    return "File not found" in text and "_delta_log/" in text
 
 
 # _purge_s3_prefix is idempotent (every step is existence-gated), so retrying it whole after a brief

products/warehouse_sources/backend/temporal/data_imports/pipelines/core/test/test_delta_table_helper.py

@@ -1265,6 +1265,19 @@ class TestIsTransientDeltaMaintenanceError:
             ("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),
+            # A concurrent `reset_table` purging the whole prefix (e.g. a full_refresh sync) out from
+            # under a still-running maintenance pass: safe to skip and retry.
+            (
+                "missing_delta_log_commit_file",
+                deltalake.exceptions.DeltaError(
+                    "Generic error: Kernel error: File not found: "
+                    "dlt/team_1_source_2/table/_delta_log/00000000000000000001.json"
+                ),
+                True,
+            ),
+            # "File not found" alone, without the log directory, must not match - a missing data file
+            # for some other reason is a real failure to capture, not this specific log-commit race.
+            ("file_not_found_outside_delta_log", deltalake.exceptions.DeltaError("File not found: some/file"), False),
         ]
     )
     def test_matches_only_the_racy_optimize_scan_signature(self, _name: str, error: Exception, expected: bool):