Repository: databricks/dbt-databricks
Pull request #1012 (2025-05-07 17:01:13)

Migrate to the dbt catalog framework
<!-- Please review our pull request review process in CONTRIBUTING.md before your proceed. -->

Resolves #

<!---
  Include the number of the issue addressed by this PR above if applicable.

  Example:
    resolves #1234

  Please review our pull request review process in CONTRIBUTING.md before your proceed.
-->

### Description

dbt will support catalogs and catalog integrations as stand-alone objects, just like relations.
Since `dbt-databricks` already had the concept of catalogs, there is some work to migrate the existing concept into the generic catalog concept. This PR attempts to migrate to the new framework while maintaining backwards compatibility. Points of discussion:

- We need to support existing user behavior for all macros and all methods made available (`@available`) on the adapter in the jinja context.
- We make the assumption that every relation has a corresponding catalog integration, and a catalog relation within this catalog integration as a result. This means we need to supply a "default" catalog integration for all existing materializations that use tables (e.g. `table`, `incremental`, `snapshot`, `seed`).
- We are not supporting python models yet.
- dbt combines the concepts of a metadata catalog (e.g. Iceberg), a file format (e.g. Parquet), and an external volume (e.g. DBX's `location`). Models can override this if needed, but the idea is that the catalog integration can carry defaults for these settings.

### 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.



Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 77227116..9347a420 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,6 +3,7 @@
 ### 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
 
diff --git a/dbt/adapters/databricks/catalogs/__init__.py b/dbt/adapters/databricks/catalogs/__init__.py
new file mode 100644
index 00000000..a6f1bed3
--- /dev/null
+++ b/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",
+]
diff --git a/dbt/adapters/databricks/catalogs/_hive_metastore.py b/dbt/adapters/databricks/catalogs/_hive_metastore.py
new file mode 100644
index 00000000..d7dc0107
--- /dev/null
+++ b/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),
+        )
diff --git a/dbt/adapters/databricks/catalogs/_relation.py b/dbt/adapters/databricks/catalogs/_relation.py
new file mode 100644
index 00000000..c892d3ce
--- /dev/null
+++ b/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 {}
diff --git a/dbt/adapters/databricks/catalogs/_unity.py b/dbt/adapters/databricks/catalogs/_unity.py
new file mode 100644
index 00000000..7b92f373
--- /dev/null
+++ b/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),
+        )
diff --git a/dbt/adapters/databricks/constants.py b/dbt/adapters/databricks/constants.py
new file mode 100644
index 00000000..b8fe8b14
--- /dev/null
+++ b/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
diff --git a/dbt/adapters/databricks/impl.py b/dbt/adapters/databricks/impl.py
index 074540b3..b846039d 100644
--- a/dbt/adapters/databricks/impl.py
+++ b/dbt/adapters/databricks/impl.py
@@ -23,14 +23,20 @@ from dbt.adapters.base.impl import catch_as_completed, log_code_execution
 from dbt.adapters.base.meta import available
 from dbt.adapters.base.relation import BaseRelation
 from dbt.adapters.capability import Capability, CapabilityDict, CapabilitySupport, Support
+from dbt.adapters.catalogs import CatalogRelation
 from dbt.adapters.contracts.connection import AdapterResponse, Connection
 from dbt.adapters.contracts.relation import RelationConfig, RelationType
-from dbt.adapters.databricks import constraints
+from dbt.adapters.databricks import constants, constraints, parse_model
 from dbt.adapters.databricks.behaviors.columns import (
     GetColumnsBehavior,
     GetColumnsByDescribe,
     GetColumnsByInformationSchema,
 )
+from dbt.adapters.databricks.catalogs import (
+    DatabricksCatalogRelation,
+    HiveMetastoreCatalogIntegration,
+    UnityCatalogIntegration,
+)
 from dbt.adapters.databricks.column import DatabricksColumn
 from dbt.adapters.databricks.connections import DatabricksConnectionManager
 from dbt.adapters.databricks.global_state import GlobalState
@@ -180,6 +186,10 @@ class DatabricksAdapter(SparkAdapter):
         }
     )
 
+    CATALOG_INTEGRATIONS = [
+        HiveMetastoreCatalogIntegration,
+        UnityCatalogIntegration,
+    ]
     CONSTRAINT_SUPPORT = constraints.CONSTRAINT_SUPPORT
 
     get_column_behavior: GetColumnsBehavior
