fix: recover materialized views with missing Delta targets (#311)

Allow an explicitly configured rebuild to recover a catalog-visible target whose Delta path is missing. Preserve fail-closed behavior for new targets, default policy, and unrelated errors.

Co-authored-by: Raki Rahman <mdrrakiburrahman@gmail.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: f3b6e4f1-5868-4fb6-93e6-51a3516ce9b6

src/dbt/adapters/fabricspark/impl.py

@@ -172,6 +172,31 @@ class FabricSparkAdapter(SQLAdapter):
     def execute_streaming_table(self, relation: Any, sql: str) -> Tuple[Any, "agate.Table"]:
         return experimental.execute_streaming_table(self, relation, sql)
 
+    @available
+    def get_materialized_view_query_hash(
+        self, relation: BaseRelation, allow_missing_delta: bool = False
+    ) -> Optional[str]:
+        """Read the target hash, allowing explicit rebuild of a missing Delta target."""
+        try:
+            _, result = self.execute(
+                f"SHOW TBLPROPERTIES {relation} ('dbt.fabricspark.materialized_view.query_hash')",
+                auto_begin=False,
+                fetch=True,
+            )
+        except DbtRuntimeError as exc:
+            missing_delta = re.match(
+                r"(?:(?:Runtime|Database) Error\s+)*\[DELTA_PATH_DOES_NOT_EXIST\]",
+                str(exc).lstrip(),
+            )
+            if not allow_missing_delta or missing_delta is None:
+                raise
+            logger.warning(
+                f"Materialized-view target {relation} has no committed Delta log; "
+                "proceeding with the explicitly requested on_query_change='rebuild'."
+            )
+            return None
+        return result.rows[0][1] if len(result.rows) == 1 else None
+
     @available
     def get_workspace_name_from_config(self, config) -> Optional[str]:
         """Resolve ``workspace_name`` from a model's ``config`` (top-level or

src/dbt/include/fabricspark/macros/materializations/materialized_view.sql

@@ -89,12 +89,10 @@
     {% endcall %}
   {% endif %}
 
-  {% call statement('materialized_view_query_hash', fetch_result=True) %}
-    SHOW TBLPROPERTIES {{ target_relation }} ('{{ query_hash_property }}')
-  {% endcall %}
-  {% set property_result = load_result('materialized_view_query_hash') %}
-  {% set property_rows = property_result['data'] if property_result is not none else [] %}
-  {% set existing_hash = property_rows[0][1] if property_rows | length == 1 else none %}
+  {% set existing_hash = adapter.get_materialized_view_query_hash(
+    target_relation,
+    allow_missing_delta=(existing_relation is not none and on_query_change == 'rebuild')
+  ) %}
   {% if existing_hash != query_hash %}
     {% if on_query_change == 'fail' %}
       {{ exceptions.raise_compiler_error(

tests/unit/test_materialized_view_recovery.py

@@ -0,0 +1,177 @@
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import agate
+import pytest
+from dbt_common.exceptions import CompilationError, DbtRuntimeError
+from jinja2 import Environment
+
+from dbt.adapters.fabricspark import FabricSparkAdapter
+
+QUERY_HASH = "a" * 32
+MACRO = (
+    Path(__file__).resolve().parents[2]
+    / "src/dbt/include/fabricspark/macros/materializations/materialized_view.sql"
+)
+MISSING_DELTA = "[DELTA_PATH_DOES_NOT_EXIST] /warehouse/mv_example is not a Delta table"
+
+
+def _materialize(*, error=None, stored_hash=QUERY_HASH, policy="rebuild", exists=True):
+    """Render the real materialization body with dbt side effects stubbed."""
+    relation = MagicMock()
+    relation.__str__.return_value = "`analytics`.`dbo`.`mv_example`"
+    relation.incorporate.return_value = relation
+    relation.schema = "dbo"
+    relation.database = "analytics"
+    relation.workspace = None
+    adapter = MagicMock(spec=FabricSparkAdapter)
+    rows = [] if stored_hash is None else [["query_hash", stored_hash]]
+    adapter.execute.return_value = (None, agate.Table(rows, ["key", "value"]))
+    adapter.execute.side_effect = error
+    adapter.get_materialized_view_query_hash = MagicMock(
+        side_effect=lambda target, **kwargs: FabricSparkAdapter.get_materialized_view_query_hash(
+            adapter, target, **kwargs
+        )
+    )
+    config = MagicMock()
+    config.get.side_effect = lambda key, default=None: {"on_query_change": policy}.get(
+        key, default
+    )
+    statements = []
+
+    def statement(name, fetch_result=False, caller=None):
+        sql = " ".join(caller().split())
+        if sql.startswith("SHOW TBLPROPERTIES"):
+            adapter.execute(sql, auto_begin=False, fetch=True)
+        else:
+            statements.append(sql)
+        return ""
+
+    def compiler_error(message):
+        raise CompilationError(message)
+
+    body = (
+        MACRO.read_text()
+        .split("{% materialization materialized_view, adapter='fabricspark' %}", 1)[1]
+        .split("{% endmaterialization %}", 1)[0]
+    )
+    context = {
+        "adapter": adapter,
+        "config": config,
+        "model": {"language": "sql"},
+        "flags": SimpleNamespace(FULL_REFRESH=False),
+        "should_full_refresh": lambda: False,
+        "fabricspark_materialized_view_columns": lambda *args: [],
+        "compiled_code": "select id from source_events",
+        "local_md5": lambda code: QUERY_HASH,
+        "this": relation,
+        "load_cached_relation": lambda target: relation if exists else None,
+        "ensure_database_exists": MagicMock(),
+        "run_hooks": lambda *args, **kwargs: "",
+        "pre_hooks": [],
+        "post_hooks": [],
+        "statement": statement,
+        "load_result": lambda name: {"data": rows},
+        "exceptions": SimpleNamespace(raise_compiler_error=compiler_error),
+        "return": lambda value: "",
+    }
+    return (
+        adapter,
+        relation,
+        statements,
+        lambda: Environment(extensions=["jinja2.ext.do"]).from_string(body).render(context),
+    )
+
+
+@pytest.mark.parametrize("nested", [False, True])
+def test_missing_delta_rebuilds_cached_target(nested):
+    error = DbtRuntimeError(MISSING_DELTA)
+    if nested:
+        error = DbtRuntimeError(str(error))
+    adapter, relation, statements, run = _materialize(error=error)
+    with patch("dbt.adapters.fabricspark.impl.logger") as logger:
+        run()
+    logger.warning.assert_called_once()
+    adapter.drop_relation.assert_called_once_with(relation)
+    assert len(statements) == 1
+    assert statements[0].startswith("CREATE MATERIALIZED VIEW ")
+    assert "IF NOT EXISTS" not in statements[0]
+    assert QUERY_HASH in statements[0]
+    adapter.execute.assert_called_once_with(
+        f"SHOW TBLPROPERTIES {relation} ('dbt.fabricspark.materialized_view.query_hash')",
+        auto_begin=False,
+        fetch=True,
+    )
+
+
+@pytest.mark.parametrize(("policy", "exists"), [("fail", True), ("rebuild", False)])
+def test_missing_delta_requires_existing_target_and_explicit_rebuild(policy, exists):
+    error = DbtRuntimeError(MISSING_DELTA)
+    adapter, _, statements, run = _materialize(error=error, policy=policy, exists=exists)
+    with pytest.raises(DbtRuntimeError) as raised:
+        run()
+    assert raised.value is error
+    adapter.drop_relation.assert_not_called()
+    assert len(statements) == (0 if exists else 1)
+
+
+@pytest.mark.parametrize(
+    "message",
+    [
+        "[PERMISSION_DENIED] access denied",
+        "[PERMISSION_DENIED] diagnostic mentions [DELTA_PATH_DOES_NOT_EXIST]",
+        "[SPARK_JOB_CANCELLED] job cancelled",
+        "[TABLE_OR_VIEW_NOT_FOUND] target missing",
+        "BlobStorageException: Status code 503 ServerBusy",
+        "UnknownHostException: Temporary failure in name resolution",
+    ],
+)
+def test_unrelated_errors_are_not_recovered(message):
+    error = DbtRuntimeError(message)
+    adapter, _, statements, run = _materialize(error=error)
+    with pytest.raises(DbtRuntimeError) as raised:
+        run()
+    assert raised.value is error
+    adapter.drop_relation.assert_not_called()
+    assert statements == []
+
+
+def test_matching_hash_refreshes_without_drop():
+    adapter, _, statements, run = _materialize()
+    run()
+    adapter.drop_relation.assert_not_called()
+    assert statements == ["REFRESH MATERIALIZED VIEW `analytics`.`dbo`.`mv_example`"]
+
+
+def test_changed_managed_hash_uses_materialized_view_drop():
+    adapter, _, statements, run = _materialize(stored_hash="b" * 32)
+    run()
+    adapter.drop_relation.assert_not_called()
+    assert statements[0] == "DROP MATERIALIZED VIEW `analytics`.`dbo`.`mv_example`"
+    assert statements[1].startswith("CREATE MATERIALIZED VIEW ")
+
+
+@pytest.mark.parametrize("stored_hash", [None, "missing property"])
+def test_unmanaged_target_rebuild_behavior_is_preserved(stored_hash):
+    adapter, relation, statements, run = _materialize(stored_hash=stored_hash)
+    run()
+    adapter.drop_relation.assert_called_once_with(relation)
+    assert len(statements) == 1
+    assert statements[0].startswith("CREATE MATERIALIZED VIEW ")
+
+
+def test_changed_hash_still_fails_without_rebuild():
+    adapter, _, statements, run = _materialize(stored_hash="b" * 32, policy="fail")
+    with pytest.raises(CompilationError, match="SQL changed"):
+        run()
+    adapter.drop_relation.assert_not_called()
+    assert statements == []
+
+
+def test_new_target_still_creates_then_refreshes():
+    adapter, _, statements, run = _materialize(exists=False)
+    run()
+    adapter.drop_relation.assert_not_called()
+    assert statements[0].startswith("CREATE MATERIALIZED VIEW IF NOT EXISTS ")
+    assert statements[1].startswith("REFRESH MATERIALIZED VIEW ")