Merge remote-tracking branch 'origin/1.12.latest' into fix/issue-1331-warehouse-only-parse-stub

Author: sd-db

CHANGELOG.md

@@ -6,6 +6,12 @@
 - Add support for row filters ([#1294](https://github.com/databricks/dbt-databricks/pull/1294))
 - 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))
+- Support `SCHEDULE EVERY` and `TRIGGER ON UPDATE` refresh modes for materialized views and streaming tables, with parser and diff coverage so relations whose actual refresh is not CRON no longer crash on subsequent runs ([#1293](https://github.com/databricks/dbt-databricks/issues/1293))
+
+### Fixes
+
+- Fix `RefreshConfig.__eq__` self/other typo where two configs with the same `cron` but different `time_zone_value` compared equal
+- Fix streaming-table DROP-SCHEDULE path that was silently filtered out of the changeset
 
 ### Fixes
 
@@ -18,6 +24,7 @@
 ### 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.
+- Skip `information_schema.tags` and `information_schema.column_tags` metadata fetches when table tags and column tags are not configured on a model. ([#1387](https://github.com/databricks/dbt-databricks/pull/1387))
 
 ## dbt-databricks 1.11.8 (TBD)
 

dbt/adapters/databricks/impl.py

@@ -81,6 +81,9 @@
     StreamingTableConfig,
 )
 from dbt.adapters.databricks.relation_configs.table_format import TableFormat
+from dbt.adapters.databricks.relation_configs.tags import (
+    TagsProcessor,
+)
 from dbt.adapters.databricks.relation_configs.tblproperties import TblPropertiesConfig
 from dbt.adapters.databricks.relation_configs.view import ViewConfig
 from dbt.adapters.databricks.utils import (
@@ -987,17 +990,21 @@ def parse_columns_and_constraints(
         return enriched_columns, parsed_constraints
 
     @available.parse(lambda *a, **k: {})
-    def get_relation_config(self, relation: DatabricksRelation) -> DatabricksRelationConfigBase:
+    def get_relation_config(
+        self,
+        relation: DatabricksRelation,
+        model_config: Optional[DatabricksRelationConfigBase] = None,
+    ) -> DatabricksRelationConfigBase:
         if relation.type == DatabricksRelationType.MaterializedView:
-            return MaterializedViewAPI.get_from_relation(self, relation)
+            return MaterializedViewAPI.get_from_relation(self, relation, model_config)
         elif relation.type == DatabricksRelationType.StreamingTable:
-            return StreamingTableAPI.get_from_relation(self, relation)
+            return StreamingTableAPI.get_from_relation(self, relation, model_config)
         elif relation.type == DatabricksRelationType.Table:
-            return IncrementalTableAPI.get_from_relation(self, relation)
+            return IncrementalTableAPI.get_from_relation(self, relation, model_config)
         elif relation.type == DatabricksRelationType.View:
-            return ViewAPI.get_from_relation(self, relation)
+            return ViewAPI.get_from_relation(self, relation, model_config)
         elif relation.type == DatabricksRelationType.MetricView:
-            return MetricViewAPI.get_from_relation(self, relation)
+            return MetricViewAPI.get_from_relation(self, relation, model_config)
         else:
             raise NotImplementedError(f"Relation type {relation.type} is not supported.")
 
@@ -1084,12 +1091,15 @@ def config_type(cls) -> type[DatabricksRelationConfig]:
 
     @classmethod
     def get_from_relation(
-        cls, adapter: DatabricksAdapter, relation: DatabricksRelation
+        cls,
+        adapter: DatabricksAdapter,
+        relation: DatabricksRelation,
+        model_config: Optional[DatabricksRelationConfigBase] = None,
     ) -> DatabricksRelationConfig:
         """Get the relation config from the relation."""
 
         assert relation.type == cls.relation_type
-        results = cls._describe_relation(adapter, relation)
+        results = cls._describe_relation(adapter, relation, model_config)
         return cls.config_type().from_results(results)
 
     @classmethod
@@ -1101,7 +1111,10 @@ def get_from_relation_config(cls, relation_config: RelationConfig) -> Databricks
     @classmethod
     @abstractmethod
     def _describe_relation(
-        cls, adapter: DatabricksAdapter, relation: DatabricksRelation
+        cls,
+        adapter: DatabricksAdapter,
+        relation: DatabricksRelation,
+        model_config: Optional[DatabricksRelationConfigBase] = None,
     ) -> RelationResults:
         """Describe the relation and return the results."""
 
@@ -1111,11 +1124,14 @@ def _describe_relation(
 class DeltaLiveTableAPIBase(RelationAPIBase[DatabricksRelationConfig]):
     @classmethod
     def get_from_relation(
-        cls, adapter: DatabricksAdapter, relation: DatabricksRelation
+        cls,
+        adapter: DatabricksAdapter,
+        relation: DatabricksRelation,
+        model_config: Optional[DatabricksRelationConfigBase] = None,
     ) -> DatabricksRelationConfig:
         """Get the relation config from the relation."""
 
-        relation_config = super().get_from_relation(adapter, relation)
+        relation_config = super().get_from_relation(adapter, relation, model_config)
 
         # Ensure any current refreshes are completed before returning the relation config
         tblproperties = cast(TblPropertiesConfig, relation_config.config["tblproperties"])
@@ -1135,7 +1151,10 @@ def config_type(cls) -> type[MaterializedViewConfig]:
 
     @classmethod
     def _describe_relation(
-        cls, adapter: DatabricksAdapter, relation: DatabricksRelation
+        cls,
+        adapter: DatabricksAdapter,
+        relation: DatabricksRelation,
+        model_config: Optional[DatabricksRelationConfigBase] = None,
     ) -> RelationResults:
         kwargs = {"table_name": relation}
         results: RelationResults = dict()
@@ -1147,9 +1166,18 @@ def _describe_relation(
         results["information_schema.views"] = get_first_row(
             adapter.execute_macro("get_view_description", kwargs=kwargs)
         )
-        results["information_schema.tags"] = adapter.execute_macro("fetch_tags", kwargs=kwargs)
         results["show_tblproperties"] = adapter.execute_macro("fetch_tbl_properties", kwargs=kwargs)
+
+        # To be backward compatible model_config can be None. In that case, tags should be fetched
+        # to maintain backward compatibility.
+        table_tag_config = model_config.config.get(TagsProcessor.name) if model_config else None
+        if table_tag_config is None or table_tag_config.requires_server_metadata_for_diff():
+            results["information_schema.tags"] = adapter.execute_macro("fetch_tags", kwargs=kwargs)
+        else:
+            results["information_schema.tags"] = None
+
         results["row_filters"] = adapter.execute_macro("fetch_row_filters", kwargs=kwargs)
+
         return results
 
 
@@ -1162,7 +1190,10 @@ def config_type(cls) -> type[StreamingTableConfig]:
 
     @classmethod
     def _describe_relation(
-        cls, adapter: DatabricksAdapter, relation: DatabricksRelation
+        cls,
+        adapter: DatabricksAdapter,
+        relation: DatabricksRelation,
+        model_config: Optional[DatabricksRelationConfigBase] = None,
     ) -> RelationResults:
         kwargs = {"table_name": relation}
         results: RelationResults = dict()
@@ -1186,16 +1217,37 @@ def config_type(cls) -> type[IncrementalTableConfig]:
 
     @classmethod
     def _describe_relation(
-        cls, adapter: DatabricksAdapter, relation: DatabricksRelation
+        cls,
+        adapter: DatabricksAdapter,
+        relation: DatabricksRelation,
+        model_config: Optional[DatabricksRelationConfigBase] = None,
     ) -> RelationResults:
         results = {}
         kwargs = {"relation": relation}
 
         if not relation.is_hive_metastore():
-            results["information_schema.tags"] = adapter.execute_macro("fetch_tags", kwargs=kwargs)
-            results["information_schema.column_tags"] = adapter.execute_macro(
-                "fetch_column_tags", kwargs=kwargs
+            # To be backward compatible model_config can be None. In that case, tags should be
+            # fetched to maintain backward compatibility.
+            table_tag_config = model_config.config.get(TagsProcessor.name) if model_config else None
+            if table_tag_config is None or table_tag_config.requires_server_metadata_for_diff():
+                results["information_schema.tags"] = adapter.execute_macro(
+                    "fetch_tags", kwargs=kwargs
+                )
+            else:
+                results["information_schema.tags"] = None
+
+            # To be backward compatible model_config can be None. In that case, tags should be
+            # fetched to maintain backward compatibility.
+            column_tag_config = (
+                model_config.config.get(ColumnTagsProcessor.name) if model_config else None
             )
+            if column_tag_config is None or column_tag_config.requires_server_metadata_for_diff():
+                results["information_schema.column_tags"] = adapter.execute_macro(
+                    "fetch_column_tags", kwargs=kwargs
+                )
+            else:
+                results["information_schema.column_tags"] = None
+
             results["non_null_constraint_columns"] = adapter.execute_macro(
                 "fetch_non_null_constraint_columns", kwargs=kwargs
             )
@@ -1225,15 +1277,26 @@ def config_type(cls) -> type[ViewConfig]:
 
     @classmethod
     def _describe_relation(
-        cls, adapter: DatabricksAdapter, relation: DatabricksRelation
+        cls,
+        adapter: DatabricksAdapter,
+        relation: DatabricksRelation,
+        model_config: Optional[DatabricksRelationConfigBase] = None,
     ) -> RelationResults:
         results = {}
         kwargs = {"relation": relation}
 
         results["information_schema.views"] = get_first_row(
             adapter.execute_macro("get_view_description", kwargs=kwargs)
         )
-        results["information_schema.tags"] = adapter.execute_macro("fetch_tags", kwargs=kwargs)
+
+        # To be backward compatible model_config can be None. In that case, tags should be fetched
+        # to maintain backward compatibility.
+        table_tag_config = model_config.config.get(TagsProcessor.name) if model_config else None
+        if table_tag_config is None or table_tag_config.requires_server_metadata_for_diff():
+            results["information_schema.tags"] = adapter.execute_macro("fetch_tags", kwargs=kwargs)
+        else:
+            results["information_schema.tags"] = None
+
         results["show_tblproperties"] = adapter.execute_macro("fetch_tbl_properties", kwargs=kwargs)
 
         kwargs = {"table_name": relation}
@@ -1252,14 +1315,25 @@ def config_type(cls) -> type[MetricViewConfig]:
 
     @classmethod
     def _describe_relation(
-        cls, adapter: DatabricksAdapter, relation: DatabricksRelation
+        cls,
+        adapter: DatabricksAdapter,
+        relation: DatabricksRelation,
+        model_config: Optional[DatabricksRelationConfigBase] = None,
     ) -> RelationResults:
         results = {}
         kwargs = {"relation": relation}
-        results["information_schema.tags"] = adapter.execute_macro("fetch_tags", kwargs=kwargs)
         results["show_tblproperties"] = adapter.execute_macro("fetch_tbl_properties", kwargs=kwargs)
+
+        kwargs = {"relation": relation}
+        table_tag_config = model_config.config.get(TagsProcessor.name) if model_config else None
+        if table_tag_config is None or table_tag_config.requires_server_metadata_for_diff():
+            results["information_schema.tags"] = adapter.execute_macro("fetch_tags", kwargs=kwargs)
+        else:
+            results["information_schema.tags"] = None
+
         kwargs = {"table_name": relation}
         results["describe_extended"] = adapter.execute_macro(
             DESCRIBE_TABLE_EXTENDED_MACRO_NAME, kwargs=kwargs
         )
+
         return results

dbt/adapters/databricks/relation_configs/base.py

@@ -34,6 +34,12 @@ def get_diff(self, other: Self) -> Optional[Self]:
             return self
         return None
 
+    def requires_server_metadata_for_diff(self) -> bool:
+        """
+        Indicates whether server metadata is required to compute the diff for this component.
+        """
+        return True
+
 
 class DatabricksRelationChangeSet(BaseModel):
     """Class for encapsulating the changes that need to be applied to a Databricks relation."""

dbt/adapters/databricks/relation_configs/column_tags.py

@@ -41,6 +41,12 @@ def get_diff(self, other: "ColumnTagsConfig") -> Optional["ColumnTagsConfig"]:
             return ColumnTagsConfig(set_column_tags=set_column_tags)
         return None
 
+    def requires_server_metadata_for_diff(self) -> bool:
+        """
+        Indicates whether server metadata is required to compute the diff for this component.
+        """
+        return self.set_column_tags is not None and len(self.set_column_tags) > 0
+
 
 class ColumnTagsProcessor(DatabricksComponentProcessor[ColumnTagsConfig]):
     name: ClassVar[str] = "column_tags"