@@ -196,6 +206,9 @@ class DatabricksAdapter(SparkAdapter):
         except CompilationError:
             pass
 
+        self.add_catalog_integration(constants.DEFAULT_UNITY_CATALOG)
+        self.add_catalog_integration(constants.DEFAULT_HIVE_METASTORE_CATALOG)
+
     @property
     def _behavior_flags(self) -> list[BehaviorFlag]:
         return [USE_INFO_SCHEMA_FOR_COLUMNS, USE_USER_FOLDER_FOR_PYTHON, USE_MATERIALIZATION_V2]
@@ -205,20 +218,18 @@ class DatabricksAdapter(SparkAdapter):
         self, config: BaseConfig, tblproperties: Optional[dict[str, str]] = None
     ) -> dict[str, str]:
         result = tblproperties or config.get("tblproperties", {})
-        if config.get("table_format") == TableFormat.ICEBERG:
+
+        catalog_relation: DatabricksCatalogRelation = self.build_catalog_relation(config.model)  # type:ignore
+        if catalog_relation.table_format == constants.ICEBERG_TABLE_FORMAT:
             if self.compare_dbr_version(14, 3) < 0:
                 raise DbtConfigError("Iceberg support requires Databricks Runtime 14.3 or later.")
-            if config.get("file_format", "delta") != "delta":
-                raise DbtConfigError(
-                    "When table_format is 'iceberg', cannot set file_format to other than delta."
-                )
             if config.get("materialized") not in ("incremental", "table", "snapshot"):
                 raise DbtConfigError(
                     "When table_format is 'iceberg', materialized must be 'incremental'"
                     ", 'table', or 'snapshot'."
                 )
-            result["delta.enableIcebergCompatV2"] = "true"
-            result["delta.universalFormat.enabledFormats"] = "iceberg"
+            result.update(catalog_relation.iceberg_table_properties)
+
         return result
 
     @available.parse(lambda *a, **k: 0)
@@ -820,6 +831,26 @@ class DatabricksAdapter(SparkAdapter):
     def clean_sql(self, sql: str) -> str:
         return SqlUtils.clean_sql(sql)
 
+    @available
+    def build_catalog_relation(self, model: RelationConfig) -> Optional[CatalogRelation]:
+        """
+        Builds a relation for a given configuration.
+
+        This method uses the provided configuration to determine the appropriate catalog
+        integration and config parser for building the relation. It defaults to the built-in Delta
+        catalog if none is provided in the configuration for backward compatibility.
+
+        Args:
+            model (RelationConfig): `config.model` (not `model`) from the jinja context
+
+        Returns:
+            Any: The relation object generated through the catalog integration and parser
+        """
+        if catalog := parse_model.catalog_name(model):
+            catalog_integration = self.get_catalog_integration(catalog)
+            return catalog_integration.build_relation(model)
+        return None
+
 
 @dataclass(frozen=True)
 class RelationAPIBase(ABC, Generic[DatabricksRelationConfig]):
diff --git a/dbt/adapters/databricks/parse_model.py b/dbt/adapters/databricks/parse_model.py
new file mode 100644
index 00000000..2eda40dc
--- /dev/null
+++ b/dbt/adapters/databricks/parse_model.py
@@ -0,0 +1,48 @@
+import posixpath
+from typing import Optional
+
+from dbt.adapters.contracts.relation import RelationConfig
+from dbt.adapters.databricks import constants
+
+
+def catalog_name(model: RelationConfig) -> Optional[str]:
+    """
+    The user has not set the catalog, which is common as it's the legacy behavior.
+    We need to infer a catalog based on the model config since macros now expect it.
+    Luckily, all catalog attribution in the legacy behavior is on the model.
+    This means we can take a default catalog and rely on the model overrides to supply the rest.
+    """
+    return _get(model, "catalog") or constants.DEFAULT_CATALOG.name
+
+
+def file_format(model: RelationConfig) -> Optional[str]:
+    return _get(model, "file_format")
+
+
+def location_path(model: RelationConfig) -> Optional[str]:
+    if not model.config:
+        return None
+
+    if model.config.get("include_full_name_in_path"):
+        return posixpath.join(model.database, model.schema, model.identifier)
+    return model.identifier
+
+
+def location_root(model: RelationConfig) -> Optional[str]:
+    return _get(model, "location_root")
+
+
+def table_format(model: RelationConfig) -> Optional[str]:
+    return _get(model, "table_format")
+
+
+def _get(
+    model: RelationConfig, setting: str, case_sensitive: Optional[bool] = False
+) -> Optional[str]:
+    if not model.config:
+        return None
+
+    if value := model.config.get(setting):
+        return value if case_sensitive else value.casefold()
+
+    return None
diff --git a/dbt/adapters/databricks/relation_configs/table_format.py b/dbt/adapters/databricks/relation_configs/table_format.py
index 10f4fb88..907e69c5 100644
--- a/dbt/adapters/databricks/relation_configs/table_format.py
+++ b/dbt/adapters/databricks/relation_configs/table_format.py
@@ -1,5 +1,7 @@
 from dbt_common.dataclass_schema import StrEnum
 
