Merge branch 'main' into 1.10.latest

Author: benc-db

.github/CODEOWNERS

@@ -2,4 +2,4 @@
 # the repo. Unless a later match takes precedence, these
 # users will be requested for review when someone opens a
 # pull request.
-*       @andrefurlan-db @susodapop @benc-db @rcypher-databricks
+*       @benc-db @ericj-db

CHANGELOG.md

@@ -1,3 +1,18 @@
+## dbt-databricks 1.10.3 (TBD)
+
+## dbt-databricks 1.10.2 (May 21, 2025)
+
+### Features
+
+- Support constraint updates on incremental runs (with Materialization V2) ([1013](https://github.com/databricks/dbt-databricks/pull/1013))
+- Add catalog integration support - set table formats, file formats, and locations in `catalogs.yml` ([1012](https://github.com/databricks/dbt-databricks/pull/1012))
+
+### Fixes
+
+- Fix bug with multiple not_null constraints defined on the model level ([1008](https://github.com/databricks/dbt-databricks/pull/1008))
+- Fix bug with temp tables not being dropped after python model is materialized ([1010](https://github.com/databricks/dbt-databricks/issues/1010))
+- Fix bug with alter view dispatch where dbt could not find the approprate macro ([1029](https://github.com/databricks/dbt-databricks/pull/1029))
+
 ## dbt-databricks 1.10.1 (Apr 29, 2025)
 
 ### Fixes

dbt/adapters/databricks/__version__.py

@@ -1 +1 @@
-version = "1.10.1"
+version = "1.10.2"

dbt/adapters/databricks/api_client.py

@@ -354,7 +354,7 @@ def _get_exception(self, response: Response) -> None:
         result_state = state.get("result_state")
         life_cycle_state = state["life_cycle_state"]
 
-        if result_state is not None and result_state != "SUCCESS":
+        if result_state == "CANCELED":
             raise DbtRuntimeError(f"Python model run ended in result_state {result_state}")
 
         if life_cycle_state != "TERMINATED":

dbt/adapters/databricks/catalogs/__init__.py

@@ -0,0 +1,9 @@
+from dbt.adapters.databricks.catalogs._hive_metastore import HiveMetastoreCatalogIntegration
+from dbt.adapters.databricks.catalogs._relation import DatabricksCatalogRelation
+from dbt.adapters.databricks.catalogs._unity import UnityCatalogIntegration
+
+__all__ = [
+    "DatabricksCatalogRelation",
+    "HiveMetastoreCatalogIntegration",
+    "UnityCatalogIntegration",
+]

dbt/adapters/databricks/catalogs/_hive_metastore.py

@@ -0,0 +1,42 @@
+from typing import Optional
+
+from dbt.adapters.catalogs import CatalogIntegration, CatalogIntegrationConfig
+from dbt.adapters.contracts.relation import RelationConfig
+from dbt.adapters.databricks import constants, parse_model
+from dbt.adapters.databricks.catalogs._relation import DatabricksCatalogRelation
+
+
+class HiveMetastoreCatalogIntegration(CatalogIntegration):
+    catalog_type = constants.HIVE_METASTORE_CATALOG_TYPE
+    allows_writes = True
+
+    def __init__(self, config: CatalogIntegrationConfig) -> None:
+        super().__init__(config)
+        self.file_format: str = config.adapter_properties.get("file_format")
+
+    @property
+    def location_root(self) -> Optional[str]:
+        """
+        Volumes in Databricks are non-tabular datasets, which is something
+        different from what we mean by "external_volume" in dbt.
+        However, the protocol expects "external_volume" to be set.
+        """
+        return self.external_volume
+
+    @location_root.setter
+    def location_root(self, value: Optional[str]) -> None:
+        self.external_volume = value
+
+    def build_relation(self, model: RelationConfig) -> DatabricksCatalogRelation:
+        """
+        Args:
+            model: `config.model` (not `model`) from the jinja context
+        """
+        return DatabricksCatalogRelation(
+            catalog_type=self.catalog_type,
+            catalog_name=self.catalog_name,
+            table_format=parse_model.table_format(model) or self.table_format,
+            file_format=parse_model.file_format(model) or self.file_format,
+            external_volume=parse_model.location_root(model) or self.external_volume,
+            location_path=parse_model.location_path(model),
+        )

dbt/adapters/databricks/catalogs/_relation.py

@@ -0,0 +1,49 @@
+import posixpath
+from dataclasses import dataclass
+from typing import Optional
+
+from dbt_common.exceptions import DbtConfigError
+
+from dbt.adapters.databricks import constants
+
+
+@dataclass
+class DatabricksCatalogRelation:
+    catalog_type: str = constants.DEFAULT_CATALOG.catalog_type
+    catalog_name: Optional[str] = constants.DEFAULT_CATALOG.name
+    table_format: Optional[str] = constants.DEFAULT_CATALOG.adapter_properties.get("table_format")
+    file_format: Optional[str] = constants.DEFAULT_CATALOG.adapter_properties.get("file_format")
+    external_volume: Optional[str] = constants.DEFAULT_CATALOG.external_volume
+    location_path: Optional[str] = None
+
+    @property
+    def location_root(self) -> Optional[str]:
+        """
+        Volumes in Databricks are non-tabular datasets, which is something
+        different from what we mean by "external_volume" in dbt.
+        However, the protocol expects "external_volume" to be set.
+        """
+        return self.external_volume
+
+    @location_root.setter
+    def location_root(self, value: Optional[str]) -> None:
+        self.external_volume = value
+
+    @property
+    def location(self) -> Optional[str]:
+        if self.location_root and self.location_path:
+            return posixpath.join(self.location_root, self.location_path)
+        return None
+
+    @property
+    def iceberg_table_properties(self) -> dict[str, str]:
+        if self.table_format == constants.ICEBERG_TABLE_FORMAT:
+            if self.file_format != constants.DELTA_FILE_FORMAT:
+                raise DbtConfigError(
+                    "When table_format is 'iceberg', cannot set file_format to other than delta."
+                )
+            return {
+                "delta.enableIcebergCompatV2": "true",
+                "delta.universalFormat.enabledFormats": constants.ICEBERG_TABLE_FORMAT,
+            }
+        return {}

dbt/adapters/databricks/catalogs/_unity.py

@@ -0,0 +1,44 @@
+from typing import Optional
+
+from dbt.adapters.catalogs import CatalogIntegration, CatalogIntegrationConfig
+from dbt.adapters.contracts.relation import RelationConfig
+from dbt.adapters.databricks import constants, parse_model
+from dbt.adapters.databricks.catalogs._relation import DatabricksCatalogRelation
+
+
+class UnityCatalogIntegration(CatalogIntegration):
+    catalog_type = constants.UNITY_CATALOG_TYPE
+    allows_writes = True
+
+    def __init__(self, config: CatalogIntegrationConfig) -> None:
+        super().__init__(config)
+        if location_root := config.adapter_properties.get("location_root"):
+            self.external_volume: Optional[str] = location_root
+        self.file_format: str = config.adapter_properties.get("file_format")
+
+    @property
+    def location_root(self) -> Optional[str]:
+        """
+        Volumes in Databricks are non-tabular datasets, which is something
+        different from what we mean by "external_volume" in dbt.
+        However, the protocol expects "external_volume" to be set.
+        """
+        return self.external_volume
+
+    @location_root.setter
+    def location_root(self, value: Optional[str]) -> None:
+        self.external_volume = value
+
+    def build_relation(self, model: RelationConfig) -> DatabricksCatalogRelation:
+        """
+        Args:
+            model: `config.model` (not `model`) from the jinja context
+        """
+        return DatabricksCatalogRelation(
+            catalog_type=self.catalog_type,
+            catalog_name=self.catalog_name,
+            table_format=parse_model.table_format(model) or self.table_format,
+            file_format=parse_model.file_format(model) or self.file_format,
+            external_volume=parse_model.location_root(model) or self.external_volume,
+            location_path=parse_model.location_path(model),
+        )

dbt/adapters/databricks/constants.py

@@ -0,0 +1,38 @@
+from types import SimpleNamespace
+
+DEFAULT_TABLE_FORMAT = "default"
+ICEBERG_TABLE_FORMAT = "iceberg"
+
+
+DELTA_FILE_FORMAT = "delta"
+PARQUET_FILE_FORMAT = "parquet"
+HUDI_FILE_FORMAT = "hudi"
+
+
+UNITY_CATALOG_TYPE = "unity"
+HIVE_METASTORE_CATALOG_TYPE = "hive_metastore"
+
+
+DEFAULT_UNITY_CATALOG = SimpleNamespace(
+    name="unity",
+    catalog_name="default_unity",
+    catalog_type=UNITY_CATALOG_TYPE,
+    # requires the model to specify the external_volume (location_root) property
+    external_volume=None,
+    table_format=DEFAULT_TABLE_FORMAT,
+    adapter_properties={
+        "file_format": DELTA_FILE_FORMAT,
+    },
+)
+DEFAULT_HIVE_METASTORE_CATALOG = SimpleNamespace(
+    name="hive_metastore",
+    catalog_name="default_hive_metastore",
+    catalog_type=HIVE_METASTORE_CATALOG_TYPE,
+    # requires the model to specify the external_volume (location_root) property
+    external_volume=None,
+    table_format=DEFAULT_TABLE_FORMAT,
+    adapter_properties={
+        "file_format": DELTA_FILE_FORMAT,
+    },
+)
+DEFAULT_CATALOG = DEFAULT_UNITY_CATALOG

dbt/adapters/databricks/constraints.py

@@ -1,4 +1,5 @@
 from abc import ABC, abstractmethod
+from dataclasses import dataclass
 from typing import Any, ClassVar, Optional, TypeVar
 from uuid import uuid4
 
@@ -27,6 +28,7 @@
 T = TypeVar("T", bound="TypedConstraint")
 
 
+@dataclass
 class TypedConstraint(ModelLevelConstraint, ABC):
     """Constraint that enforces type because it has render logic"""
 
@@ -61,6 +63,19 @@ def _render_error(self, missing: list[list[str]]) -> DbtValidationError:
             f"{self.str_type} constraint '{name}' is missing required field(s): {fields}"
         )
 
+    # Enables set equality checks, especially for convenient unit testing
+    def __hash__(self) -> int:
+        # Create a tuple of all the fields that should be used for equality comparison
+        fields = (
+            self.type,
+            self.name,
+            tuple(self.columns) if self.columns else None,
+            self.expression,
+            self.to,
+            tuple(self.to_columns) if self.to_columns else None,
+        )
+        return hash(fields)
+
 
 class CustomConstraint(TypedConstraint):
     str_type = "custom"
@@ -203,7 +218,7 @@ def parse_model_constraints(
         if constraint["type"] == ConstraintType.not_null:
             if not constraint.get("columns"):
                 raise DbtValidationError("not_null constraint on model must have 'columns' defined")
-            column_names.add(*constraint["columns"])
+            column_names.update(constraint["columns"])
         else:
             constraints.append(parse_constraint(constraint))