feat: catalogs.yml v2 support via bridge_v2_catalog hooks (#1440)

## Summary

Adds catalogs.yml v2 support to dbt-databricks using the adapter-owned
bridge architecture.

### What changed

**`impl.py`**
- Declares `Capability.CatalogsV2` — enables v2 path when
`use_catalogs_v2: true` is set in `dbt_project.yml`
- Adds `_v2_to_v1_type` hook: `unity → unity`, `hive_metastore →
hive_metastore`

**`catalogs/_unity.py`**
- Adeed warnings if `use_uniform` is set in config properties
- `location_root` must be non-blank if provided

**`catalogs/_hive_metastore.py`**
- Adds `file_format` enum validation in
`HiveMetastoreCatalogIntegration.__init__`: must be one of `delta |
parquet | hudi`

### Architecture

dbt-core parses `catalogs.yml` v2 structurally, checks
`Capability.CatalogsV2`, then calls
`adapter.bridge_v2_catalog(catalog)`. The base adapter template method
reads `catalog.config.get("databricks", {})` and calls `_v2_to_v1_type`
to resolve the v1 catalog_type. Field-level validation happens in
`CatalogIntegration.__init__` — same as v1, no new interface.

Companion PRs:
- dbt-core: dbt-labs/dbt-core#12930
- dbt-adapters: dbt-labs/dbt-adapters#1920

## Test plan

- [x] Unit tests for `UnityCatalogIntegration.__init__` validation
- [x] Unit tests for `HiveMetastoreCatalogIntegration.__init__`
validation
- [ ] Functional tests end-to-end once dbt-core and dbt-adapters PRs are
merged

---------

Co-authored-by: Shubham Dhal <shubham.dhal@databricks.com>

Author: aahel

CHANGELOG.md

@@ -1,5 +1,9 @@
 ## dbt-databricks 1.12.2 (TBD)
 
+### Features
+
+- Add catalogs.yml v2 support (requires `use_catalogs_v2: true` in dbt-core) ([1440](https://github.com/databricks/dbt-databricks/pull/1440))
+
 ### Under the Hood
 
 - Raise the `dbt-adapters` upper bound to `<1.25.0` ([#1507](https://github.com/databricks/dbt-databricks/pull/1507))

dbt/adapters/databricks/catalogs/_unity.py

@@ -2,9 +2,11 @@
 
 from dbt.adapters.catalogs import CatalogIntegration, CatalogIntegrationConfig
 from dbt.adapters.contracts.relation import RelationConfig
+from dbt_common.exceptions import DbtValidationError
 
 from dbt.adapters.databricks import constants, parse_model
 from dbt.adapters.databricks.catalogs._relation import DatabricksCatalogRelation
+from dbt.adapters.databricks.logging import logger
 
 
 class UnityCatalogIntegration(CatalogIntegration):
@@ -13,9 +15,20 @@ class UnityCatalogIntegration(CatalogIntegration):
 
     def __init__(self, config: CatalogIntegrationConfig) -> None:
         super().__init__(config)
-        if location_root := config.adapter_properties.get("location_root"):
+        location_root = config.adapter_properties.get("location_root")
+        if location_root is not None:
+            if not str(location_root).strip():
+                raise DbtValidationError(
+                    f"Catalog '{config.name}' unity/databricks location_root cannot be blank"
+                )
             self.external_volume: Optional[str] = location_root
         self.file_format: Optional[str] = config.file_format
+        if config.adapter_properties.get("use_uniform") is not None:
+            logger.warning(
+                f"Catalog '{config.name}': use_uniform is not yet supported by the adapter "
+                "and has no effect. Use the use_managed_iceberg behavior flag to control "
+                "Iceberg table creation. Support for use_uniform will be added in a future release."
+            )
 
     @property
     def location_root(self) -> Optional[str]:

dbt/adapters/databricks/impl.py

@@ -236,6 +236,17 @@ def get_identifier_list_string(table_names: set[str]) -> str:
     return _identifier
 
 
+def _adapter_capabilities() -> CapabilityDict:
+    capabilities: dict[Capability, CapabilitySupport] = {
+        Capability.TableLastModifiedMetadata: CapabilitySupport(support=Support.Full),
+        Capability.SchemaMetadataByRelations: CapabilitySupport(support=Support.Full),
+    }
+    catalogs_v2 = getattr(Capability, "CatalogsV2", None)
+    if catalogs_v2 is not None:
+        capabilities[catalogs_v2] = CapabilitySupport(support=Support.Full)
+    return CapabilityDict(capabilities)
+
+
 class DatabricksAdapter(SparkAdapter):
     INFORMATION_COMMENT_REGEX = re.compile(r"Comment: (.*)\n[A-Z][A-Za-z ]+:", re.DOTALL)
 
@@ -248,17 +259,16 @@ class DatabricksAdapter(SparkAdapter):
 
     AdapterSpecificConfigs = DatabricksConfig  # type: ignore[assignment]
 
-    _capabilities = CapabilityDict(
-        {
-            Capability.TableLastModifiedMetadata: CapabilitySupport(support=Support.Full),
-            Capability.SchemaMetadataByRelations: CapabilitySupport(support=Support.Full),
-        }
-    )
+    _capabilities = _adapter_capabilities()
 
     CATALOG_INTEGRATIONS = [
         HiveMetastoreCatalogIntegration,
         UnityCatalogIntegration,
     ]
+    _V2_TO_V1_TYPE: ClassVar[dict[str, str]] = {
+        "unity": constants.UNITY_CATALOG_TYPE,
+        "hive_metastore": constants.HIVE_METASTORE_CATALOG_TYPE,
+    }
     CONSTRAINT_SUPPORT = constraints.CONSTRAINT_SUPPORT
 
     get_column_behavior: GetColumnsBehavior
@@ -297,6 +307,9 @@ def _has_dbr_capability_parse(self, capability_name: str) -> bool:
             return False
         return DBRCapabilities(is_sql_warehouse=True).has_capability(capability)
 
+    def _v2_to_v1_type(self, catalog_type: str) -> str:
+        return self._V2_TO_V1_TYPE.get(catalog_type, catalog_type)
+
     @property
     def _behavior_flags(self) -> list[BehaviorFlag]:
         return [

tests/unit/test_catalogs_v2.py

@@ -0,0 +1,75 @@
+from dataclasses import dataclass, field
+from typing import Any, Optional
+
+import pytest
+from dbt.adapters.capability import Capability, Support
+from dbt_common.exceptions import DbtValidationError
+
+from dbt.adapters.databricks.catalogs import UnityCatalogIntegration
+from dbt.adapters.databricks.impl import DatabricksAdapter
+
+
+@dataclass
+class _Config:
+    """Minimal CatalogIntegrationConfig stub for testing __init__ validation."""
+
+    name: str = "test_cat"
+    catalog_type: str = "unity"
+    catalog_name: Optional[str] = None
+    table_format: Optional[str] = "iceberg"
+    external_volume: Optional[str] = None
+    file_format: Optional[str] = None
+    adapter_properties: dict[str, Any] = field(default_factory=dict)
+
+
+# ===== Adapter-level =====
+
+
+def test_catalogs_v2_capability_declared():
+    catalogs_v2 = getattr(Capability, "CatalogsV2", None)
+    if catalogs_v2 is None:
+        pytest.skip("CatalogsV2 not available in this dbt-adapters version")
+    cap = DatabricksAdapter._capabilities[catalogs_v2]
+    assert cap.support == Support.Full
+
+
+def test_v2_to_v1_type_unity():
+    adapter = object.__new__(DatabricksAdapter)
+    assert adapter._v2_to_v1_type("unity") == "unity"
+
+
+def test_v2_to_v1_type_hive_metastore():
+    adapter = object.__new__(DatabricksAdapter)
+    assert adapter._v2_to_v1_type("hive_metastore") == "hive_metastore"
+
+
+def test_v2_to_v1_type_unknown_passthrough():
+    adapter = object.__new__(DatabricksAdapter)
+    assert adapter._v2_to_v1_type("custom_type") == "custom_type"
+
+
+# ===== UnityCatalogIntegration =====
+
+
+def test_unity_parquet_without_uniform():
+    cfg = _Config(file_format="parquet")
+    integration = UnityCatalogIntegration(cfg)
+    assert integration.file_format == "parquet"
+
+
+def test_unity_with_location_root():
+    cfg = _Config(file_format="parquet", adapter_properties={"location_root": "/mnt/data"})
+    integration = UnityCatalogIntegration(cfg)
+    assert integration.external_volume == "/mnt/data"
+
+
+def test_unity_blank_location_root_raises():
+    cfg = _Config(file_format="parquet", adapter_properties={"location_root": "  "})
+    with pytest.raises(DbtValidationError, match="location_root cannot be blank"):
+        UnityCatalogIntegration(cfg)
+
+
+def test_unity_empty_location_root_raises():
+    cfg = _Config(file_format="parquet", adapter_properties={"location_root": ""})
+    with pytest.raises(DbtValidationError, match="location_root cannot be blank"):
+        UnityCatalogIntegration(cfg)