+from dbt.adapters.databricks import constants
+
 
 class TableFormat(StrEnum):
     """
@@ -8,8 +10,8 @@ class TableFormat(StrEnum):
     simplify things for users.
     """
 
-    DEFAULT = "default"
-    ICEBERG = "iceberg"
+    DEFAULT = constants.DEFAULT_TABLE_FORMAT
+    ICEBERG = constants.ICEBERG_TABLE_FORMAT
 
     def __str__(self) -> str:
         return self.value
diff --git a/dbt/include/databricks/macros/materializations/incremental/incremental.sql b/dbt/include/databricks/macros/materializations/incremental/incremental.sql
index 0d3c861a..21e8bba6 100644
--- a/dbt/include/databricks/macros/materializations/incremental/incremental.sql
+++ b/dbt/include/databricks/macros/materializations/incremental/incremental.sql
@@ -1,16 +1,17 @@
 {% materialization incremental, adapter='databricks', supported_languages=['sql', 'python'] -%}
   {{ log("MATERIALIZING INCREMENTAL") }}
-  
+
+  {%- set catalog_relation = adapter.build_catalog_relation(config.model) -%}
+
   {% set existing_relation = load_relation_with_metadata(this) %}
   {% set target_relation = this.incorporate(type='table') %}
-  {% set file_format = get_file_format() %}
-  {% set incremental_strategy = get_incremental_strategy(file_format) %}
+  {% set incremental_strategy = get_incremental_strategy(catalog_relation.file_format) %}
   {% set grant_config = config.get('grants') %}
   {% set full_refresh = should_full_refresh() %}
   {% set partition_by = config.get('partition_by') %}
   {% set language = model['language'] %}
   {% set on_schema_change = incremental_validate_on_schema_change(config.get('on_schema_change'), default='ignore') %}
-  {% set is_delta = (file_format == 'delta' and existing_relation.is_delta) %}
+  {% set is_delta = (catalog_relation.file_format == 'delta' and existing_relation.is_delta) %}
   {% set compiled_code = adapter.clean_sql(model['compiled_code']) %}
 
   {% if adapter.behavior.use_materialization_v2 %}
diff --git a/dbt/include/databricks/macros/materializations/seeds/helpers.sql b/dbt/include/databricks/macros/materializations/seeds/helpers.sql
index d1ddc997..486a71d4 100644
--- a/dbt/include/databricks/macros/materializations/seeds/helpers.sql
+++ b/dbt/include/databricks/macros/materializations/seeds/helpers.sql
@@ -60,6 +60,9 @@
 {% endmacro %}
 
 {% macro create_or_replace_csv_table(model, agate_table, replace=False) %}
+
+  {%- set catalog_relation = adapter.build_catalog_relation(config.model) -%}
+
   {%- set column_override = model['config'].get('column_types', {}) -%}
   {%- set quote_seed_column = model['config'].get('quote_columns', None) -%}
   {%- set column_comment = config.persist_column_docs() and model.columns %}
@@ -86,10 +89,10 @@
             {{ adapter.quote_seed_column(column_name, quote_seed_column) }} {{ type }} {{ column_comment_clause }}{%- if not loop.last -%}, {%- endif -%}
         {%- endfor -%}
     )
-    {{ file_format_clause() }}
+    {{ file_format_clause(catalog_relation) }}
     {{ partition_cols(label="partitioned by") }}
     {{ clustered_cols(label="clustered by") }}
