diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/load.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/load.py index 70ee1b491ebb..8a4b705361a8 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/load.py +++ b/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) diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/test/test_load.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/test/test_load.py index 9c0ccc5f6c12..f456f98cc380 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/common/test/test_load.py +++ b/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 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 5054c0910b45..97fc26b02d2a 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 @@ -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 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 1104500088a4..6d48d7b7c42c 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 @@ -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):