fix: auto-quote backtick-rendered refs in metric_view YAML body (#1361)

CHANGELOG.md

@@ -7,6 +7,10 @@
 - Add support for Python UDFs ([#1336](https://github.com/databricks/dbt-databricks/pull/1336))
 - Add support for key-only `databricks_tags` for table and column tagging. This can now be configured by setting tag values to empty strings `""` or `None`. ([#1339](https://github.com/databricks/dbt-databricks/pull/1339))
 
+### Fixes
+
+- Fix `metric_view` failing with `METRIC_VIEW_INVALID_VIEW_DEFINITION` when models use bare `{{ ref(...) }}` for the `source:` field ([#1361](https://github.com/databricks/dbt-databricks/issues/1361))
+
 ### Under the Hood
 
 - **BREAKING:** `databricks_tags` defined at different hierarchy levels (e.g. project-level and model-level) now merge additively instead of the child config completely replacing the parent.

dbt/adapters/databricks/handle.py

@@ -267,6 +267,10 @@ class SqlUtils:
     """
 
     DBR_VERSION_REGEX = re.compile(r"([1-9][0-9]*)\.(x|0|[1-9][0-9]*)")
+    YAML_BACKTICK_VALUE_REGEX = re.compile(
+        r"^(?P<indent>[ \t]*)(?P<key>\w+:[ \t]+)(?P<value>`[^`\n]+`(?:\.`[^`\n]+`)*)[ \t]*$",
+        re.MULTILINE,
+    )
     user_agent = f"dbt-databricks/{__version__}"
 
     @staticmethod
@@ -295,6 +299,13 @@ def clean_sql(sql: str) -> str:
             cleaned = cleaned[:-1]
         return cleaned
 
+    @staticmethod
+    def yaml_quote_backtick_values(yaml_body: str) -> str:
+        """Wrap backtick-rendered SQL identifiers in YAML double quotes (see #1361)."""
+        if "`" not in yaml_body:
+            return yaml_body
+        return SqlUtils.YAML_BACKTICK_VALUE_REGEX.sub(r'\g<indent>\g<key>"\g<value>"', yaml_body)
+
     @staticmethod
     def prepare_connection_arguments(
         creds: DatabricksCredentials,

dbt/adapters/databricks/impl.py

@@ -973,6 +973,10 @@ def is_cluster(self) -> bool:
     def clean_sql(self, sql: str) -> str:
         return SqlUtils.clean_sql(sql)
 
+    @available.parse(lambda *a, **k: "")
+    def yaml_quote_backtick_values(self, yaml_body: str) -> str:
+        return SqlUtils.yaml_quote_backtick_values(yaml_body)
+
     @available
     def build_catalog_relation(self, model: RelationConfig) -> Optional[CatalogRelation]:
         """

dbt/include/databricks/macros/relations/metric_view/alter.sql

@@ -33,8 +33,9 @@
 {%- endmacro %}
 
 {% macro databricks__get_alter_metric_view_as_sql(relation, yaml_content) %}
+{%- set yaml_body = adapter.yaml_quote_backtick_values(yaml_content) -%}
 alter view {{ relation.render() }} as $$
-{{ yaml_content }}
+{{ yaml_body }}
 $$
 {% endmacro %}
 

dbt/include/databricks/macros/relations/metric_view/create.sql

@@ -3,10 +3,11 @@
 {%- endmacro %}
 
 {% macro databricks__get_create_metric_view_as_sql(relation, sql) %}
+{%- set yaml_body = adapter.yaml_quote_backtick_values(sql) -%}
 create or replace view {{ relation.render() }}
 with metrics
 language yaml
 as $$
-{{ sql }}
+{{ yaml_body }}
 $$
 {% endmacro %}

tests/functional/adapter/metric_views/fixtures.py

@@ -41,6 +41,21 @@
     expr: sum(revenue)
 """
 
+metric_view_bare_ref = """
+{{ config(materialized='metric_view') }}
+
+version: 0.1
+source: {{ ref('source_orders') }}
+dimensions:
+  - name: order_date
+    expr: order_date
+  - name: status
+    expr: status
+measures:
+  - name: total_orders
+    expr: count(1)
+"""
+
 metric_view_with_config = """
 {{
   config(

tests/functional/adapter/metric_views/test_metric_view_materialization.py

@@ -3,6 +3,7 @@
 
 from tests.functional.adapter.metric_views.fixtures import (
     basic_metric_view,
+    metric_view_bare_ref,
     metric_view_with_config,
     metric_view_with_filter,
     source_table,
@@ -211,3 +212,31 @@ def test_metric_view_with_tags(self, project):
         status_data = {row[0]: row[1] for row in query_result}
         assert status_data["completed"] == 2
         assert status_data["pending"] == 1
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestMetricViewBareRef:
+    """Regression for #1361: bare {{ ref(...) }} in metric_view body."""
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "source_orders.sql": source_table,
+            "bare_ref_metrics.sql": metric_view_bare_ref,
+        }
+
+    def test_bare_ref_metric_view_creates_and_queries(self, project):
+        results = run_dbt(["run"])
+        assert len(results) == 2
+        assert all(r.status == "success" for r in results), (
+            f"Expected all models to succeed, got: "
+            f"{[(r.node.name, r.status, r.message) for r in results]}"
+        )
+
+        # Run-success alone doesn't catch a non-functional view.
+        metric_view_name = f"{project.database}.{project.test_schema}.bare_ref_metrics"
+        query_result = project.run_sql(
+            f"SELECT MEASURE(total_orders) FROM {metric_view_name}",
+            fetch="all",
+        )
+        assert query_result and query_result[0][0] == 3

tests/unit/macros/base.py

@@ -43,6 +43,7 @@ def default_context(self) -> dict:
         # Create a mock adapter with a working quote method
         mock_adapter = Mock()
         mock_adapter.quote = lambda identifier: f"`{identifier}`"
+        mock_adapter.yaml_quote_backtick_values = lambda yaml_body: yaml_body
 
         # Create a mock for api.Relation.create that returns a relation with proper render method
         def mock_relation_create(database=None, schema=None, identifier=None, type=None):

tests/unit/test_handle.py

@@ -23,6 +23,52 @@ def test_translate_bindings(self, bindings, expected):
     def test_clean_sql(self, sql, expected):
         assert SqlUtils.clean_sql(sql) == expected
 
+    @pytest.mark.parametrize(
+        "yaml_body, expected",
+        [
+            pytest.param(
+                "version: 0.1\nsource: `db`.`schema`.`name`\n",
+                'version: 0.1\nsource: "`db`.`schema`.`name`"\n',
+                id="three_part_bare",
+            ),
+            pytest.param("source: `name`\n", 'source: "`name`"\n', id="one_part_bare"),
+            pytest.param(
+                "source: `schema`.`name`\n",
+                'source: "`schema`.`name`"\n',
+                id="two_part_bare",
+            ),
+            pytest.param(
+                'source: "`db`.`schema`.`name`"\n',
+                'source: "`db`.`schema`.`name`"\n',
+                id="already_quoted_idempotent",
+            ),
+            pytest.param(
+                "source: db.schema.name\n",
+                "source: db.schema.name\n",
+                id="no_backticks",
+            ),
+            pytest.param(
+                "expr: count(`some col`)\n",
+                "expr: count(`some col`)\n",
+                id="inline_backtick_in_expr",
+            ),
+            pytest.param(
+                "  source: `db`.`s`.`n`\n",
+                '  source: "`db`.`s`.`n`"\n',
+                id="indented_mapping",
+            ),
+            pytest.param(
+                "version: 0.1\nsource: `db`.`s`.`a`\nfilter: `db`.`s`.`b`\n",
+                'version: 0.1\nsource: "`db`.`s`.`a`"\nfilter: "`db`.`s`.`b`"\n',
+                id="multiple_lines",
+            ),
+            pytest.param("", "", id="empty"),
+            pytest.param("   \n", "   \n", id="whitespace_only"),
+        ],
+    )
+    def test_yaml_quote_backtick_values(self, yaml_body, expected):
+        assert SqlUtils.yaml_quote_backtick_values(yaml_body) == expected
+
     @pytest.mark.parametrize("result, expected", [("14.x", (14, sys.maxsize)), ("12.1", (12, 1))])
     def test_extract_dbr_version(self, result, expected):
         assert SqlUtils.extract_dbr_version(result) == expected