-    {{ location_clause(relation) }}
+    {{ location_clause(catalog_relation) }}
     {{ comment_clause() }}
     {{ tblproperties_clause() }}
   {% endset %}
diff --git a/dbt/include/databricks/macros/relations/file_format.sql b/dbt/include/databricks/macros/relations/file_format.sql
index 78313044..bff39204 100644
--- a/dbt/include/databricks/macros/relations/file_format.sql
+++ b/dbt/include/databricks/macros/relations/file_format.sql
@@ -1,9 +1,28 @@
-{% macro file_format_clause() %}
-  {%- set file_format = config.get('file_format', default='delta') -%}
+{% macro file_format_clause(catalog_relation=none) %}
+  {#--
+    Moving forward, this macro should require a `catalog_relation`, which is covered by the first condition.
+    However, there could be existing macros that is still passing no arguments, including user macros.
+    Hence, we need to support the old code still, which is covered by the second condition.
+  --#}
+  {% if catalog_relation is not none %}
+    {%- set file_format = catalog_relation.file_format -%}
+  {% else %}
+    {%- set file_format = config.get('file_format', default='delta') -%}
+  {% endif %}
   using {{ file_format }}
 {%- endmacro -%}
 
-{% macro get_file_format() %}
-  {%- set raw_file_format = config.get('file_format', default='delta') -%}
+
+{% macro get_file_format(catalog_relation=none) %}
+  {#-
+    Moving forward, this macro should require a `catalog_relation`, which is covered by the first condition.
+    However, there could be existing macros that is still passing no arguments, including user macros.
+    Hence, we need to support the old code still, which is covered by the second condition.
+  -#}
+  {% if catalog_relation is not none %}
+    {%- set raw_file_format = catalog_relation.file_format -%}
+  {% else %}
+    {%- set raw_file_format = config.get('file_format', default='delta') -%}
+  {% endif %}
   {% do return(dbt_databricks_validate_get_file_format(raw_file_format)) %}
-{% endmacro %}
\ No newline at end of file
+{% endmacro %}
diff --git a/dbt/include/databricks/macros/relations/location.sql b/dbt/include/databricks/macros/relations/location.sql
index b4079a3f..05792274 100644
--- a/dbt/include/databricks/macros/relations/location.sql
+++ b/dbt/include/databricks/macros/relations/location.sql
@@ -1,4 +1,17 @@
 {% macro location_clause(relation) %}
+  {#--
+    Moving forward, `relation` should be a `CatalogRelation`, which is covered by the first condition.
+    However, there could be existing macros that are still passing in a `BaseRelation`, including user macros.
+    Hence, we need to support the old code still, which is covered by the second condition.
+  --#}
+  {%- if relation.catalog_type is not none -%}
+
+    {%- if relation.location is not none -%}
+    location '{{ relation.location }}{% if is_incremental() %}_tmp{% endif %}'
+    {%- endif -%}
+
+  {%- else -%}
+
   {%- set location_root = config.get('location_root', validator=validation.any[basestring]) -%}
   {%- set file_format = config.get('file_format', default='delta') -%}
   {%- set identifier = model['alias'] -%}
@@ -10,4 +23,6 @@
         'Incompatible configuration: `location_root` must be set when using a non-delta file format with Unity Catalog'
     ) }}
   {%- endif %}
+
+  {%- endif %}
 {%- endmacro -%}
diff --git a/dbt/include/databricks/macros/relations/table/create.sql b/dbt/include/databricks/macros/relations/table/create.sql
index dc56bf7e..0c7a5a96 100644
--- a/dbt/include/databricks/macros/relations/table/create.sql
+++ b/dbt/include/databricks/macros/relations/table/create.sql
@@ -19,36 +19,40 @@
 {% endmacro %}
 
 {% macro get_create_table_sql(target_relation, columns, compiled_code) %}
-  {%- set file_format = config.get('file_format', default='delta') -%}
+
+  {%- set catalog_relation = adapter.build_catalog_relation(config.model) -%}
+
   {%- set contract = config.get('contract') -%}
   {%- set contract_enforced = contract and contract.enforced -%}
   {%- if contract_enforced -%}
     {{ get_assert_columns_equivalent(compiled_code) }}
   {%- endif -%}
 
