feat: Replace information schema with describe calls (#1432)

### Description
Replace expensive `information_schema` queries with a single `DESCRIBE
TABLE EXTENDED ... AS JSON` call for metadata fetching in
`_describe_relation` methods. Gated behind
`DBRCapability.DESCRIBE_TABLE_EXTENDED_AS_JSON` (DBR 17.3+), falls back
to existing info_schema queries on older runtimes.
   
**Changes:**
                  
- `IncrementalTableAPI._describe_relation`: replaces 4 info_schema
queries (PK, FK, non-null constraints, column masks) with AS JSON
parsing
- `MaterializedViewAPI._describe_relation`: replaces
`get_view_description` (info_schema.views) with AS JSON parsing
- `ViewAPI._describe_relation`: same as MV
- New `DatabricksDescribeJsonMetadata` parser class with composite PK/FK
support
- New `is_describe_as_json_supported()` gating method (checks HMS,
foreign table, capability)
- New `is_foreign_table` property on `DatabricksRelation`
  - New `describe_table_extended_as_json` Jinja macro
  **Testing:**
- 56 unit tests: parser, processor roundtrip, diff validation, edge
cases
  - Capability boundary tests (17.2 vs 17.3)
- `is_describe_as_json_supported` unit tests (UC, HMS, foreign table, no
capability)
- Jinja macro SQL generation test
  - Functional tests validated on both code paths:
- SQL warehouse (DBR 17.3+): all 14 constraint/mask tests passed via
DESCRIBE AS JSON path
- UC cluster (DBR 16.4 LTS): all 14 tests passed via information_schema
fallback path
### 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.
   

PECOBLR-2546

---------

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

Author: tejassp-db

CHANGELOG.md

@@ -6,17 +6,15 @@
 - 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))
