Allow auto liquid clustering (#935)

Author: ShaneMazur

CHANGELOG.md

@@ -1,10 +1,15 @@
 ## dbt-databricks 1.9.5 (TBD)
 
+### Features
+
+- Add `auto_liquid_cluster` config to enable Auto Liquid Clustering for Delta-based dbt models ([935](https://github.com/databricks/dbt-databricks/pull/935))
+
 ### Fixes
 
 - table_format: iceberg is unblocked for snapshots ([930](https://github.com/databricks/dbt-databricks/pull/930))
 - Fix for regression in glue table listing behavior ([934](https://github.com/databricks/dbt-databricks/pull/934))
 
+
 ### Under the Hood
 
 - Collapsing to a single connection manager (since the old one no longer works) ([910](https://github.com/databricks/dbt-databricks/pull/910))

dbt/adapters/databricks/impl.py

@@ -110,6 +110,7 @@ class DatabricksConfig(AdapterConfig):
     partition_by: Optional[Union[list[str], str]] = None
     clustered_by: Optional[Union[list[str], str]] = None
     liquid_clustered_by: Optional[Union[list[str], str]] = None
+    auto_liquid_cluster: Optional[bool] = None
     buckets: Optional[int] = None
     options: Optional[dict[str, str]] = None
     merge_update_columns: Optional[str] = None

dbt/include/databricks/macros/relations/liquid_clustering.sql

@@ -1,21 +1,29 @@
 {% macro liquid_clustered_cols() -%}
   {%- 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 -%}
     CLUSTER BY ({{ cols | join(', ') }})
+    {%- elif auto_cluster -%}
+    CLUSTER BY AUTO
   {%- 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 -%}
     {%- call statement('set_cluster_by_columns') -%}
         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
+    {%- endcall -%}
   {%- endif %}
 {%- endmacro -%}

dbt/include/databricks/macros/relations/optimize.sql

@@ -6,7 +6,7 @@
   {%- if var('DATABRICKS_SKIP_OPTIMIZE', 'false')|lower != 'true' and
         var('databricks_skip_optimize', 'false')|lower != 'true' and
         config.get('file_format', 'delta') == 'delta' -%}
-    {%- if (config.get('zorder', False) or config.get('liquid_clustered_by', False)) -%}
+    {%- if (config.get('zorder', False) or config.get('liquid_clustered_by', False)) or config.get('auto_liquid_cluster', False) -%}
       {%- call statement('run_optimize_stmt') -%}
         {{ get_optimize_sql(relation) }}
       {%- endcall -%}
@@ -17,8 +17,8 @@
 {%- macro get_optimize_sql(relation) %}
   optimize {{ relation }}
   {%- if config.get('zorder', False) and config.get('file_format', 'delta') == 'delta' %}
-    {%- if config.get('liquid_clustered_by', False) %}
-      {{ exceptions.warn("Both zorder and liquid_clustered_by are set but they are incompatible. zorder will be ignored.") }}
+    {%- if config.get('liquid_clustered_by', False) or config.get('auto_liquid_cluster', False) %}
+      {{ exceptions.warn("Both zorder and liquid_clustering are set but they are incompatible. zorder will be ignored.") }}
     {%- else %}
       {%- set zorder = config.get('zorder', none) %}
       {# TODO: predicates here? WHERE ...  #}

tests/functional/adapter/liquid_clustering/fixtures.py

@@ -2,3 +2,8 @@
 {{ config(materialized='incremental', liquid_clustered_by='id') }}
 select 1 as id, 'Joe' as name
 """
+
+auto_liquid_cluster_sql = """
+{{ config(materialized='incremental', auto_liquid_cluster=true) }}
+select 1 as id, 'Joe' as name
+"""

tests/functional/adapter/liquid_clustering/test_liquid_clustering.py

@@ -15,3 +15,16 @@ def models(self):
     def test_liquid_clustering(self, project):
         _, logs = util.run_dbt_and_capture(["--debug", "run"])
         assert "optimize" in logs
+
+
+class TestAutoLiquidClustering:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "liquid_clustering.sql": fixtures.liquid_cluster_sql,
+        }
+
+    @pytest.mark.skip_profile("databricks_uc_cluster", "databricks_cluster")
+    def test_liquid_clustering(self, project):
+        _, logs = util.run_dbt_and_capture(["--debug", "run"])
+        assert "optimize" in logs

tests/unit/macros/relations/test_table_macros.py

@@ -180,6 +180,16 @@ def test_macros_create_table_as_liquid_clusters(self, config, template_bundle):
 
         assert sql == expected
 
+    def test_macros_create_table_as_liquid_cluster_auto(self, config, template_bundle):
+        config["auto_liquid_cluster"] = True
+        sql = self.render_create_table_as(template_bundle)
+        expected = (
+            f"create or replace table {template_bundle.relation} using"
+            " delta CLUSTER BY AUTO as select 1"
+        )
+
+        assert sql == expected
+
     def test_macros_create_table_as_comment(self, config, template_bundle):
         config["persist_docs"] = {"relation": True}
         template_bundle.context["model"].description = "Description Test"