-  {%- if file_format == 'delta' %}
+  {%- if catalog_relation.file_format == 'delta' %}
   create or replace table {{ target_relation.render() }}
   {% else %}
   create table {{ target_relation.render() }}
   {% endif -%}
   {{ get_column_and_constraints_sql(target_relation, columns) }}
-  {{ file_format_clause() }}
-  {{ options_clause() }}
+  {{ file_format_clause(catalog_relation) }}
+  {{ databricks__options_clause(catalog_relation) }}
   {{ partition_cols(label="partitioned by") }}
   {{ liquid_clustered_cols() }}
   {{ clustered_cols(label="clustered by") }}
-  {{ location_clause(target_relation) }}
+  {{ location_clause(catalog_relation) }}
   {{ comment_clause() }}
   {{ tblproperties_clause() }}
 {% endmacro %}
 
 {% macro databricks__create_table_as(temporary, relation, compiled_code, language='sql') -%}
+
+  {%- set catalog_relation = adapter.build_catalog_relation(config.model) -%}
+
   {%- if language == 'sql' -%}
     {%- if temporary -%}
       {{ create_temporary_view(relation, compiled_code) }}
     {%- else -%}
-      {%- set file_format = config.get('file_format', default='delta') -%}
-      {% if file_format == 'delta' %}
+      {% if catalog_relation.file_format == 'delta' %}
         create or replace table {{ relation.render() }}
       {% else %}
         create table {{ relation.render() }}
@@ -58,12 +62,12 @@
         {{ get_assert_columns_equivalent(compiled_code) }}
         {%- set compiled_code = get_select_subquery(compiled_code) %}
       {% endif %}
-      {{ file_format_clause() }}
-      {{ options_clause() }}
+      {{ file_format_clause(catalog_relation) }}
+      {{ databricks__options_clause(catalog_relation) }}
       {{ partition_cols(label="partitioned by") }}
       {{ liquid_clustered_cols() }}
       {{ clustered_cols(label="clustered by") }}
-      {{ location_clause(relation) }}
+      {{ location_clause(catalog_relation) }}
       {{ comment_clause() }}
       {{ tblproperties_clause() }}
       as
@@ -81,9 +85,23 @@
   {%- endif -%}
 {%- endmacro -%}
 