+- Fetch relation metadata like constraints, column masks, row filters, etc with a single `DESCRIBE TABLE EXTENDED ... AS JSON` call, replacing multiple `information_schema` queries. Falls back to `information_schema` on older runtimes. Gated behind `use_describe_as_json_for_relation_metadata` behavior flag, off by default. ([#1432](https://github.com/databricks/dbt-databricks/pull/1432))
 - 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 `metric_view` failing with `METRIC_VIEW_INVALID_VIEW_DEFINITION` when models use bare `{{ ref(...) }}` for the `source:` field ([#1361](https://github.com/databricks/dbt-databricks/issues/1361))
 - 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
-
-- Fix `metric_view` failing with `METRIC_VIEW_INVALID_VIEW_DEFINITION` when models use bare `{{ ref(...) }}` for the `source:` field ([#1361](https://github.com/databricks/dbt-databricks/issues/1361))
-
 ### 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.

dbt/adapters/databricks/dbr_capabilities.py

@@ -13,13 +13,14 @@
 class DBRCapability(Enum):
     """Named capabilities that depend on DBR version."""
 
-    TIMESTAMPDIFF = "timestampdiff"
-    ICEBERG = "iceberg"
     COMMENT_ON_COLUMN = "comment_on_column"
-    JSON_COLUMN_METADATA = "json_column_metadata"
-    STREAMING_TABLE_JSON_METADATA = "streaming_table_json_metadata"
+    DESCRIBE_TABLE_EXTENDED_AS_JSON = "describe_table_extended_as_json"
+    ICEBERG = "iceberg"
     INSERT_BY_NAME = "insert_by_name"
+    JSON_COLUMN_METADATA = "json_column_metadata"
     REPLACE_ON = "replace_on"
+    STREAMING_TABLE_JSON_METADATA = "streaming_table_json_metadata"
+    TIMESTAMPDIFF = "timestampdiff"
 
 
 @dataclass
@@ -61,6 +62,9 @@ class DBRCapabilities:
         DBRCapability.REPLACE_ON: CapabilitySpec(
             min_version=(17, 1),
         ),
+        DBRCapability.DESCRIBE_TABLE_EXTENDED_AS_JSON: CapabilitySpec(
+            min_version=(17, 3),
+        ),
     }
 
     def __init__(

dbt/adapters/databricks/impl.py

@@ -150,6 +150,10 @@ def is_metric_view(self) -> bool:
     def is_streaming_table(self) -> bool:
         return self.type == DatabricksRelationType.StreamingTable
 
+    @property
+    def is_foreign_table(self) -> bool:
+        return self.type == DatabricksRelationType.Foreign
+
     @property
     def is_external_table(self) -> bool:
         return self.databricks_table_type == DatabricksTableType.External

dbt/adapters/databricks/relation.py

@@ -129,9 +129,17 @@ SELECT
     NULL
   ) AS databricks_table_type
 FROM `system`.`information_schema`.`tables`
-WHERE table_catalog = '{{ relation.database|lower }}' 
+WHERE table_catalog = '{{ relation.database|lower }}'
   AND table_schema = '{{ relation.schema|lower }}'
   {%- if relation.identifier %}
   AND table_name = '{{ relation.identifier|lower }}'
   {% endif %}
 {% endmacro %}
+
+{% macro describe_table_extended_as_json(relation) %}
+    {{ return(run_query_as(describe_table_extended_as_json_sql(relation), 'describe_table_extended_as_json')) }}
+{% endmacro %}
+
+{% macro describe_table_extended_as_json_sql(relation) %}
+DESCRIBE TABLE EXTENDED {{ relation.render() }} AS JSON
+{% endmacro %}

dbt/include/databricks/macros/adapters/metadata.sql

@@ -1,4 +1,7 @@
 import pytest
+from dbt.tests import util
+
+from dbt.adapters.databricks.dbr_capabilities import DBRCapability
 
 fail_if_tag_fetch_called_macros = """
 {% macro fetch_tags(relation) %}
@@ -33,3 +36,13 @@ class ManagedIcebergMixin:
     @pytest.fixture(scope="class")
     def project_config_update(self):
         return {"flags": {"use_managed_iceberg": True}}
+
+
+class RequiresDescribeAsJsonCapabilityMixin:
+    """Skip the test class if the connected compute lacks DESCRIBE TABLE EXTENDED AS JSON."""
+
+    @pytest.fixture(scope="class", autouse=True)
+    def require_describe_as_json_capability(self, project):
+        with util.get_connection(project.adapter):
+            if not project.adapter.has_capability(DBRCapability.DESCRIBE_TABLE_EXTENDED_AS_JSON):
+                pytest.skip("DESCRIBE TABLE EXTENDED AS JSON not supported on this compute")

tests/functional/adapter/fixtures.py

@@ -1,7 +1,10 @@
 import pytest
 from dbt.tests import util
 
-from tests.functional.adapter.fixtures import MaterializationV2Mixin
+from tests.functional.adapter.fixtures import (
+    MaterializationV2Mixin,
+    RequiresDescribeAsJsonCapabilityMixin,
+)
 from tests.functional.adapter.incremental import fixtures
 
 
@@ -61,3 +64,17 @@ def test_changing_column_masks(self, project):
         assert result[0][1] == "hello"  # name (unmasked)
         assert result[0][2] == "********@example.com"  # email (partially masked)
         assert result[0][3] == "*****"  # password (masked)
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestIncrementalColumnMasksDescribeJsonOn(
+    RequiresDescribeAsJsonCapabilityMixin, TestIncrementalColumnMasks
+):
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "flags": {
+                "use_materialization_v2": True,
+                "use_describe_as_json_for_relation_metadata": True,
+            }
+        }

tests/functional/adapter/incremental/test_incremental_column_masks.py

@@ -2,6 +2,7 @@
 from dbt.contracts.results import RunStatus
 from dbt.tests import util
 
+from tests.functional.adapter.fixtures import RequiresDescribeAsJsonCapabilityMixin
 from tests.functional.adapter.incremental import fixtures
 
 
@@ -30,6 +31,20 @@ def test_add_non_null_constraint(self, project):
         assert "DELTA_NOT_NULL_CONSTRAINT_VIOLATED" in results.results[0].message
 
 
+@pytest.mark.skip_profile("databricks_cluster")
+class TestIncrementalSetNonNullConstraintDescribeJsonOn(
+    RequiresDescribeAsJsonCapabilityMixin, TestIncrementalSetNonNullConstraint
+):
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "flags": {
+                "use_materialization_v2": True,
+                "use_describe_as_json_for_relation_metadata": True,
+            }
+        }
+
+
 @pytest.mark.skip_profile("databricks_cluster")
 class TestIncrementalUnsetNonNullConstraint:
     @pytest.fixture(scope="class")
@@ -175,6 +190,20 @@ def test_update_primary_key_constraint(self, project):
         assert not any(constraint[0] == "pk_model" for constraint in primary_key_constraints)
 
 
+@pytest.mark.skip_profile("databricks_cluster")
+class TestIncrementalUpdatePrimaryKeyConstraintDescribeJsonOn(
+    RequiresDescribeAsJsonCapabilityMixin, TestIncrementalUpdatePrimaryKeyConstraint
+):
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "flags": {
+                "use_materialization_v2": True,
+                "use_describe_as_json_for_relation_metadata": True,
+            }
+        }
+
+
 @pytest.mark.skip_profile("databricks_cluster")
 class TestCascadingConstraintDrop:
     @pytest.fixture(scope="class")

tests/functional/adapter/incremental/test_incremental_constraints.py

@@ -9,11 +9,13 @@
     MaterializedViewChangesContinueMixin,
     MaterializedViewChangesFailMixin,
 )
+from dbt_common.contracts.config.materialization import OnConfigurationChangeOption
 
 from dbt.adapters.databricks.relation_configs.materialized_view import (
     MaterializedViewConfig,
 )
 from dbt.adapters.databricks.relation_configs.tblproperties import TblPropertiesConfig
+from tests.functional.adapter.fixtures import RequiresDescribeAsJsonCapabilityMixin
 from tests.functional.adapter.materialized_view_tests import fixtures
 
 
@@ -86,6 +88,29 @@ class TestMaterializedViewApplyChanges(
     pass
 
 
+@pytest.mark.dlt
+@pytest.mark.skip_profile("databricks_cluster", "databricks_uc_cluster")
+class TestMaterializedViewApplyChangesDescribeJsonOn(
+    RequiresDescribeAsJsonCapabilityMixin, TestMaterializedViewApplyChanges
+):
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "models": {"on_configuration_change": OnConfigurationChangeOption.Apply.value},
+            "flags": {"use_describe_as_json_for_relation_metadata": True},
+        }
+
+    @pytest.mark.skip(reason="Full-refresh bypasses get_configuration_changes(); JSON path unused")
+    def test_full_refresh_occurs_with_changes(self):
+        pass
+
+    @pytest.mark.skip(
+        reason="Alter test only mutates refresh schedule. view_text is parsed but not asserted on"
+    )
+    def test_change_is_applied_via_alter(self):
+        pass
+
+
 @pytest.mark.dlt
 @pytest.mark.skip_profile("databricks_cluster", "databricks_uc_cluster")
 class TestMaterializedViewContinueOnChanges(

tests/functional/adapter/materialized_view_tests/test_changes.py

@@ -1,7 +1,11 @@
 import pytest
 from dbt.tests.util import run_dbt, write_file
 
-from tests.functional.adapter.fixtures import MaterializationV1Mixin, MaterializationV2Mixin
+from tests.functional.adapter.fixtures import (
+    MaterializationV1Mixin,
+    MaterializationV2Mixin,
+    RequiresDescribeAsJsonCapabilityMixin,
+)
 from tests.functional.adapter.row_filters.fixtures import (
     base_model_mv,
     base_model_sql,
@@ -142,6 +146,20 @@ def test_incremental_row_filter_lifecycle(self, project):
         assert len(filters) == 0
 
 
+@pytest.mark.skip_profile("databricks_cluster")
+class TestIncrementalRowFilterDescribeJsonOn(
+    RequiresDescribeAsJsonCapabilityMixin, TestIncrementalRowFilter
+):
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "flags": {
+                "use_materialization_v2": True,
+                "use_describe_as_json_for_relation_metadata": True,
+            }
+        }
+
+
 @pytest.mark.skip_profile("databricks_cluster", "databricks_uc_cluster")
 class TestMaterializedViewRowFilter(RowFilterMixin):
     """Test row filters on materialized view models."""
@@ -177,6 +195,20 @@ def test_mv_row_filter_lifecycle(self, project):
         assert len(filters) == 0
 
 
+@pytest.mark.skip_profile("databricks_cluster", "databricks_uc_cluster")
+class TestMaterializedViewRowFilterDescribeJsonOn(
+    RequiresDescribeAsJsonCapabilityMixin, TestMaterializedViewRowFilter
+):
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "flags": {
+                "use_materialization_v2": True,
+                "use_describe_as_json_for_relation_metadata": True,
+            }
+        }
+
+
 @pytest.mark.skip_profile("databricks_cluster", "databricks_uc_cluster")
 class TestStreamingTableRowFilter(RowFilterMixin):
     """Test row filters on streaming table models."""
@@ -228,6 +260,20 @@ def test_streaming_table_row_filter_lifecycle(self, project):
         assert len(filters) == 0
 
 
+@pytest.mark.skip_profile("databricks_cluster", "databricks_uc_cluster")
+class TestStreamingTableRowFilterDescribeJsonOn(
+    RequiresDescribeAsJsonCapabilityMixin, TestStreamingTableRowFilter
+):
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "flags": {
+                "use_materialization_v2": True,
+                "use_describe_as_json_for_relation_metadata": True,
+            }
+        }
+
+
 @pytest.mark.skip_profile("databricks_cluster")
 class TestViewRowFilterFailure(MaterializationV2Mixin):
     """Test that row filters on regular views fail with clear error."""