fix: handle empty view definition in QueryProcessor (#1459)

CHANGELOG.md

@@ -1,3 +1,9 @@
+## dbt-databricks 1.12.3 (TBD)
+
+### Fixes
+
+- Handle missing or empty view-definition metadata when creating materialized views from streaming tables or newly-created materialized views (thanks @aarushisingh04!) ([#1462](https://github.com/databricks/dbt-databricks/pull/1462) resolves [#1459](https://github.com/databricks/dbt-databricks/issues/1459))
+
 ## dbt-databricks 1.12.2 (Jul 9, 2026)
 
 ### Features

dbt/adapters/databricks/relation_configs/query.py

@@ -27,7 +27,10 @@ class QueryProcessor(DatabricksComponentProcessor[QueryConfig]):
 
     @classmethod
     def from_relation_results(cls, result: RelationResults) -> QueryConfig:
-        view_definition = result["information_schema.views"]["view_definition"].strip()
+        view_definition_raw = result["information_schema.views"].get("view_definition") or ""
+        view_definition = view_definition_raw.strip()
+        if not view_definition:
+            return QueryConfig(query="")
         if view_definition.startswith("(") and view_definition.endswith(")"):
             view_definition = view_definition[1:-1]
         return QueryConfig(query=SqlUtils.clean_sql(view_definition))
@@ -48,7 +51,9 @@ class DescribeQueryProcessor(QueryProcessor):
     @classmethod
     def from_relation_results(cls, result: RelationResults) -> QueryConfig:
         table = result["describe_extended"]
-        row = next(x for x in table if x[0] == "View Text")
+        row = next((x for x in table if x[0] == "View Text"), None)
+        if row is None:
+            return QueryConfig(query="")
         if len(row) < 2:
             raise DbtRuntimeError(
                 "Unexpected result from DESCRIBE EXTENDED: missing View Text value"

tests/functional/adapter/materialized_view_tests/fixtures.py

@@ -221,6 +221,20 @@ def materialized_view_with_every(every_value: str) -> str:
 select * from {{ ref('mv_metadata_fetch_seed') }}
 """
 
+materialized_view_streaming_source_seed_csv = """id,value
+1,100
+""".lstrip()
+
+materialized_view_streaming_source_table_sql = """
+{{ config(materialized='streaming_table') }}
+select * from stream {{ ref('materialized_view_streaming_source_seed') }}
+"""
+
+materialized_view_streaming_source_sql = """
+{{ config(materialized='materialized_view') }}
+select * from {{ ref('materialized_view_streaming_source_table') }}
+"""
+
 
 mv_norebuild_seed_csv = """id,value
 1,100

tests/functional/adapter/materialized_view_tests/test_basic.py

@@ -6,6 +6,7 @@
 from dbt.tests.adapter.materialized_view import files
 from dbt.tests.adapter.materialized_view.basic import MaterializedViewBasic
 
+from tests.functional.adapter.fixtures import RerunSafeMixin
 from tests.functional.adapter.materialized_view_tests import fixtures
 
 
@@ -162,6 +163,50 @@ def created_time():
         assert created_time() == created_before
 
 
+@pytest.mark.dlt
+@pytest.mark.skip_profile("databricks_cluster", "databricks_uc_cluster")
+class TestMaterializedViewStreamingTableSource(RerunSafeMixin, TestMaterializedViewsMixin):
+    @pytest.fixture(scope="class")
+    def seeds(self):
+        return {
+            "materialized_view_streaming_source_seed.csv": (
+                fixtures.materialized_view_streaming_source_seed_csv
+            )
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "materialized_view_streaming_source_table.sql": (
+                fixtures.materialized_view_streaming_source_table_sql
+            ),
+            "materialized_view_streaming_source.sql": (
+                fixtures.materialized_view_streaming_source_sql
+            ),
+        }
+
+    @pytest.fixture(scope="class")
+    def relations_to_reset(self):
+        return (
+            "materialized_view_streaming_source_seed",
+            "materialized_view_streaming_source_table",
+            "materialized_view_streaming_source",
+        )
+
+    def test_create_materialized_view_from_streaming_table(self, project):
+        util.run_dbt(["seed"])
+        util.run_dbt(["run", "--select", "+materialized_view_streaming_source"])
+        util.run_dbt(["run", "--select", "+materialized_view_streaming_source"])
+
+        relation = project.adapter.Relation.create(
+            identifier="materialized_view_streaming_source",
+            schema=project.test_schema,
+            database=project.database,
+        )
+        assert self.query_relation_type(project, relation) == "materialized_view"
+        assert self.query_row_count(project, relation) == 1
+
+
 @pytest.mark.dlt
 @pytest.mark.skip_profile("databricks_cluster", "databricks_uc_cluster")
 class TestMaterializedViewLiquidClustering(TestMaterializedViewsMixin):

tests/unit/relation_configs/test_query.py

@@ -60,12 +60,49 @@ def test_get_diff__different_query(self):
         other = QueryProcessor.from_relation_results(results)
         assert model.get_diff(other) is model
 
+    def test_from_relation_results__none_view_definition(self):
+        results = {"information_schema.views": Row([None, ""], ["view_definition", "comment"])}
+        spec = QueryProcessor.from_relation_results(results)
+        assert spec == QueryConfig(query="")
+
+    def test_from_relation_results__missing_view_definition(self):
+        empty_row = Row(values=set())
+        results = {"information_schema.views": empty_row}
+        spec = QueryProcessor.from_relation_results(results)
+        assert spec == QueryConfig(query="")
+
+    def test_get_diff__empty_existing_query(self):
+        existing = QueryConfig(query="")
+        new = QueryConfig(query="select 1")
+        assert new.get_diff(existing) is new
+
 
 class TestDescribeQueryProcessor:
+    def test_from_relation_results__missing_view_text_row(self):
+        results = {
+            "describe_extended": [
+                ("col_name", "data_type", "comment"),
+                ("id", "int", ""),
+            ]
+        }
+        spec = DescribeQueryProcessor.from_relation_results(results)
+        assert spec == QueryConfig(query="")
+
     def test_from_relation_results__malformed_view_text_row(self):
         results = {"describe_extended": [("View Text",)]}
         with pytest.raises(
             DbtRuntimeError,
             match="Unexpected result from DESCRIBE EXTENDED: missing View Text value",
         ):
             DescribeQueryProcessor.from_relation_results(results)
+
+    def test_from_relation_results__valid_view_text_row(self):
+        sql = "select 1 as id"
+        results = {
+            "describe_extended": [
+                ("col_name", "data_type", "comment"),
+                ("View Text", sql, ""),
+            ]
+        }
+        spec = DescribeQueryProcessor.from_relation_results(results)
+        assert spec == QueryConfig(query=sql)