-{% macro databricks__options_clause() -%}
+{% macro databricks__options_clause(catalog_relation=none) -%}
+  {#-
+    Moving forward, this macro should require a `catalog_relation`, which is covered by the first condition.
+    However, there could be existing macros that is still passing no arguments, including user macros.
+    Hence, we need to support the old code still, which is covered by the second condition.
+    Additionally, since this rolls up to `options_clause` in `dbt-spark`, which does not have any arguments,
+    all calls to `options_clause` will take the second path. This macro needs to be called directly
+    via `databricks__options_clause`.
+  -#}
+  {%- if catalog_relation is not none -%}
+    {%- set file_format = catalog_relation.file_format -%}
+  {%- else -%}
+    {%- set file_format = config.get('file_format', default='delta') -%}
+  {%- endif -%}
+
   {%- set options = config.get('options') -%}
-  {%- if config.get('file_format', default='delta') == 'hudi' -%}
+  {%- if file_format == 'hudi' -%}
     {%- set unique_key = config.get('unique_key') -%}
     {%- if unique_key is not none and options is none -%}
       {%- set options = {'primaryKey': config.get('unique_key')} -%}
diff --git a/pyproject.toml b/pyproject.toml
index 15735f00..b6976b94 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -24,7 +24,7 @@ classifiers = [
 dependencies = [
     "databricks-sdk>=0.41, <0.48.0",
     "databricks-sql-connector[pyarrow]>=4.0.0, <5.0.0",
-    "dbt-adapters>=1.9.0, <2.0",
+    "dbt-adapters>=1.14.3, <2.0",
     "dbt-common>=1.12.0, <2.0",
     "dbt-core>=1.9.0, <2.0",
     "dbt-spark>=1.9.0, <2.0",


Test patch:
diff --git a/tests/functional/adapter/hooks/test_model_hooks.py b/tests/functional/adapter/hooks/test_model_hooks.py
index bf1a4e6c..0d51783e 100644
--- a/tests/functional/adapter/hooks/test_model_hooks.py
+++ b/tests/functional/adapter/hooks/test_model_hooks.py
@@ -11,7 +11,7 @@ class TestPrePostModelHooks(BaseTestPrePost):
     def setUp(self, project):
         util.run_sql_with_adapter(
             project.adapter,
-            f"drop table if exists { project.test_schema }.on_model_hook",
+            f"drop table if exists {project.test_schema}.on_model_hook",
         )
         util.run_sql_with_adapter(project.adapter, override_fixtures.create_table_statement)
 
diff --git a/tests/functional/adapter/hooks/test_run_hooks.py b/tests/functional/adapter/hooks/test_run_hooks.py
index 1f133d86..5a4ca47e 100644
--- a/tests/functional/adapter/hooks/test_run_hooks.py
+++ b/tests/functional/adapter/hooks/test_run_hooks.py
@@ -10,10 +10,10 @@ from tests.functional.adapter.hooks import fixtures as override_fixtures
 class TestPrePostRunHooks(BasePrePostRunHooks):
     @pytest.fixture(scope="function")
     def setUp(self, project):
-        project.run_sql(f"drop table if exists { project.test_schema }.on_run_hook")
+        project.run_sql(f"drop table if exists {project.test_schema}.on_run_hook")
         util.run_sql_with_adapter(project.adapter, override_fixtures.create_table_run_statement)
-        project.run_sql(f"drop table if exists { project.test_schema }.schemas")
-        project.run_sql(f"drop table if exists { project.test_schema }.db_schemas")
+        project.run_sql(f"drop table if exists {project.test_schema}.schemas")
+        project.run_sql(f"drop table if exists {project.test_schema}.db_schemas")
         os.environ["TERM_TEST"] = "TESTING"
 
     @pytest.fixture(scope="class")
diff --git a/tests/unit/macros/relations/test_table_macros.py b/tests/unit/macros/relations/test_table_macros.py
index 8116d349..8f972b92 100644
--- a/tests/unit/macros/relations/test_table_macros.py
+++ b/tests/unit/macros/relations/test_table_macros.py
@@ -1,8 +1,28 @@
+from typing import Optional
+
 import pytest
 
+from dbt.adapters.databricks import catalogs, constants
 from tests.unit.macros.base import MacroTestBase
 
 
+def unity_relation(
+    table_format: Optional[str] = None,
+    file_format: Optional[str] = None,
+    location_root: Optional[str] = None,
+    location_path: Optional[str] = None,
+) -> catalogs.DatabricksCatalogRelation:
+    catalog_integration = constants.DEFAULT_UNITY_CATALOG
+    return catalogs.DatabricksCatalogRelation(
+        catalog_type=catalog_integration.catalog_type,
+        catalog_name=catalog_integration.catalog_name,
+        table_format=table_format or catalog_integration.table_format,
+        file_format=file_format or catalog_integration.adapter_properties.get("file_format"),
+        external_volume=location_root or catalog_integration.external_volume,
+        location_path=location_path,
+    )
+
+
 class TestCreateTableAs(MacroTestBase):
     @pytest.fixture(scope="class")
     def template_name(self) -> str:
@@ -40,6 +60,7 @@ class TestCreateTableAs(MacroTestBase):
         )
 
     def test_macros_create_table_as(self, template_bundle):
+        template_bundle.context["adapter"].build_catalog_relation.return_value = unity_relation()
         sql = self.render_create_table_as(template_bundle)
         assert (
             sql == f"create or replace table {template_bundle.relation.render()}"
@@ -47,10 +68,11 @@ class TestCreateTableAs(MacroTestBase):
         )
 
     def test_macros_create_table_as_with_iceberg(self, template_bundle):
-        template_bundle.context["adapter"].update_tblproperties_for_iceberg.return_value = {
-            "delta.enableIcebergCompatV2": "true",
-            "delta.universalFormat.enabledFormats": "iceberg",
-        }
+        catalog_relation = unity_relation(table_format=constants.ICEBERG_TABLE_FORMAT)
+        template_bundle.context["adapter"].build_catalog_relation.return_value = catalog_relation
+        template_bundle.context[
+            "adapter"
+        ].update_tblproperties_for_iceberg.return_value = catalog_relation.iceberg_table_properties  # type: ignore
         sql = self.render_create_table_as(template_bundle)
         assert sql == self.clean_sql(
             f"create or replace table {template_bundle.relation.render()} using delta"
@@ -58,10 +80,14 @@ class TestCreateTableAs(MacroTestBase):
             "'delta.universalFormat.enabledFormats' = 'iceberg') as select 1"
         )
 
-    @pytest.mark.parametrize("format", ["parquet", "hudi"])
+    @pytest.mark.parametrize("format", [constants.PARQUET_FILE_FORMAT, constants.HUDI_FILE_FORMAT])
     def test_macros_create_table_as_file_format(self, format, config, template_bundle):
-        config["file_format"] = format
-        config["location_root"] = "/mnt/root"
+        catalog_relation = unity_relation(
+            file_format=format,
+            location_root="/mnt/root",
+            location_path=template_bundle.relation.identifier,
+        )
+        template_bundle.context["adapter"].build_catalog_relation.return_value = catalog_relation
         sql = self.render_create_table_as(template_bundle)
         expected = (
             f"create table {template_bundle.relation.render()} using {format} location "
@@ -70,6 +96,7 @@ class TestCreateTableAs(MacroTestBase):
         assert sql == expected
 
     def test_macros_create_table_as_options(self, config, template_bundle):
+        template_bundle.context["adapter"].build_catalog_relation.return_value = unity_relation()
         config["options"] = {"compression": "gzip"}
         sql = self.render_create_table_as(template_bundle)
         expected = self.clean_sql(
@@ -80,9 +107,14 @@ class TestCreateTableAs(MacroTestBase):
         assert sql == expected
 
     def test_macros_create_table_as_hudi_unique_key(self, config, template_bundle):
-        config["file_format"] = "hudi"
+        catalog_relation = unity_relation(
+            file_format=constants.HUDI_FILE_FORMAT,
+            location_root="/mnt/root",
+            location_path=template_bundle.relation.identifier,
+        )
+        template_bundle.context["adapter"].build_catalog_relation.return_value = catalog_relation
+
         config["unique_key"] = "id"
-        config["location_root"] = "/mnt/root"
         sql = self.render_create_table_as(template_bundle, sql="select 1 as id")
 
         expected = self.clean_sql(
@@ -96,10 +128,14 @@ class TestCreateTableAs(MacroTestBase):
     def test_macros_create_table_as_hudi_unique_key_primary_key_match(
         self, config, template_bundle
     ):
-        config["file_format"] = "hudi"
+        catalog_relation = unity_relation(
+            file_format=constants.HUDI_FILE_FORMAT,
+            location_root="/mnt/root",
+            location_path=template_bundle.relation.identifier,
+        )
+        template_bundle.context["adapter"].build_catalog_relation.return_value = catalog_relation
         config["unique_key"] = "id"
         config["options"] = {"primaryKey": "id"}
-        config["location_root"] = "/mnt/root"
         sql = self.render_create_table_as(template_bundle, sql="select 1 as id")
 
         expected = self.clean_sql(
@@ -112,13 +148,15 @@ class TestCreateTableAs(MacroTestBase):
     def test_macros_create_table_as_hudi_unique_key_primary_key_mismatch(
         self, config, template_bundle
     ):
-        config["file_format"] = "hudi"
+        catalog_relation = unity_relation(file_format=constants.HUDI_FILE_FORMAT)
+        template_bundle.context["adapter"].build_catalog_relation.return_value = catalog_relation
         config["unique_key"] = "uuid"
         config["options"] = {"primaryKey": "id"}
         sql = self.render_create_table_as(template_bundle, sql="select 1 as id, 2 as uuid")
         assert "mock.raise_compiler_error()" in sql
 
     def test_macros_create_table_as_partition(self, config, template_bundle):
+        template_bundle.context["adapter"].build_catalog_relation.return_value = unity_relation()
         config["partition_by"] = "partition_1"
         sql = self.render_create_table_as(template_bundle)
 
@@ -129,6 +167,7 @@ class TestCreateTableAs(MacroTestBase):
         assert sql == expected
 
     def test_macros_create_table_as_partitions(self, config, template_bundle):
+        template_bundle.context["adapter"].build_catalog_relation.return_value = unity_relation()
         config["partition_by"] = ["partition_1", "partition_2"]
         sql = self.render_create_table_as(template_bundle)
         expected = (
@@ -139,6 +178,7 @@ class TestCreateTableAs(MacroTestBase):
         assert sql == expected
 
     def test_macros_create_table_as_cluster(self, config, template_bundle):
+        template_bundle.context["adapter"].build_catalog_relation.return_value = unity_relation()
         config["clustered_by"] = "cluster_1"
         config["buckets"] = "1"
         sql = self.render_create_table_as(template_bundle)
@@ -151,6 +191,7 @@ class TestCreateTableAs(MacroTestBase):
         assert sql == expected
 
     def test_macros_create_table_as_clusters(self, config, template_bundle):
+        template_bundle.context["adapter"].build_catalog_relation.return_value = unity_relation()
         config["clustered_by"] = ["cluster_1", "cluster_2"]
         config["buckets"] = "1"
         sql = self.render_create_table_as(template_bundle)
@@ -163,6 +204,7 @@ class TestCreateTableAs(MacroTestBase):
         assert sql == expected
 
     def test_macros_create_table_as_liquid_cluster(self, config, template_bundle):
+        template_bundle.context["adapter"].build_catalog_relation.return_value = unity_relation()
         config["liquid_clustered_by"] = "cluster_1"
         sql = self.render_create_table_as(template_bundle)
         expected = self.clean_sql(
@@ -173,6 +215,7 @@ class TestCreateTableAs(MacroTestBase):
         assert sql == expected
 
     def test_macros_create_table_as_liquid_clusters(self, config, template_bundle):
+        template_bundle.context["adapter"].build_catalog_relation.return_value = unity_relation()
         config["liquid_clustered_by"] = ["cluster_1", "cluster_2"]
         config["buckets"] = "1"
         sql = self.render_create_table_as(template_bundle)
@@ -184,6 +227,7 @@ class TestCreateTableAs(MacroTestBase):
         assert sql == expected
 
     def test_macros_create_table_as_liquid_cluster_auto(self, config, template_bundle):
+        template_bundle.context["adapter"].build_catalog_relation.return_value = unity_relation()
         config["auto_liquid_cluster"] = True
         sql = self.render_create_table_as(template_bundle)
         expected = self.clean_sql(
@@ -194,6 +238,7 @@ class TestCreateTableAs(MacroTestBase):
         assert sql == expected
 
     def test_macros_create_table_as_comment(self, config, template_bundle):
+        template_bundle.context["adapter"].build_catalog_relation.return_value = unity_relation()
         config["persist_docs"] = {"relation": True}
         template_bundle.context["model"].description = "Description Test"
 
@@ -207,7 +252,13 @@ class TestCreateTableAs(MacroTestBase):
         assert sql == expected
 
     def test_macros_create_table_as_all_delta(self, config, template_bundle):
-        config["location_root"] = "/mnt/root"
+        catalog_relation = unity_relation(
+            file_format=constants.DELTA_FILE_FORMAT,
+            location_root="/mnt/root",
+            location_path=template_bundle.relation.identifier,
+        )
+        template_bundle.context["adapter"].build_catalog_relation.return_value = catalog_relation
+
         config["partition_by"] = ["partition_1", "partition_2"]
         config["liquid_clustered_by"] = ["cluster_1", "cluster_2"]
         config["clustered_by"] = ["cluster_1", "cluster_2"]
@@ -218,7 +269,6 @@ class TestCreateTableAs(MacroTestBase):
         }
         template_bundle.context["model"].description = "Description Test"
 
-        config["file_format"] = "delta"
         sql = self.render_create_table_as(template_bundle)
 
         expected = self.clean_sql(
@@ -236,7 +286,13 @@ class TestCreateTableAs(MacroTestBase):
         assert sql == expected
 
     def test_macros_create_table_as_all_hudi(self, config, template_bundle):
-        config["location_root"] = "/mnt/root"
+        catalog_relation = unity_relation(
+            file_format=constants.HUDI_FILE_FORMAT,
+            location_root="/mnt/root",
+            location_path=template_bundle.relation.identifier,
+        )
+        template_bundle.context["adapter"].build_catalog_relation.return_value = catalog_relation
+
         config["partition_by"] = ["partition_1", "partition_2"]
         config["clustered_by"] = ["cluster_1", "cluster_2"]
         config["buckets"] = "1"
@@ -246,7 +302,6 @@ class TestCreateTableAs(MacroTestBase):
         }
         template_bundle.context["model"].description = "Description Test"
 
-        config["file_format"] = "hudi"
         sql = self.render_create_table_as(template_bundle)
 
         expected = self.clean_sql(