Repository: databricks/dbt-databricks
Pull request #968 (2025-03-19 16:42:23)

Moving liquid clustering to the check for difference approach
### Description

For incremental models, liquid clustering column setting was running on every incremental run.  We can use the check for difference approach basically for free, since it reuses data we were already gathering, so moving to that.

### 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 9ace91b4..6bf9abe9 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -16,6 +16,7 @@
 - Streamlining debug logging to make it more usable ([946](https://github.com/databricks/dbt-databricks/pull/946))
 - Upgrading Databricks SQL Connector to V4 ([962](https://github.com/databricks/dbt-databricks/pull/962))
 - Validation of sample mode ([961](https://github.com/databricks/dbt-databricks/pull/961))
+- Move liquid clustering behavior to on_config_change approach for incremental ([968](https://github.com/databricks/dbt-databricks/pull/968))
 
 ## dbt-databricks 1.9.8 (TBD)
 
diff --git a/dbt/adapters/databricks/credentials.py b/dbt/adapters/databricks/credentials.py
index c68d8704..3a8ad394 100644
--- a/dbt/adapters/databricks/credentials.py
+++ b/dbt/adapters/databricks/credentials.py
@@ -1,7 +1,6 @@
 import itertools
 import json
 import re
-import threading
 from collections.abc import Iterable
 from dataclasses import dataclass, field
 from typing import Any, Callable, Optional, cast
@@ -58,7 +57,6 @@ class DatabricksCredentials(Credentials):
     retry_all: bool = False
     connect_max_idle: Optional[int] = None
     _credentials_manager: Optional["DatabricksCredentialManager"] = None
-    _lock = threading.Lock()  # to avoid concurrent auth
 
     _ALIASES = {
         "catalog": "database",
@@ -329,60 +327,54 @@ class DatabricksCredentialManager(DataClassDictMixin):
         )
 
     def __post_init__(self) -> None:
-        self._lock = threading.Lock()
-        with self._lock:
-            if not hasattr(self, "_config"):
-                self._config: Optional[Config] = None
-            if self._config is not None:
-                return
-
-            if self.token:
-                self._config = self.authenticate_with_pat()
-            elif self.azure_client_id and self.azure_client_secret:
-                self._config = self.authenticate_with_azure_client_secret()
-            elif not self.client_secret:
-                self._config = self.authenticate_with_external_browser()
+        if not hasattr(self, "_config"):
+            self._config: Optional[Config] = None
+        if self._config is not None:
+            return
+
+        if self.token:
+            self._config = self.authenticate_with_pat()
+        elif self.azure_client_id and self.azure_client_secret:
+            self._config = self.authenticate_with_azure_client_secret()
+        elif not self.client_secret:
+            self._config = self.authenticate_with_external_browser()
+        else:
+            auth_methods = {
+                "oauth-m2m": self.authenticate_with_oauth_m2m,
+                "legacy-azure-client-secret": self.legacy_authenticate_with_azure_client_secret,
+            }
+
+            # If the secret starts with dose, high chance is it is a databricks secret
+            if self.client_secret.startswith("dose"):
+                auth_sequence = ["oauth-m2m", "legacy-azure-client-secret"]
             else:
-                auth_methods = {
-                    "oauth-m2m": self.authenticate_with_oauth_m2m,
-                    "legacy-azure-client-secret": self.legacy_authenticate_with_azure_client_secret,
-                }
-
-                # If the secret starts with dose, high chance is it is a databricks secret
-                if self.client_secret.startswith("dose"):
-                    auth_sequence = ["oauth-m2m", "legacy-azure-client-secret"]
-                else:
-                    auth_sequence = ["legacy-azure-client-secret", "oauth-m2m"]
-
-                exceptions = []
-                for i, auth_type in enumerate(auth_sequence):
-                    try:
-                        # The Config constructor will implicitly init auth and throw if failed
-                        self._config = auth_methods[auth_type]()
-                        if auth_type == "legacy-azure-client-secret":
-                            logger.warning(
-                                "You are using Azure Service Principal, "
-                                "please use 'azure_client_id' and 'azure_client_secret' instead."
-                            )
-                        break  # Exit loop if authentication is successful
-                    except Exception as e:
-                        exceptions.append((auth_type, e))
-                        next_auth_type = (
-                            auth_sequence[i + 1] if i + 1 < len(auth_sequence) else None
+                auth_sequence = ["legacy-azure-client-secret", "oauth-m2m"]
+
+            exceptions = []
+            for i, auth_type in enumerate(auth_sequence):
+                try:
+                    # The Config constructor will implicitly init auth and throw if failed
+                    self._config = auth_methods[auth_type]()
+                    if auth_type == "legacy-azure-client-secret":
+                        logger.warning(
+                            "You are using Azure Service Principal, "
+                            "please use 'azure_client_id' and 'azure_client_secret' instead."
+                        )
+                    break  # Exit loop if authentication is successful
+                except Exception as e:
+                    exceptions.append((auth_type, e))
+                    next_auth_type = auth_sequence[i + 1] if i + 1 < len(auth_sequence) else None
+                    if next_auth_type:
+                        logger.warning(
+                            f"Failed to authenticate with {auth_type}, "
+                            f"trying {next_auth_type} next. Error: {e}"
+                        )
+                    else:
+                        logger.error(
+                            f"Failed to authenticate with {auth_type}. "
+                            f"No more authentication methods to try. Error: {e}"
                         )
-                        if next_auth_type:
-                            logger.warning(
-                                f"Failed to authenticate with {auth_type}, "
-                                f"trying {next_auth_type} next. Error: {e}"
-                            )
-                        else:
-                            logger.error(
-                                f"Failed to authenticate with {auth_type}. "
-                                f"No more authentication methods to try. Error: {e}"
-                            )
-                            raise Exception(
-                                f"All authentication methods failed. Details: {exceptions}"
-                            )
+                        raise Exception(f"All authentication methods failed. Details: {exceptions}")
 
     @property
     def api_client(self) -> WorkspaceClient:
diff --git a/dbt/adapters/databricks/relation_configs/incremental.py b/dbt/adapters/databricks/relation_configs/incremental.py
index d32ea603..7c62fd13 100644
--- a/dbt/adapters/databricks/relation_configs/incremental.py
+++ b/dbt/adapters/databricks/relation_configs/incremental.py
@@ -1,9 +1,10 @@
 from dbt.adapters.databricks.relation_configs.base import (
     DatabricksRelationConfigBase,
 )
+from dbt.adapters.databricks.relation_configs.liquid_clustering import LiquidClusteringProcessor
 from dbt.adapters.databricks.relation_configs.tags import TagsProcessor
 from dbt.adapters.databricks.relation_configs.tblproperties import TblPropertiesProcessor
 
 
 class IncrementalTableConfig(DatabricksRelationConfigBase):
-    config_components = [TagsProcessor, TblPropertiesProcessor]
+    config_components = [TagsProcessor, TblPropertiesProcessor, LiquidClusteringProcessor]
diff --git a/dbt/adapters/databricks/relation_configs/liquid_clustering.py b/dbt/adapters/databricks/relation_configs/liquid_clustering.py
new file mode 100644
index 00000000..b55d1e40
--- /dev/null
+++ b/dbt/adapters/databricks/relation_configs/liquid_clustering.py
@@ -0,0 +1,56 @@
+from typing import ClassVar, Union
+
+from dbt.adapters.contracts.relation import RelationConfig
+from dbt.adapters.databricks.relation_configs import base
+from dbt.adapters.databricks.relation_configs.base import (
+    DatabricksComponentConfig,
+    DatabricksComponentProcessor,
+)
+from dbt.adapters.relation_configs.config_base import RelationResults
+
+
+class LiquidClusteringConfig(DatabricksComponentConfig):
+    """Component encapsulating the liquid clustering options."""
+
+    auto_cluster: bool = False
+    cluster_by: list[str] = []
+
+
+class LiquidClusteringProcessor(DatabricksComponentProcessor[LiquidClusteringConfig]):
+    name: ClassVar[str] = "liquid_clustering"
+
+    @classmethod
+    def from_relation_results(cls, results: RelationResults) -> LiquidClusteringConfig:
+        cluster_by_auto = False
+        cluster_by: list[str] = []
+        table = results["show_tblproperties"]
+        for row in table.rows:
+            if row[0] == "clusterByAuto":
+                cluster_by_auto = bool(row[1])
+            if row[0] == "clusteringColumns":
+                cluster_by = cls.extract_cluster_by(row[1])
+        return LiquidClusteringConfig(cluster_by=cluster_by, auto_cluster=cluster_by_auto)
+
+    @classmethod
+    def from_relation_config(cls, relation_config: RelationConfig) -> LiquidClusteringConfig:
+        liquid_clustered_by: Union[str, list[str], None] = base.get_config_value(
+            relation_config, "liquid_clustered_by"
+        )
+        cluster_by_auto: bool = bool(
+            base.get_config_value(relation_config, "auto_liquid_cluster") or False
+        )
+        if not liquid_clustered_by:
+            return LiquidClusteringConfig(cluster_by=[], auto_cluster=cluster_by_auto)
+        if isinstance(liquid_clustered_by, str):
+            return LiquidClusteringConfig(
+                cluster_by=[liquid_clustered_by], auto_cluster=cluster_by_auto
+            )
+        return LiquidClusteringConfig(cluster_by=liquid_clustered_by, auto_cluster=cluster_by_auto)
+
+    @staticmethod
+    def extract_cluster_by(cluster_by: str) -> list[str]:
+        if not cluster_by or cluster_by == "[]":
+            return []
+        trimmed = cluster_by[1:-1]
+        parts = trimmed.split(",")
+        return [part[2:-2] for part in parts]
diff --git a/dbt/adapters/databricks/relation_configs/tblproperties.py b/dbt/adapters/databricks/relation_configs/tblproperties.py
index d708c7f8..11c06d0b 100644
--- a/dbt/adapters/databricks/relation_configs/tblproperties.py
+++ b/dbt/adapters/databricks/relation_configs/tblproperties.py
@@ -19,7 +19,7 @@ class TblPropertiesConfig(DatabricksComponentConfig):
 
     # List of tblproperties that should be ignored when comparing configs. These are generally
     # set by Databricks and are not user-configurable.
-    ignore_list: list[str] = [
+    ignore_list: ClassVar[list[str]] = [
         "pipelines.pipelineId",
         "delta.enableChangeDataFeed",
         "delta.minReaderVersion",
@@ -27,6 +27,7 @@ class TblPropertiesConfig(DatabricksComponentConfig):
         "pipeline_internal.catalogType",
         "pipelines.metastore.tableName",
         "pipeline_internal.enzymeMode",
+        "clusterByAuto",
         "clusteringColumns",
         "delta.enableRowTracking",
         "delta.feature.appendOnly",
@@ -68,7 +69,8 @@ class TblPropertiesProcessor(DatabricksComponentProcessor[TblPropertiesConfig]):
             for row in table.rows:
                 if str(row[0]) == "pipelines.pipelineId":
                     pipeline_id = str(row[1])
-                tblproperties[str(row[0])] = str(row[1])
+                elif str(row[0]) not in TblPropertiesConfig.ignore_list:
+                    tblproperties[str(row[0])] = str(row[1])
 
         return TblPropertiesConfig(tblproperties=tblproperties, pipeline_id=pipeline_id)
 
diff --git a/dbt/include/databricks/macros/materializations/incremental/incremental.sql b/dbt/include/databricks/macros/materializations/incremental/incremental.sql
index 7c37efac..8005ed46 100644
--- a/dbt/include/databricks/macros/materializations/incremental/incremental.sql
+++ b/dbt/include/databricks/macros/materializations/incremental/incremental.sql
@@ -148,16 +148,19 @@
         Also, why does not either drop_relation or adapter.drop_relation work here?!
         --#}
       {%- endif -%}
-      {% do apply_liquid_clustered_cols(target_relation) %}
       {% if _configuration_changes is not none %}
         {% set tags = _configuration_changes.changes.get("tags", None) %}
         {% set tblproperties = _configuration_changes.changes.get("tblproperties", None) %}
+        {% set liquid_clustering = _configuration_changes.changes.get("liquid_clustered_by") %}
         {% if tags is not none %}
           {% do apply_tags(target_relation, tags.set_tags, tags.unset_tags) %}
         {%- endif -%}
         {% if tblproperties is not none %}
           {% do apply_tblproperties(target_relation, tblproperties.tblproperties) %}
         {%- endif -%}
+        {% if liquid_clustered_by is not none %}
+          
+        {% endif %}
       {%- endif -%}
       {% do persist_docs(target_relation, model, for_relation=True) %}
     {%- endif -%}
diff --git a/dbt/include/databricks/macros/relations/liquid_clustering.sql b/dbt/include/databricks/macros/relations/liquid_clustering.sql
index 43a3b113..8fca9a1e 100644
--- a/dbt/include/databricks/macros/relations/liquid_clustering.sql
+++ b/dbt/include/databricks/macros/relations/liquid_clustering.sql
@@ -11,19 +11,20 @@
   {%- endif %}
 {%- endmacro -%}
 
-{% macro apply_liquid_clustered_cols(target_relation) -%}
-  {%- set cols = config.get('liquid_clustered_by', validator=validation.any[list, basestring]) -%}
-  {%- set auto_cluster = config.get('auto_liquid_cluster', validator=validation.any[boolean]) -%}
-  {%- if cols is not none %}
-    {%- if cols is string -%}
-      {%- set cols = [cols] -%}
-    {%- endif -%}
+{% macro apply_liquid_clustered_cols(target_relation, liquid_clustering) -%}
+  {%- set cols = liquid_clustering.cluster_by -%}
+  {%- set auto_cluster = liquid_clustering.auto_cluster -%}
+  {%- if cols and cols != [] %}
     {%- call statement('set_cluster_by_columns') -%}
-        ALTER {{ target_relation.type }} {{ target_relation.render() }} CLUSTER BY ({{ cols | join(', ') }})
+      ALTER {{ target_relation.type }} {{ target_relation.render() }} CLUSTER BY ({{ cols | join(', ') }})
     {%- endcall -%}
   {%- elif auto_cluster -%}
     {%- call statement('set_cluster_by_auto') -%}
-        ALTER {{ target_relation.type }} {{ target_relation.render() }} CLUSTER BY AUTO
+      ALTER {{ target_relation.type }} {{ target_relation.render() }} CLUSTER BY AUTO
+    {%- endcall -%}
+  {% else %}
+    {%- call statement('unset_cluster_by') -%}
+      ALTER {{ target_relation.type }} {{ target_relation.render() }} CLUSTER BY NONE
     {%- endcall -%}
   {%- endif %}
 {%- endmacro -%}
\ No newline at end of file
diff --git a/dbt/include/databricks/macros/relations/table/alter.sql b/dbt/include/databricks/macros/relations/table/alter.sql
index bd9ebf4b..dd5f3dfa 100644
--- a/dbt/include/databricks/macros/relations/table/alter.sql
+++ b/dbt/include/databricks/macros/relations/table/alter.sql
@@ -1,15 +1,18 @@
 {% macro apply_config_changeset(target_relation, model, configuration_changes) %}
     {{ log("Applying configuration changes to relation " ~ target_relation) }}
-    {% do apply_liquid_clustered_cols(target_relation) %}
     {% if configuration_changes %}
       {% set tags = configuration_changes.changes.get("tags") %}
       {% set tblproperties = configuration_changes.changes.get("tblproperties") %}
+      {% set liquid_clustering = configuration_changes.changes.get("liquid_clustering")%}
       {% if tags is not none %}
         {% do apply_tags(target_relation, tags.set_tags, tags.unset_tags) %}
       {%- endif -%}
       {% if tblproperties is not none %}
         {% do apply_tblproperties(target_relation, tblproperties.tblproperties) %}
       {%- endif -%}
+      {% if liquid_clustering is not none %}
+        {% do apply_liquid_clustered_cols(target_relation, liquid_clustering) %}
+      {%- endif -%}
     {%- endif -%}
     {% do persist_docs(target_relation, model, for_relation=True) %}
 {% endmacro %}
\ No newline at end of file


Test patch:
diff --git a/tests/unit/relation_configs/test_incremental_config.py b/tests/unit/relation_configs/test_incremental_config.py
index d52f8ac3..39fee754 100644
--- a/tests/unit/relation_configs/test_incremental_config.py
+++ b/tests/unit/relation_configs/test_incremental_config.py
@@ -1,6 +1,7 @@
 from agate import Table
 
 from dbt.adapters.databricks.relation_configs.incremental import IncrementalTableConfig
+from dbt.adapters.databricks.relation_configs.liquid_clustering import LiquidClusteringConfig
 from dbt.adapters.databricks.relation_configs.tags import TagsConfig
 from dbt.adapters.databricks.relation_configs.tblproperties import TblPropertiesConfig
 
@@ -15,7 +16,14 @@ class TestIncrementalConfig:
                 ],
                 column_names=["tag_name", "tag_value"],
             ),
-            "show_tblproperties": Table(rows=[["prop", "f1"]], column_names=["key", "value"]),
+            "show_tblproperties": Table(
+                rows=[
+                    ["prop", "f1"],
+                    ["clusterByAuto", "true"],
+                    ["clusteringColumns", '[["col1"],[""a""]]'],
+                ],
+                column_names=["key", "value"],
+            ),
         }
 
         config = IncrementalTableConfig.from_results(results)
@@ -24,5 +32,9 @@ class TestIncrementalConfig:
             config={
                 "tags": TagsConfig(set_tags={"tag1": "value1", "tag2": "value2"}),
                 "tblproperties": TblPropertiesConfig(tblproperties={"prop": "f1"}),
+                "liquid_clustering": LiquidClusteringConfig(
+                    auto_cluster=True,
+                    cluster_by=["col1", '"a"'],
+                ),
             }
         )