diff --git a/plugins/snowflake/dbt/adapters/snowflake/impl.py b/plugins/snowflake/dbt/adapters/snowflake/impl.py
index c6df92f2ee0..5df44254b1e 100644
--- a/plugins/snowflake/dbt/adapters/snowflake/impl.py
+++ b/plugins/snowflake/dbt/adapters/snowflake/impl.py
@@ -10,11 +10,13 @@ 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):
@@ -23,20 +25,21 @@ def _catalog_filter_table(cls, table, manifest):
         lowered = table.rename(
             column_names=[c.lower() for c in table.column_names]
         )
-        return super(SnowflakeAdapter, cls)._catalog_filter_table(lowered,
-                                                                  manifest)
+        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}
+        )
diff --git a/plugins/snowflake/dbt/include/snowflake/macros/adapters.sql b/plugins/snowflake/dbt/include/snowflake/macros/adapters.sql
index 4bd3333a7c8..a344993625d 100644
--- a/plugins/snowflake/dbt/include/snowflake/macros/adapters.sql
+++ b/plugins/snowflake/dbt/include/snowflake/macros/adapters.sql
@@ -1,14 +1,33 @@
 {% 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) -%}
+  {%- 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) -%}
diff --git a/plugins/snowflake/dbt/include/snowflake/macros/materializations/incremental.sql b/plugins/snowflake/dbt/include/snowflake/macros/materializations/incremental.sql
index 238fad995f2..f7a26363fa0 100644
--- a/plugins/snowflake/dbt/include/snowflake/macros/materializations/incremental.sql
+++ b/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 -%}