Implement Snowflake clustering

plugins/snowflake/dbt/adapters/snowflake/impl.py

@@ -10,33 +10,30 @@ class SnowflakeAdapter(SQLAdapter):
     Relation = SnowflakeRelation
     ConnectionManager = SnowflakeConnectionManager
 
-    AdapterSpecificConfigs = frozenset({"transient"})
+    AdapterSpecificConfigs = frozenset({"transient", "cluster_by", "automatic_clustering"})
 
     @classmethod
     def date_function(cls):
-        return 'CURRENT_TIMESTAMP()'
+        return "CURRENT_TIMESTAMP()"
 
     @classmethod
     def _catalog_filter_table(cls, table, manifest):
         # On snowflake, users can set QUOTED_IDENTIFIERS_IGNORE_CASE, so force
         # the column names to their lowercased forms.
-        lowered = table.rename(
-            column_names=[c.lower() for c in table.column_names]
-        )
-        return super(SnowflakeAdapter, cls)._catalog_filter_table(lowered,
-                                                                  manifest)
+        lowered = table.rename(column_names=[c.lower() for c in table.column_names])
+        return super(SnowflakeAdapter, cls)._catalog_filter_table(lowered, manifest)
 
     def _make_match_kwargs(self, database, schema, identifier):
         quoting = self.config.quoting
-        if identifier is not None and quoting['identifier'] is False:
+        if identifier is not None and quoting["identifier"] is False:
             identifier = identifier.upper()
 
-        if schema is not None and quoting['schema'] is False:
+        if schema is not None and quoting["schema"] is False:
             schema = schema.upper()
 
-        if database is not None and quoting['database'] is False:
+        if database is not None and quoting["database"] is False:
             database = database.upper()
 
-        return filter_null_values({'identifier': identifier,
-                                   'schema': schema,
-                                   'database': database})
+        return filter_null_values(
+            {"identifier": identifier, "schema": schema, "database": database}
+        )

plugins/snowflake/dbt/include/snowflake/macros/adapters.sql

@@ -1,14 +1,37 @@
 {% macro snowflake__create_table_as(temporary, relation, sql) -%}
   {%- set transient = config.get('transient', default=true) -%}
+  {%- set cluster_by_keys = config.get('cluster_by', default=none) -%}
+  {%- set enable_automatic_clustering = config.get('automatic_clustering', default=false) -%}
+  {%- set is_incremental = config.get('materialized') == 'incremental' -%}
+  {%- if cluster_by_keys is not none and cluster_by_keys is string -%}
+    {%-  set cluster_by_keys = [cluster_by_keys] -%}
+    {%- set cluster_by_string = cluster_by_keys|join(", ")-%}
+  {%- endif -%}
+
+      create or replace {% if temporary -%}
+        temporary
+      {%- elif transient -%}
+        transient
+      {%- endif %} table {{ relation }}
+      as (
+        {%- if cluster_by_keys is not none -%}
+          select * from(
+            {{ sql }}
+            ) order by ({{ cluster_by_string }})
+        {%- else -%}
+          {{ sql }}
+        {%- endif %}
+      );
+    {% if cluster_by_keys is not none -%}
+      alter table {{relation}} cluster by ({{cluster_by_string}});
+    {%- endif -%}
+    {% if enable_automatic_clustering  -%}
+      alter table {{relation}} resume recluster;
+    {%- endif -%}
+
+
+
 
-  create or replace {% if temporary -%}
-    temporary
-  {%- elif transient -%}
-    transient
-  {%- endif %} table {{ relation }}
-  as (
-    {{ sql }}
-  );
 {% endmacro %}
 
 {% macro snowflake__create_view_as(relation, sql) -%}

plugins/snowflake/dbt/include/snowflake/macros/materializations/incremental.sql

@@ -42,6 +42,7 @@
 
     {%- call statement('main') -%}
       {{ create_table_as(false, target_relation, sql) }}
+
     {%- endcall -%}
 
   {%- else -%}