[FIX]: Fix issue with cloning of external tables (databricks#1095)

Author: ericj-db

CHANGELOG.md

@@ -1,10 +1,15 @@
 ## dbt-databricks 1.10.5 (TBD)
 
+### Features
+
+- Add cloning support for the the external tables (thanks @samgans!) ([1079](https://github.com/databricks/dbt-databricks/pull/1079))
+
 ### Fixes
 
 - Fix inefficient query when getting column schema for MV/STs ([1074](https://github.com/databricks/dbt-databricks/issues/1074))
 - Fix bug causing false positives in diffing constraints between existing relation and model config for incremental runs ([1081](https://github.com/databricks/dbt-databricks/issues/1081))
 - Fix bug causing "main is not being called during running model" errors for some view updates ([1077](https://github.com/databricks/dbt-databricks/issues/1077))
+- Fix the bugs with external tabls cloning [1095](https://github.com/databricks/dbt-databricks/pull/1095) (thanks @samgans!)
 
 ### Under the Hood
 
@@ -16,7 +21,6 @@
 
 - Support column tags (with Materialization V2) ([649](https://github.com/databricks/dbt-databricks/issues/649))
 - Support column masking (with Materialization V2) ([670](https://github.com/databricks/dbt-databricks/issues/670))
-- Add cloning support for the the external tables (thanks @samgans!) ([1079](https://github.com/databricks/dbt-databricks/pull/1079))
 
 ### Fixes
 

dbt/adapters/databricks/impl.py

@@ -8,7 +8,7 @@
 from dataclasses import dataclass
 from importlib import metadata
 from multiprocessing.context import SpawnContext
-from typing import TYPE_CHECKING, Any, ClassVar, Generic, Optional, Union, cast
+from typing import TYPE_CHECKING, Any, ClassVar, Generic, NamedTuple, Optional, Union, cast
 from uuid import uuid4
 
 from dbt_common.behavior_flags import BehaviorFlag
@@ -123,6 +123,14 @@
 )  # type: ignore[typeddict-item]
 
 
+class DatabricksRelationInfo(NamedTuple):
+    table_name: str
+    table_type: str
+    file_format: Optional[str]
+    table_owner: Optional[str]
+    databricks_table_type: Optional[str]
+
+
 @dataclass
 class DatabricksConfig(AdapterConfig):
     file_format: str = "delta"
@@ -326,23 +334,30 @@ def execute(
     def list_relations_without_caching(  # type: ignore[override]
         self, schema_relation: DatabricksRelation
     ) -> list[DatabricksRelation]:
-        empty: list[tuple[Optional[str], Optional[str], Optional[str], Optional[str]]] = []
+        empty: list[DatabricksRelationInfo] = []
         results = handle_missing_objects(
             lambda: self.get_relations_without_caching(schema_relation), empty
         )
 
         relations = []
         for row in results:
-            name, kind, file_format, owner = row
+            name, kind, file_format, owner, table_type = row
             metadata = None
             if file_format:
                 metadata = {KEY_TABLE_OWNER: owner, KEY_TABLE_PROVIDER: file_format}
+
+            if table_type:
+                databricks_table_type = DatabricksRelation.get_databricks_table_type(table_type)
+            else:
+                databricks_table_type = None
+
             relations.append(
                 DatabricksRelation.create(
                     database=schema_relation.database,
                     schema=schema_relation.schema,
                     identifier=name,
                     type=DatabricksRelation.get_relation_type(kind),
+                    databricks_table_type=databricks_table_type,
                     metadata=metadata,
                     is_delta=file_format == "delta",
                 )
@@ -352,24 +367,26 @@ def list_relations_without_caching(  # type: ignore[override]
 
     def get_relations_without_caching(
         self, relation: DatabricksRelation
-    ) -> list[tuple[Optional[str], Optional[str], Optional[str], Optional[str]]]:
+    ) -> list[DatabricksRelationInfo]:
         if relation.is_hive_metastore():
             return self._get_hive_relations(relation)
         return self._get_uc_relations(relation)
 
-    def _get_uc_relations(
-        self, relation: DatabricksRelation
-    ) -> list[tuple[Optional[str], Optional[str], Optional[str], Optional[str]]]:
+    def _get_uc_relations(self, relation: DatabricksRelation) -> list[DatabricksRelationInfo]:
         kwargs = {"relation": relation}
         results = self.execute_macro("get_uc_tables", kwargs=kwargs)
         return [
-            (row["table_name"], row["table_type"], row["file_format"], row["table_owner"])
+            DatabricksRelationInfo(
+                row["table_name"],
+                row["table_type"],
+                row["file_format"],
+                row["table_owner"],
+                row["databricks_table_type"],
+            )
             for row in results
         ]
 
-    def _get_hive_relations(
-        self, relation: DatabricksRelation
-    ) -> list[tuple[Optional[str], Optional[str], Optional[str], Optional[str]]]:
+    def _get_hive_relations(self, relation: DatabricksRelation) -> list[DatabricksRelationInfo]:
         kwargs = {"relation": relation}
 
         new_rows: list[tuple[str, Optional[str]]]
@@ -383,8 +400,8 @@ def _get_hive_relations(
             for row in tables:
                 # list_tables returns TABLE_TYPE as view for both materialized views and for
                 # streaming tables.  Set type to "" in this case and it will be resolved below.
-                type = row["TABLE_TYPE"].lower() if row["TABLE_TYPE"] else None
-                row = (row["TABLE_NAME"], type)
+                rel_type = row["TABLE_TYPE"].lower() if row["TABLE_TYPE"] else None
+                row = (row["TABLE_NAME"], rel_type)
                 new_rows.append(row)
 
         else:
@@ -401,7 +418,16 @@ def _get_hive_relations(
                     for row in new_rows
                 ]
 
-        return [(row[0], row[1], None, None) for row in new_rows]
+        return [
+            DatabricksRelationInfo(
+                row[0],
+                row[1],  # type: ignore[arg-type]
+                None,
+                None,
+                None,
+            )
+            for row in new_rows
+        ]
 
     @available.parse(lambda *a, **k: [])
     def get_column_schema_from_query(self, sql: str) -> list[DatabricksColumn]:
@@ -495,6 +521,7 @@ def _get_updated_relation(
                 schema=relation.schema,
                 identifier=relation.identifier,
                 type=relation.type,  # type: ignore
+                databricks_table_type=relation.databricks_table_type,
                 metadata=metadata,
                 is_delta=metadata.get(KEY_TABLE_PROVIDER) == "delta",
             ),

dbt/adapters/databricks/relation.py

@@ -40,11 +40,15 @@ class DatabricksRelationType(StrEnum):
     MaterializedView = "materialized_view"
     Foreign = "foreign"
     StreamingTable = "streaming_table"
+    MetricView = "metric_view"
+    Unknown = "unknown"
+
+
+class DatabricksTableType(StrEnum):
     External = "external"
+    Managed = "managed"
     ManagedShallowClone = "managed_shallow_clone"
     ExternalShallowClone = "external_shallow_clone"
-    MetricView = "metric_view"
-    Unknown = "unknown"
 
 
 @dataclass(frozen=True, eq=False, repr=False)
@@ -69,6 +73,7 @@ class DatabricksRelation(BaseRelation):
     metadata: Optional[dict[str, Any]] = None
     renameable_relations = (DatabricksRelationType.Table, DatabricksRelationType.View)
     replaceable_relations = (DatabricksRelationType.Table, DatabricksRelationType.View)
+    databricks_table_type: Optional[DatabricksTableType] = None
 
     @classmethod
     def __pre_deserialize__(cls, data: dict[Any, Any]) -> dict[Any, Any]:
@@ -95,7 +100,7 @@ def is_streaming_table(self) -> bool:
 
     @property
     def is_external_table(self) -> bool:
-        return self.type == DatabricksRelationType.External
+        return self.databricks_table_type == DatabricksTableType.External
 
     @property
     def is_dlt(self) -> bool:
@@ -158,6 +163,10 @@ def matches(
     def get_relation_type(cls) -> Type[DatabricksRelationType]:  # noqa
         return DatabricksRelationType
 
+    @classproperty
+    def get_databricks_table_type(cls) -> Type[DatabricksTableType]:  # noqa
+        return DatabricksTableType
+
     def information_schema(self, view_name: Optional[str] = None) -> InformationSchema:
         # some of our data comes from jinja, where things can be `Undefined`.
         if not isinstance(view_name, str):

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

@@ -96,11 +96,21 @@ SELECT
   table_name,
   if(table_type IN ('EXTERNAL', 'MANAGED', 'MANAGED_SHALLOW_CLONE', 'EXTERNAL_SHALLOW_CLONE'), 'table', lower(table_type)) AS table_type,
   lower(data_source_format) AS file_format,
-  table_owner
+  table_owner,
+  if(
+    table_type IN (
+      'EXTERNAL',
+      'MANAGED',
+      'MANAGED_SHALLOW_CLONE',
+      'EXTERNAL_SHALLOW_CLONE'
+    ),
+    lower(table_type),
+    NULL
+  ) AS databricks_table_type
 FROM `system`.`information_schema`.`tables`
 WHERE table_catalog = '{{ relation.database|lower }}' 
   AND table_schema = '{{ relation.schema|lower }}'
   {%- if relation.identifier %}
   AND table_name = '{{ relation.identifier|lower }}'
   {% endif %}
-{% endmacro %}
+{% endmacro %}

dbt/include/databricks/macros/materializations/clone/clone.sql

@@ -38,7 +38,7 @@
       {% endif %}
 
       -- as a general rule, data platforms that can clone tables can also do atomic 'create or replace'
-      {% if other_existing_relation.is_external %}
+      {% if other_existing_relation.is_external_table %}
           {% call statement('main') %}
               {{ create_or_replace_clone_external(target_relation, defer_relation) }}
           {% endcall %}

dbt/include/databricks/macros/materializations/clone/strategies.sql

@@ -4,11 +4,13 @@
     shallow clone {{ defer_relation.render() }}
 {% endmacro %}
 
-{% macro databricks__create_or_replace_clone_external(this_relation, defer_relation) %}
+{% macro create_or_replace_clone_external(this_relation, defer_relation) %}
+
     {%- set catalog_relation = adapter.build_catalog_relation(config.model) -%}
 
     create or replace
     table {{ this_relation.render() }}
     shallow clone {{ defer_relation.render() }}
     {{ location_clause(catalog_relation) }}
+
 {% endmacro %}

tests/unit/macros/adapters/test_metadata_macros.py

@@ -137,7 +137,8 @@ def test_get_uc_tables_sql_with_identifier(self, template_bundle, relation):
               table_name,
               if(table_type IN ('EXTERNAL', 'MANAGED', 'MANAGED_SHALLOW_CLONE', 'EXTERNAL_SHALLOW_CLONE'), 'table', lower(table_type)) AS table_type,
               lower(data_source_format) AS file_format,
-              table_owner
+              table_owner,
+              if(table_type IN ('EXTERNAL', 'MANAGED', 'MANAGED_SHALLOW_CLONE', 'EXTERNAL_SHALLOW_CLONE'), lower(table_type), null) AS databricks_table_type
             FROM `system`.`information_schema`.`tables`
             WHERE table_catalog = 'some_database'
               AND table_schema = 'some_schema'
@@ -155,9 +156,10 @@ def test_get_uc_tables_sql_without_identifier(self, template_bundle, mock_schema
               table_name,
               if(table_type IN ('EXTERNAL', 'MANAGED', 'MANAGED_SHALLOW_CLONE', 'EXTERNAL_SHALLOW_CLONE'), 'table', lower(table_type)) AS table_type,
               lower(data_source_format) AS file_format,
-              table_owner
+              table_owner,
+              if(table_type IN ('EXTERNAL', 'MANAGED', 'MANAGED_SHALLOW_CLONE', 'EXTERNAL_SHALLOW_CLONE'), lower(table_type), null) AS databricks_table_type
             FROM `system`.`information_schema`.`tables`
-            WHERE table_catalog = 'test_db' 
+            WHERE table_catalog = 'test_db'
               AND table_schema = 'test_schema'
         """  # noqa
         self.assert_sql_equal(result, expected_sql)
@@ -176,9 +178,10 @@ def test_case_sensitivity(self, template_bundle):
               table_name,
               if(table_type IN ('EXTERNAL', 'MANAGED', 'MANAGED_SHALLOW_CLONE', 'EXTERNAL_SHALLOW_CLONE'), 'table', lower(table_type)) AS table_type,
               lower(data_source_format) AS file_format,
-              table_owner
+              table_owner,
+              if(table_type IN ('EXTERNAL', 'MANAGED', 'MANAGED_SHALLOW_CLONE', 'EXTERNAL_SHALLOW_CLONE'), lower(table_type), null) AS databricks_table_type
             FROM `system`.`information_schema`.`tables`
-            WHERE table_catalog = 'test_db' 
+            WHERE table_catalog = 'test_db'
               AND table_schema = 'test_schema'
               AND table_name = 'test_table'
         """  # noqa

tests/unit/macros/materializations/test_clone_macros.py

@@ -73,7 +73,7 @@ def test_create_or_replace_clone(self, template_bundle, s_relation):
     def test_create_or_replace_clone_external(self, template_bundle, catalog_relation, s_relation):
         sql = self.render_clone_macro(
             template_bundle,
-            "databricks__create_or_replace_clone_external",
+            "create_or_replace_clone_external",
             s_relation,
             template_bundle.relation,
         )

tests/unit/test_adapter.py

@@ -13,8 +13,15 @@
 from dbt.adapters.databricks.credentials import (
     CATALOG_KEY_IN_SESSION_PROPERTIES,
 )
-from dbt.adapters.databricks.impl import get_identifier_list_string
-from dbt.adapters.databricks.relation import DatabricksRelation, DatabricksRelationType
+from dbt.adapters.databricks.impl import (
+    DatabricksRelationInfo,
+    get_identifier_list_string,
+)
+from dbt.adapters.databricks.relation import (
+    DatabricksRelation,
+    DatabricksRelationType,
+    DatabricksTableType,
+)
 from dbt.adapters.databricks.utils import check_not_found_error
 from dbt.config import RuntimeConfig
 from tests.unit.utils import config_from_parts_or_dicts
@@ -356,7 +363,9 @@ def test_list_relations_without_caching__no_relations(self, _):
     @patch("dbt.adapters.databricks.api_client.DatabricksApiClient.create")
     def test_list_relations_without_caching__some_relations(self, _):
         with patch.object(DatabricksAdapter, "get_relations_without_caching") as mocked:
-            mocked.return_value = [("name", "table", "hudi", "owner")]
+            mocked.return_value = [
+                DatabricksRelationInfo("name", "table", "hudi", "owner", "external")
+            ]
             adapter = DatabricksAdapter(Mock(flags={}), get_context("spawn"))
             relations = adapter.list_relations("database", "schema")
             assert len(relations) == 1
@@ -365,13 +374,15 @@ def test_list_relations_without_caching__some_relations(self, _):
             assert relation.database == "database"
             assert relation.schema == "schema"
             assert relation.type == DatabricksRelationType.Table
+            assert relation.databricks_table_type == DatabricksTableType.External
             assert relation.owner == "owner"
+            assert relation.is_external_table
             assert relation.is_hudi
 
     @patch("dbt.adapters.databricks.api_client.DatabricksApiClient.create")
     def test_list_relations_without_caching__hive_relation(self, _):
         with patch.object(DatabricksAdapter, "get_relations_without_caching") as mocked:
-            mocked.return_value = [("name", "table", None, None)]
+            mocked.return_value = [DatabricksRelationInfo("name", "table", None, None, None)]
             adapter = DatabricksAdapter(Mock(flags={}), get_context("spawn"))
             relations = adapter.list_relations("database", "schema")
             assert len(relations) == 1

tests/unit/test_relation.py

@@ -184,6 +184,12 @@ class TestRelationsFunctions:
     def test_is_hive_metastore(self, database, expected):
         assert relation.is_hive_metastore(database) is expected
 
+    def test_is_external_table(self):
+        relation = DatabricksRelation.create(
+            identifier="external_table", databricks_table_type="external"
+        )
+        assert relation.is_external_table is True
+
     @pytest.mark.parametrize(
         "input, expected",
         [