fix: Diff issues with STs that lead to always triggering config change (#1181)

### Description

In order to not always trigger on_config_change with streaming tables,
we need to track when config changes actually happen; further, as part
of debugging the failing tests, discovered that the Refresh schedule
would always trigger config change due to difference between None and
default of Utc that Databricks stores with.

### Checklist

- [x] I have run this code in development and it appears to resolve the
stated issue
- [x] This PR includes tests, or tests are not required/relevant for
this PR
- [x] I have updated the `CHANGELOG.md` and added information about my
change to the "dbt-databricks next" section.

Author: benc-db

CHANGELOG.md

@@ -1,5 +1,9 @@
 ## dbt-databricks 1.10.13 (TBD)
 
+### Fixes
+
+- Fix issue causing MV/STs to always trigger as having their config changed ([1181](http://github.com/databricks/dbt-databricks/pull/1181))
+
 ## dbt-databricks 1.10.12 (September 8, 2025)
 
 ### Under the hood

dbt/adapters/databricks/relation_configs/query.py

@@ -42,3 +42,11 @@ def from_relation_config(cls, relation_config: RelationConfig) -> QueryConfig:
             raise DbtRuntimeError(
                 f"Cannot compile model {relation_config.identifier} with no SQL query"
             )
+
+
+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")
+        return QueryConfig(query=SqlUtils.clean_sql(row[1]))

dbt/adapters/databricks/relation_configs/refresh.py

@@ -1,5 +1,5 @@
 import re
-from typing import ClassVar, Optional
+from typing import Any, ClassVar, Optional
 
 from dbt_common.exceptions import DbtRuntimeError
 
@@ -25,6 +25,18 @@ class RefreshConfig(DatabricksComponentConfig):
     # switching from manual refresh to scheduled or vice versa.
     is_altered: bool = False
 
+    def __eq__(self, other: Any) -> bool:
+        if not isinstance(other, RefreshConfig):
+            return False
+        return self.cron == other.cron and (
+            (
+                self.time_zone_value is None
+                and other.time_zone_value
+                and "utc" in other.time_zone_value.lower()
+            )
+            or (other.time_zone_value == other.time_zone_value)
+        )
+
     def get_diff(self, other: "RefreshConfig") -> Optional["RefreshConfig"]:
         if self != other:
             return RefreshConfig(

dbt/adapters/databricks/relation_configs/streaming_table.py

@@ -11,6 +11,7 @@
 from dbt.adapters.databricks.relation_configs.partitioning import (
     PartitionedByProcessor,
 )
+from dbt.adapters.databricks.relation_configs.query import DescribeQueryProcessor
 from dbt.adapters.databricks.relation_configs.refresh import RefreshConfig, RefreshProcessor
 from dbt.adapters.databricks.relation_configs.tblproperties import (
     TblPropertiesProcessor,
@@ -23,6 +24,7 @@ class StreamingTableConfig(DatabricksRelationConfigBase):
         CommentProcessor,
         TblPropertiesProcessor,
         RefreshProcessor,
+        DescribeQueryProcessor,
     ]
 
     def get_changeset(
@@ -33,16 +35,21 @@ def get_changeset(
         """
         changes: dict[str, DatabricksComponentConfig] = {}
         requires_refresh = False
+        requires_replace = False
 
         for component in self.config_components:
             key = component.name
             value = self.config[key]
             diff = value.get_diff(existing.config[key])
             if key == "partition_by" and diff is not None:
                 requires_refresh = True
+            if diff and diff != RefreshConfig():
+                requires_replace = True
             diff = diff or value
-
             if diff != RefreshConfig():
                 changes[key] = diff
-
-        return DatabricksRelationChangeSet(changes=changes, requires_full_refresh=requires_refresh)
+        if requires_replace:
+            return DatabricksRelationChangeSet(
+                changes=changes, requires_full_refresh=requires_refresh
+            )
+        return None

dbt/include/databricks/macros/materializations/streaming_table.sql

@@ -35,6 +35,7 @@
         {% set on_configuration_change = config.get('on_configuration_change') %}
         {% set configuration_changes = get_configuration_changes(existing_relation) %}
         {% if configuration_changes is none %}
+            {{ log("REFRESHING STREAMING TABLE: " ~ target_relation) }}
             {% set build_sql = refresh_streaming_table(target_relation, sql) %}
 
         {% elif on_configuration_change == 'apply' %}

dbt/include/databricks/macros/relations/streaming_table/refresh.sql

@@ -3,7 +3,5 @@
 {%- endmacro %}
 
 {% macro databricks__refresh_streaming_table(relation, sql) -%}
-  create or refresh streaming table {{ relation.render() }}
-  as
-    {{ sql }}
+  refresh streaming table {{ relation.render() }}
 {% endmacro %}

tests/functional/adapter/incremental/fixtures.py

@@ -564,6 +564,7 @@ def model(dbt, spark):
 models:
   - name: tags
     config:
+      use_notebook: true
       tags: ["python"]
       databricks_tags:
         a: b
@@ -587,6 +588,7 @@ def model(dbt, spark):
 models:
   - name: tblproperties
     config:
+      use_notebook: true
       tags: ["python"]
       tblproperties:
         a: b

tests/functional/adapter/streaming_tables/test_st_basic.py

@@ -149,8 +149,11 @@ def test_streaming_table_create(self, project, my_streaming_table):
     def test_streaming_table_create_idempotent(self, project, my_streaming_table):
         # setup creates it once; verify it's there and run once
         assert self.query_relation_type(project, my_streaming_table) == "streaming_table"
-        util.run_dbt(["run", "--models", my_streaming_table.identifier])
+        _, logs = util.run_dbt_and_capture(
+            ["--debug", "run", "--models", my_streaming_table.identifier]
+        )
         assert self.query_relation_type(project, my_streaming_table) == "streaming_table"
+        util.assert_message_in_logs("REFRESHING STREAMING TABLE", logs)
 
     def test_streaming_table_full_refresh(self, project, my_streaming_table):
         _, logs = util.run_dbt_and_capture(

tests/functional/adapter/streaming_tables/test_st_changes.py

@@ -243,3 +243,11 @@ def test_change_is_not_applied_via_replace(self, project, my_streaming_table):
         )
         util.assert_message_in_logs(f"Applying ALTER to: {my_streaming_table}", logs, False)
         util.assert_message_in_logs(f"Applying REPLACE to: {my_streaming_table}", logs, False)
+
+    def test_idempotent_run_does_not_fail(self, project, my_streaming_table):
+        assert self.query_relation_type(project, my_streaming_table) == "streaming_table"
+        _, log = util.run_dbt_and_capture(
+            ["--debug", "run", "--models", my_streaming_table.identifier]
+        )
+        assert self.query_relation_type(project, my_streaming_table) == "streaming_table"
+        util.assert_message_in_logs("REFRESHING STREAMING TABLE", log)

tests/unit/relation_configs/test_streaming_table_config.py

@@ -2,6 +2,7 @@
 
 from dbt.adapters.databricks.relation_configs.comment import CommentConfig
 from dbt.adapters.databricks.relation_configs.partitioning import PartitionedByConfig
+from dbt.adapters.databricks.relation_configs.query import QueryConfig
 from dbt.adapters.databricks.relation_configs.refresh import RefreshConfig
 from dbt.adapters.databricks.relation_configs.streaming_table import (
     StreamingTableConfig,
@@ -22,6 +23,7 @@ def test_from_results(self):
                     ["Catalog:", "default", None],
                     ["Comment", "This is the table comment", None],
                     ["Refresh Schedule", "MANUAL", None],
+                    ["View Text", "select * from foo", None],
                 ],
             ),
             "show_tblproperties": fixtures.gen_tblproperties([["prop", "1"], ["other", "other"]]),
@@ -35,6 +37,7 @@ def test_from_results(self):
                 "comment": CommentConfig(comment="This is the table comment"),
                 "tblproperties": TblPropertiesConfig(tblproperties={"prop": "1", "other": "other"}),
                 "refresh": RefreshConfig(),
+                "query": QueryConfig(query="select * from foo"),
             }
         )
 
@@ -59,6 +62,7 @@ def test_from_model_node(self):
                 "comment": CommentConfig(comment="This is the table comment", persist=False),
                 "tblproperties": TblPropertiesConfig(tblproperties={"prop": "1", "other": "other"}),
                 "refresh": RefreshConfig(),
+                "query": QueryConfig(query="select * from foo"),
             }
         )
 
@@ -69,6 +73,7 @@ def test_get_changeset__no_changes(self):
                 "comment": CommentConfig(comment="This is the table comment"),
                 "tblproperties": TblPropertiesConfig(tblproperties={"prop": "1", "other": "other"}),
                 "refresh": RefreshConfig(),
+                "query": QueryConfig(query="select * from foo"),
             }
         )
         new = StreamingTableConfig(
@@ -77,16 +82,13 @@ def test_get_changeset__no_changes(self):
                 "comment": CommentConfig(comment="This is the table comment"),
                 "tblproperties": TblPropertiesConfig(tblproperties={"prop": "1", "other": "other"}),
                 "refresh": RefreshConfig(),
+                "query": QueryConfig(query="select * from foo"),
             }
         )
 
         changeset = new.get_changeset(old)
-        assert not changeset.requires_full_refresh
-        assert changeset.changes == {
-            "tblproperties": TblPropertiesConfig(tblproperties={"prop": "1", "other": "other"}),
-            "comment": CommentConfig(comment="This is the table comment"),
-            "partition_by": PartitionedByConfig(partition_by=["col_a", "col_b"]),
-        }
+        # Based on the new logic, when there are no changes, get_changeset returns None
+        assert changeset is None
 
     def test_get_changeset__some_changes(self):
         old = StreamingTableConfig(
@@ -95,6 +97,7 @@ def test_get_changeset__some_changes(self):
                 "comment": CommentConfig(comment="This is the table comment"),
                 "tblproperties": TblPropertiesConfig(tblproperties={"prop": "1", "other": "other"}),
                 "refresh": RefreshConfig(),
+                "query": QueryConfig(query="select * from foo"),
             }
         )
         new = StreamingTableConfig(
@@ -103,15 +106,18 @@ def test_get_changeset__some_changes(self):
                 "comment": CommentConfig(comment="This is the table comment"),
                 "tblproperties": TblPropertiesConfig(tblproperties={"prop": "1", "other": "other"}),
                 "refresh": RefreshConfig(cron="*/5 * * * *"),
+                "query": QueryConfig(query="select * from foo"),
             }
         )
 
         changeset = new.get_changeset(old)
+        assert changeset is not None
         assert changeset.has_changes
         assert changeset.requires_full_refresh
         assert changeset.changes == {
             "partition_by": PartitionedByConfig(partition_by=["col_a"]),
             "comment": CommentConfig(comment="This is the table comment"),
             "tblproperties": TblPropertiesConfig(tblproperties={"prop": "1", "other": "other"}),
             "refresh": RefreshConfig(cron="*/5 * * * *"),
+            "query": QueryConfig(query="select * from foo"),
         }