Repository: databricks/dbt-databricks (python)

Issue:
Allow liquid clustering `CLUSTER BY AUTO` option
### Describe the feature
Currently the dbt-databricks liquid clustering feature relies on providing columns and then generating an alter statement that looks like below

Ideally if users don't define columns to cluster by, and a config `auto_liquid_cluster` is true then their relations should be clustered 



Pull request description:
Allow auto liquid clustering
<!-- Please review our pull request review process in CONTRIBUTING.md before your proceed. -->

<!---
  Include the number of the issue addressed by this PR above if applicable.

  Example:
    resolves #1234

  Please review our pull request review process in CONTRIBUTING.md before your proceed.
-->

Resolves #932


### Description

<!--- Describe the Pull Request here -->

Adds config `auto_liquid_cluster` to allow for tables to be `CLUSTER BY AUTO` on creation or update (i.e. incremental run)

### 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 74ec15a6..c1c2d60b 100644
--- a/CHANGELOG.md
+++ b/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))
diff --git a/dbt/adapters/databricks/impl.py b/dbt/adapters/databricks/impl.py
index e9b4fce4..24dd4d27 100644
--- a/dbt/adapters/databricks/impl.py
+++ b/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
diff --git a/dbt/include/databricks/macros/relations/liquid_clustering.sql b/dbt/include/databricks/macros/relations/liquid_clustering.sql
index b30269fd..43a3b113 100644
--- a/dbt/include/databricks/macros/relations/liquid_clustering.sql
+++ b/dbt/include/databricks/macros/relations/liquid_clustering.sql
@@ -1,15 +1,19 @@
 {% 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] -%}
@@ -17,5 +21,9 @@
     {%- 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 -%}
\ No newline at end of file
diff --git a/dbt/include/databricks/macros/relations/optimize.sql b/dbt/include/databricks/macros/relations/optimize.sql
index 79f00164..a6108709 100644
--- a/dbt/include/databricks/macros/relations/optimize.sql
+++ b/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 ...  #}


Test patch:
diff --git a/tests/functional/adapter/liquid_clustering/fixtures.py b/tests/functional/adapter/liquid_clustering/fixtures.py
index 951210ea..6c730222 100644
--- a/tests/functional/adapter/liquid_clustering/fixtures.py
+++ b/tests/functional/adapter/liquid_clustering/fixtures.py
@@ -2,3 +2,8 @@ liquid_cluster_sql = """
 {{ 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
+"""
diff --git a/tests/functional/adapter/liquid_clustering/test_liquid_clustering.py b/tests/functional/adapter/liquid_clustering/test_liquid_clustering.py
index a9cc0ee0..45c7dfe2 100644
--- a/tests/functional/adapter/liquid_clustering/test_liquid_clustering.py
+++ b/tests/functional/adapter/liquid_clustering/test_liquid_clustering.py
@@ -15,3 +15,16 @@ class TestLiquidClustering:
     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
diff --git a/tests/unit/macros/relations/test_table_macros.py b/tests/unit/macros/relations/test_table_macros.py
index 9c563513..f05dc9d4 100644
--- a/tests/unit/macros/relations/test_table_macros.py
+++ b/tests/unit/macros/relations/test_table_macros.py
@@ -180,6 +180,16 @@ class TestCreateTableAs(MacroTestBase):
 
         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"