Repository: dbt-athena/dbt-athena (python)

Issue:
Support for the faster incremental materialization with Iceberg
Thanks for implementing the faster [Iceberg table materialization](https://github.com/dbt-athena/dbt-athena/issues/11). Can you please implement the same feature for the incremental materialization as well? 

- we first use parquet with limit 0
- do the schema evolution
- use the merge from the original SQL

Please see the [issue 11](https://github.com/dbt-athena/dbt-athena/issues/11) for context and more details. 

Pull request:
feat!: Iceberg features only compatible with v3 engine
Closes #72

### Description
We now support CTAS for Iceberg tables. 
- Iceberg tables can only be built using Athena Engine v3
- Hive tables can be built using Athena Engine v2 or v3

NB: Right now, we provide a destructive solution because the RENAME command for Iceberg tables seems broken (error saying Could not find results). We are waiting for a fix before addressing this.

## Models used to test

```sql
{{ config(
  materialized='table',
  partitioned_by=['hour(ts)'],
  format='iceberg',
) }}

SELECT cast(t.ts AS timestamp(6)) AS ts
FROM
    unnest(
        sequence(
            current_date - INTERVAL '1' DAY,
            cast(REPLACE(cast(current_timestamp as varchar), ' UTC', '') as timestamp(6)),
            INTERVAL '1' MINUTE
        )
    ) AS t(ts)
```

```sql
{{ config(
  materialized='incremental',
  incremental_strategy='append',
  partitioned_by=['hour(ts)'],
  format='iceberg'
) }}

SELECT
    ts
    , 'abc' AS str
FROM {{ ref('iceberg_data_source') }}
{% if is_incremental() %}
  -- this filter will only be applied on an incremental run
  WHERE ts > (SELECT max(ts) FROM {{ this }})
{% endif %}
```

```sql
{{ config(
  materialized='incremental',
  incremental_strategy='merge',
  partitioned_by=['hour(ts)'],
  unique_key='ts',
  format='iceberg'
) }}

SELECT
    ts
    , 'abc' AS str
FROM {{ ref('iceberg_data_source') }}
{% if is_incremental() %}
  -- this filter will only be applied on an incremental run
  WHERE ts > (SELECT max(ts) FROM {{ this }})
{% endif %}
```

```sql
{{ config(
  materialized='table',
  format='parquet'
) }}

SELECT cast(t.ts AS timestamp(3)) AS ts
FROM
    unnest(
        sequence(
            current_date - INTERVAL '1' DAY,
            cast(REPLACE(cast(current_timestamp as varchar), ' UTC', '') as timestamp(3)),
            INTERVAL '1' MINUTE
        )
    ) AS t(ts)
```

```sql
{{ config(
  materialized='incremental',
  incremental_strategy='append',
  format='parquet'
) }}

SELECT
    ts
    , 'abc' AS str
FROM {{ ref('parquet_data_source') }}
{% if is_incremental() %}
  -- this filter will only be applied on an incremental run
  WHERE ts > (SELECT max(ts) FROM {{ this }})
{% endif %}
```

```sql
{{ config(
  materialized='incremental',
  incremental_strategy='insert_overwrite',
  partitioned_by=['rnd'],
  format='parquet'
) }}

SELECT
    ts
    , 'abc' AS str
    , cast(random() * 5 AS int) AS rnd
FROM {{ ref('parquet_data_source') }}
{% if is_incremental() %}
  -- this filter will only be applied on an incremental run
  WHERE ts > (SELECT max(ts) FROM {{ this }})
{% endif %}
```

## Checklist
- [X] You followed [contributing section](https://github.com/dbt-athena/dbt-athena#contributing)
- [X] You added unit testing when necessary
- [X] You added functional testing when necessary


Patch:
diff --git a/README.md b/README.md
index 769124d..313453d 100644
--- a/README.md
+++ b/README.md
@@ -8,9 +8,13 @@
 * Supports dbt version `1.3.*`
 * Supports [Seeds][seeds]
 * Correctly detects views and their columns
-* Support [incremental models][incremental]
+* Supports [table materialization][table]
+  * [Iceberg tables][athena-iceberg] is supported **only with Athena Engine v3** and **a unique table location**
+  (see table location section below)
+  * Hive tables is supported by both Athena engines.
+* Supports [incremental models][incremental]
   * On iceberg tables :
-    * Support the use of `unique_key` only with query engine `v3` using the `merge` strategy
+    * Support the use of `unique_key` only with the `merge` strategy
     * Support the `append` strategy
   * On Hive tables :
     * Support two incremental update strategies: `insert_overwrite` and `append`
@@ -18,8 +22,10 @@
 * Does not support [Python models][python-models]
 
 [seeds]: https://docs.getdbt.com/docs/building-a-dbt-project/seeds
-[incremental]: https://docs.getdbt.com/docs/building-a-dbt-project/building-models/configuring-incremental-models
+[incremental]: https://docs.getdbt.com/docs/build/incremental-models
+[table]: https://docs.getdbt.com/docs/build/materializations#table
 [python-models]: https://docs.getdbt.com/docs/build/python-models#configuring-python-models
+[athena-iceberg]: https://docs.aws.amazon.com/athena/latest/ug/querying-iceberg.html
 
 ### Installation
 
@@ -92,7 +98,7 @@ _Additional information_
 #### Table Configuration
 
 * `external_location` (`default=none`)
-  * If set, the full S3 path in which the table will be saved.
+  * If set, the full S3 path in which the table will be saved. (Does not work with Iceberg table).
 * `partitioned_by` (`default=none`)
   * An array list of columns by which the table will be partitioned
   * Limited to creation of 100 partitions (_currently_)
@@ -139,7 +145,7 @@ partitions are defined, dbt will fall back to the `append` strategy.
 * `append`: Insert new records without updating, deleting or overwriting any existing data. There might be duplicate
 data (e.g. great for log or historical data).
 * `merge`: Conditionally updates, deletes, or inserts rows into an Iceberg table. Used in combination with `unique_key`.
-Only available when using Iceberg and Athena engine version 3.
+Only available when using Iceberg.
 
 #### On schema change
 
@@ -161,7 +167,7 @@ To get started just add this as your model:
 {{ config(
     materialized='table',
     format='iceberg',
-    partitioned_by=['bucket(5, user_id)'],
+    partitioned_by=['bucket(user_id, 5)'],
     table_properties={
     	'optimize_rewrite_delete_file_threshold': '2'
     	}
diff --git a/dbt/adapters/athena/impl.py b/dbt/adapters/athena/impl.py
index 1189d18..0b780ba 100755
--- a/dbt/adapters/athena/impl.py
+++ b/dbt/adapters/athena/impl.py
@@ -63,21 +63,28 @@ class AthenaAdapter(SQLAdapter):
 
     @available
     def s3_table_location(
-        self, s3_data_dir: Optional[str], s3_data_naming: str, schema_name: str, table_name: str
+        self,
+        s3_data_dir: Optional[str],
+        s3_data_naming: str,
+        schema_name: str,
+        table_name: str,
+        external_location: Optional[str] = None,
+        is_temporary_table: bool = False,
     ) -> str:
         """
         Returns either a UUID or database/table prefix for storing a table,
         depending on the value of s3_table
         """
+        if external_location and not is_temporary_table:
+            return external_location.rstrip("/")
+
         mapping = {
-            "uuid": path.join(self.s3_table_prefix(s3_data_dir), str(uuid4())) + "/",
-            "table": path.join(self.s3_table_prefix(s3_data_dir), table_name) + "/",
-            "table_unique": path.join(self.s3_table_prefix(s3_data_dir), table_name, str(uuid4())) + "/",
-            "schema_table": path.join(self.s3_table_prefix(s3_data_dir), schema_name, table_name) + "/",
-            "schema_table_unique": path.join(self.s3_table_prefix(s3_data_dir), schema_name, table_name, str(uuid4()))
-            + "/",
+            "uuid": path.join(self.s3_table_prefix(s3_data_dir), str(uuid4())),
+            "table": path.join(self.s3_table_prefix(s3_data_dir), table_name),
+            "table_unique": path.join(self.s3_table_prefix(s3_data_dir), table_name, str(uuid4())),
+            "schema_table": path.join(self.s3_table_prefix(s3_data_dir), schema_name, table_name),
+            "schema_table_unique": path.join(self.s3_table_prefix(s3_data_dir), schema_name, table_name, str(uuid4())),
         }
-
         table_location = mapping.get(s3_data_naming)
 
         if table_location is None:
diff --git a/dbt/adapters/athena/query_headers.py b/dbt/adapters/athena/query_headers.py
index bb7600d..c80570c 100644
--- a/dbt/adapters/athena/query_headers.py
+++ b/dbt/adapters/athena/query_headers.py
@@ -14,7 +14,7 @@ class _QueryComment(dbt.adapters.base.query_headers._QueryComment):
         if not self.query_comment:
             return sql
 
-        # alter statements or vacuum statements seems not to support properly query comments
+        # alter or vacuum statements don't seem to support properly query comments
         # let's just exclude them
         if any(map(sql.lower().__contains__, ["alter", "vacuum"])):
             return sql
diff --git a/dbt/include/athena/macros/adapters/relation.sql b/dbt/include/athena/macros/adapters/relation.sql
index 0bf4f86..96e3be1 100644
--- a/dbt/include/athena/macros/adapters/relation.sql
+++ b/dbt/include/athena/macros/adapters/relation.sql
@@ -1,8 +1,15 @@
-{% macro athena__drop_relation(relation) -%}
-  {% if config.get('incremental_strategy') != 'append' %}
+{% macro drop_relation(relation) -%}
+  {% if config.get('format') != 'iceberg' and config.get('incremental_strategy') != 'append' %}
     {%- do adapter.clean_up_table(relation.schema, relation.table) -%}
   {% endif %}
   {% call statement('drop_relation', auto_begin=False) -%}
     drop {{ relation.type }} if exists {{ relation }}
   {%- endcall %}
 {% endmacro %}
+
+{% macro set_table_classification(relation) -%}
+  {%- set format = config.get('format') -%}
+  {% call statement('set_table_classification', auto_begin=False) -%}
+    alter table {{ relation }} set tblproperties ('classification' = '{{ format }}')
+  {%- endcall %}
+{%- endmacro %}
diff --git a/dbt/include/athena/macros/materializations/models/helpers.sql b/dbt/include/athena/macros/materializations/models/helpers.sql
deleted file mode 100644
index 74148a9..0000000
--- a/dbt/include/athena/macros/materializations/models/helpers.sql
+++ /dev/null
@@ -1,7 +0,0 @@
-{% macro set_table_classification(relation, default_value) -%}
-    {%- set format = config.get('format', default=default_value) -%}
-
-    {% call statement('set_table_classification', auto_begin=False) -%}
-        alter table {{ relation }} set tblproperties ('classification' = '{{ format }}')
-    {%- endcall %}
-{%- endmacro %}
diff --git a/dbt/include/athena/macros/materializations/models/helpers_iceberg.sql b/dbt/include/athena/macros/materializations/models/helpers_iceberg.sql
deleted file mode 100644
index 26a78f9..0000000
--- a/dbt/include/athena/macros/materializations/models/helpers_iceberg.sql
+++ /dev/null
@@ -1,104 +0,0 @@
-{% macro drop_iceberg(relation) -%}
-  drop table if exists {{ relation }}
-{% endmacro %}
-
-{% macro create_table_iceberg(relation, old_relation, tmp_relation, sql) -%}
-  {%- set external_location = config.get('external_location', default=none) -%}
-  {%- set staging_location = config.get('staging_location') -%}
-  {%- set partitioned_by = config.get('partitioned_by', default=none) -%}
-  {%- set write_compression = config.get('write_compression', default=none) -%}
-
-  {%- set target_relation = this.incorporate(type='table') -%}
-
-  -- clean residual tmp table
-  {% if tmp_relation is not none %}
-     {% do adapter.drop_relation(tmp_relation) %}
-  {% endif %}
-
-  -- create tmp table
-  {% do run_query(create_tmp_table_iceberg(tmp_relation, sql, staging_location)) %}
-
-  -- get columns from tmp table to retrieve metadata
-  {%- set dest_columns = adapter.get_columns_in_relation(tmp_relation) -%}
-
-  -- drop old relation after tmp table is ready
-  {%- if old_relation is not none -%}
-  	{% do run_query(drop_iceberg(old_relation)) %}
-  {%- endif -%}
-
-  -- create iceberg table
-  {% do run_query(create_iceberg_table_definition(target_relation, dest_columns)) %}
-
-  -- return final insert statement
-  {{ return(insert_from_sql(target_relation, sql)) }}
-
-{% endmacro %}
-
-{% macro create_tmp_table_iceberg(relation, sql, staging_location, with_limit=true) -%}
-  create table
-    {{ relation }}
-    with (
-      write_compression='snappy',
-      format='parquet'
-    )
-  as
-    select * from (
-        {{ sql }}
-    )
-    {%- if with_limit -%}
-      limit 0
-    {%- endif -%}
-{% endmacro %}
-
-{% macro create_iceberg_table_definition(relation, dest_columns) -%}
-  {%- set external_location = config.get('external_location', default=none) -%}
-  {%- set partitioned_by = config.get('partitioned_by', default=none) -%}
-  {%- set table_properties = config.get('table_properties', default={}) -%}
-  {%- set _ = table_properties.update({'table_type': 'ICEBERG'}) -%}
-  {%- set table_properties_formatted = [] -%}
-  {%- set dest_columns_with_type = [] -%}
-  {%- set s3_data_dir = config.get('s3_data_dir', default=target.s3_data_dir) -%}
-  {%- set s3_data_naming = config.get('s3_data_naming', default=target.s3_data_naming) -%}
-
-  {%- for k in table_properties -%}
-  	{% set _ = table_properties_formatted.append("'" + k + "'='" + table_properties[k] + "'") -%}
-  {%- endfor -%}
-
-  {%- set table_properties_csv= table_properties_formatted | join(', ') -%}
-
-  {%- if external_location is none %}
-    {%- set  external_location = adapter.s3_table_location(s3_data_dir, s3_data_naming, relation.schema, relation.identifier) -%}
-  {%- endif %}
-
-  {%- for col in dest_columns -%}
-	{% set dtype = ddl_data_type(col.dtype) -%}
-  	{% set _ = dest_columns_with_type.append(col.name + ' ' + dtype) -%}
-  {%- endfor -%}
-
-  {%- set dest_columns_with_type_csv = dest_columns_with_type | join(', ') -%}
-
-  CREATE TABLE {{ relation }} (
-    {{ dest_columns_with_type_csv }}
-  )
-  {%- if partitioned_by is not none %}
-    {%- set partitioned_by_csv = partitioned_by | join(', ') -%}
-  	PARTITIONED BY ({{partitioned_by_csv}})
-  {%- endif %}
-  LOCATION '{{ external_location }}'
-  TBLPROPERTIES (
-  	{{table_properties_csv}}
-  )
-{% endmacro %}
-
-
-{% macro insert_from_sql(target_relation, sql, statement_name="main") %}
-  {%- set dest_columns = adapter.get_columns_in_relation(target_relation) -%}
-  {%- set dest_cols_csv = dest_columns | map(attribute='quoted') | join(', ') -%}
-
-  insert into {{ target_relation }} ({{ dest_cols_csv }}) (
-    select {{ dest_cols_csv }}
-    from (
-      {{sql}}
-      )
-    );
-{%- endmacro %}
diff --git a/dbt/include/athena/macros/materializations/models/incremental/incremental.sql b/dbt/include/athena/macros/materializations/models/incremental/incremental.sql
index 8983c82..8c36bd2 100644
--- a/dbt/include/athena/macros/materializations/models/incremental/incremental.sql
+++ b/dbt/include/athena/macros/materializations/models/incremental/incremental.sql
@@ -6,7 +6,6 @@
   {% set on_schema_change = incremental_validate_on_schema_change(config.get('on_schema_change'), default='ignore') %}
 
   {% set partitioned_by = config.get('partitioned_by', default=none) %}
-  {% set staging_location = config.get('staging_location') %}
   {% set target_relation = this.incorporate(type='table') %}
   {% set existing_relation = load_relation(this) %}
   {% set tmp_relation = make_temp_relation(this) %}
@@ -21,56 +20,30 @@
   -- `BEGIN` happens here:
   {{ run_hooks(pre_hooks, inside_transaction=True) }}
 
-  -- Having Iceberg in v2 engine not able to do CTAS makes it complex. Hard to factorize with table materialization
-  -- because of statement block.
-  {%- if format == 'iceberg' -%}
-    {%- set iceberg_tmp_relation = make_temp_relation(target_relation, '__incremental_iceberg_ctas') -%}
-  {%- endif -%}
-
   {% set to_drop = [] %}
   {% if existing_relation is none %}
-    {%- if format == 'iceberg' -%}
-      {%- set build_sql = create_table_iceberg(target_relation, existing_relation, iceberg_tmp_relation, sql) -%}
-      {% do to_drop.append(iceberg_tmp_relation) %}
-    {% else %}
-      {% set build_sql = create_table_as(False, target_relation, sql) -%}
-    {%- endif -%}
+    {% set build_sql = create_table_as(False, target_relation, sql) -%}
   {% elif existing_relation.is_view or should_full_refresh() %}
-    {% do adapter.drop_relation(existing_relation) %}
-    {%- if format == 'iceberg' -%}
-      {%- set build_sql = create_table_iceberg(target_relation, existing_relation, iceberg_tmp_relation, sql) -%}
-      {% do to_drop.append(iceberg_tmp_relation) %}
-    {% else %}
-      {% set build_sql = create_table_as(False, target_relation, sql) -%}
-    {%- endif -%}
+    {% do drop_relation(existing_relation) %}
+    {% set build_sql = create_table_as(False, target_relation, sql) -%}
   {% elif partitioned_by is not none and strategy == 'insert_overwrite' %}
-      {% set tmp_relation = make_temp_relation(target_relation) %}
-      {% if tmp_relation is not none %}
-        {% do adapter.drop_relation(tmp_relation) %}
-      {% endif %}
-      {% do run_query(create_table_as(True, tmp_relation, sql)) %}
-      {% do delete_overlapping_partitions(target_relation, tmp_relation, partitioned_by) %}
-      {% set build_sql = incremental_insert(on_schema_change, tmp_relation, target_relation, existing_relation) %}
-      {% do to_drop.append(tmp_relation) %}
+    {% set tmp_relation = make_temp_relation(target_relation) %}
+    {% if tmp_relation is not none %}
+      {% do drop_relation(tmp_relation) %}
+    {% endif %}
+    {% do run_query(create_table_as(True, tmp_relation, sql)) %}
+    {% do delete_overlapping_partitions(target_relation, tmp_relation, partitioned_by) %}
+    {% set build_sql = incremental_insert(on_schema_change, tmp_relation, target_relation, existing_relation) %}
+    {% do to_drop.append(tmp_relation) %}
   {% elif strategy == 'append' %}
-      {% set tmp_relation = make_temp_relation(target_relation) %}
-      {% if tmp_relation is not none %}
-        {% do adapter.drop_relation(tmp_relation) %}
-      {% endif %}
-      {%- if format == 'iceberg' -%}
-         -- We need to use the create_tmp_table_iceberg macro (parquet table) because create_table_as create a table
-         -- with the format inherited from config. For Iceberg, what we are doing is that we create a parquet table
-         -- containing the result of the incremental query (no limit), and then we insert this in the iceberg table.
-        {% do run_query(create_tmp_table_iceberg(tmp_relation, sql, staging_location, false)) %}
-      {% else %}
-        {% do run_query(create_table_as(True, tmp_relation, sql)) %}
-      {%- endif -%}
-      {% set build_sql = incremental_insert(on_schema_change, tmp_relation, target_relation, existing_relation) %}
-      {% do to_drop.append(tmp_relation) %}
+    {% set tmp_relation = make_temp_relation(target_relation) %}
+    {% if tmp_relation is not none %}
+      {% do drop_relation(tmp_relation) %}
+    {% endif %}
+    {% do run_query(create_table_as(True, tmp_relation, sql)) %}
+    {% set build_sql = incremental_insert(on_schema_change, tmp_relation, target_relation, existing_relation) %}
+    {% do to_drop.append(tmp_relation) %}
   {% elif strategy == 'merge' and format == 'iceberg' %}
-    -- Merge is only supported for Iceberg in v3 engine. In this scenario, we are going to start the same as the append
-    -- strategy, meaning creating a temp table with the incremental query and then, instead of inserting, we are
-    -- merging
     {% set unique_key = config.get('unique_key') %}
     {% set empty_unique_key -%}
       Merge strategy must implement unique_key as a single column or a list of columns.
@@ -81,9 +54,9 @@
 
     {% set tmp_relation = make_temp_relation(target_relation) %}
     {% if tmp_relation is not none %}
-      {% do adapter.drop_relation(tmp_relation) %}
+      {% do drop_relation(tmp_relation) %}
     {% endif %}
-    {% do run_query(create_tmp_table_iceberg(tmp_relation, sql, staging_location, false)) %}
+    {% do run_query(create_table_as(True, tmp_relation, sql)) %}
     {% set build_sql = iceberg_merge(on_schema_change, tmp_relation, target_relation, unique_key, existing_relation) %}
     {% do to_drop.append(tmp_relation) %}
   {% endif %}
@@ -93,8 +66,8 @@
   {% endcall %}
 
   -- set table properties
-  {% if not to_drop %}
-    {{ set_table_classification(target_relation, 'parquet') }}
+  {% if not to_drop and format != 'iceberg' %}
+    {{ set_table_classification(target_relation) }}
   {% endif %}
 
   {% do persist_docs(target_relation, model) %}
@@ -105,7 +78,7 @@
   {% do adapter.commit() %}
 
   {% for rel in to_drop %}
-    {% do adapter.drop_relation(rel) %}
+    {% do drop_relation(rel) %}
   {% endfor %}
 
   {{ run_hooks(post_hooks, inside_transaction=False) }}
diff --git a/dbt/include/athena/macros/materializations/models/incremental/on_schema_change.sql b/dbt/include/athena/macros/materializations/models/incremental/on_schema_change.sql
index 7d17f2d..7dd8010 100644
--- a/dbt/include/athena/macros/materializations/models/incremental/on_schema_change.sql
+++ b/dbt/include/athena/macros/materializations/models/incremental/on_schema_change.sql
@@ -32,8 +32,6 @@
         Columns added: {{ add_to_target_arr }}
         Columns removed: {{ remove_from_target_arr }}
         Data types changed: {{ new_target_types }}
-        {%- if format == 'iceberg' -%}(Data types can be irrelevant since parquet and iceberg tables might be compared
-        and they can differ in types.){%- endif -%}
   {% endset %}
   {% do log(schema_change_message) %}
 {% endmacro %}
diff --git a/dbt/include/athena/macros/materializations/models/table/create_table_as.sql b/dbt/include/athena/macros/materializations/models/table/create_table_as.sql
index 82cf391..03efe05 100644
--- a/dbt/include/athena/macros/materializations/models/table/create_table_as.sql
+++ b/dbt/include/athena/macros/materializations/models/table/create_table_as.sql
@@ -8,33 +8,61 @@
   {%- set write_compression = config.get('write_compression', default=none) -%}
   {%- set s3_data_dir = config.get('s3_data_dir', default=target.s3_data_dir) -%}
   {%- set s3_data_naming = config.get('s3_data_naming', default=target.s3_data_naming) -%}
+  {%- set extra_table_properties = config.get('table_properties', default=none) -%}
+
+  {%- set location_property = 'external_location' -%}
+  {%- set partition_property = 'partitioned_by' -%}
+
+  {%- if format == 'iceberg' -%}
+    {%- set location_property = 'location' -%}
+    {%- set partition_property = 'partitioning' -%}
+    {%- if bucketed_by is not none or bucket_count is not none -%}
+      {%- set ignored_bucket_iceberg -%}
+      bucketed_by or bucket_count cannot be used with Iceberg tables. You have to use the bucket function
+      when partitioning. Will be ignored
+      {%- endset -%}
+      {%- set bucketed_by = none -%}
+      {%- set bucket_count = none -%}
+      {% do log(ignored_bucket_iceberg) %}
+    {%- endif -%}
+    {%- if s3_data_naming in ['table', 'table_schema'] or external_location is not none -%}
+      {%- set error_unique_location_iceberg -%}
+        You need to have an unique table location when creating Iceberg table. Right now we are building tables in
+        a destructive way but in the near future we will be using the RENAME feature to provide near-zero downtime.
+      {%- endset -%}
+      {% do exceptions.raise_compiler_error(error_unique_location_iceberg) %}
+    {%- endif -%}
+  {%- endif %}
 
   create table
     {{ relation }}
-
-    with (
-      {%- if external_location is not none and not temporary %}
-        external_location='{{ external_location }}',
-      {%- else -%}
-        external_location='{{ adapter.s3_table_location(s3_data_dir, s3_data_naming, relation.schema, relation.identifier) }}',
-      {%- endif %}
-      {%- if partitioned_by is not none %}
-        partitioned_by=ARRAY{{ partitioned_by | tojson | replace('\"', '\'') }},
-      {%- endif %}
-      {%- if bucketed_by is not none %}
-        bucketed_by=ARRAY{{ bucketed_by | tojson | replace('\"', '\'') }},
-      {%- endif %}
-      {%- if bucket_count is not none %}
-        bucket_count={{ bucket_count }},
-      {%- endif %}
-      {%- if field_delimiter is not none %}
-        field_delimiter='{{ field_delimiter }}',
-      {%- endif %}
-      {%- if write_compression is not none %}
-        write_compression='{{ write_compression }}',
-      {%- endif %}
-        format='{{ format }}'
-    )
+  with (
+    table_type={%- if format == 'iceberg' -%}'iceberg'{%- else -%}'hive'{%- endif %},
+    is_external={%- if format == 'iceberg' -%}false{%- else -%}true{%- endif %},
+    {{ location_property }}='{{ adapter.s3_table_location(s3_data_dir, s3_data_naming, relation.schema, relation.identifier, external_location, temporary) }}',
+  {%- if partitioned_by is not none %}
+    {{ partition_property }}=ARRAY{{ partitioned_by | tojson | replace('\"', '\'') }},
+  {%- endif %}
+  {%- if bucketed_by is not none %}
+    bucketed_by=ARRAY{{ bucketed_by | tojson | replace('\"', '\'') }},
+  {%- endif %}
+  {%- if bucket_count is not none %}
+    bucket_count={{ bucket_count }},
+  {%- endif %}
+  {%- if field_delimiter is not none %}
+    field_delimiter='{{ field_delimiter }}',
+  {%- endif %}
+  {%- if write_compression is not none %}
+    write_compression='{{ write_compression }}',
+  {%- endif %}
+    format={%- if format == 'iceberg' -%}'parquet'{%- else -%}'{{ format }}'{%- endif %}
+  {%- if extra_table_properties is not none -%}
+    {%- for prop_name, prop_value in extra_table_properties.items() -%}
+    ,
+    {{ prop_name }}={{ prop_value }}
+    {%- endfor -%}
+  {% endif %}
+  )
   as
     {{ sql }}
 {% endmacro %}
diff --git a/dbt/include/athena/macros/materializations/models/table/table.sql b/dbt/include/athena/macros/materializations/models/table/table.sql
index 5b98583..65ba004 100644
--- a/dbt/include/athena/macros/materializations/models/table/table.sql
+++ b/dbt/include/athena/macros/materializations/models/table/table.sql
@@ -10,32 +10,19 @@
 
   {{ run_hooks(pre_hooks) }}
 
+  -- cleanup
   {%- if old_relation is not none -%}
-    {%- if format != 'iceberg' -%}
-      {{ adapter.drop_relation(old_relation) }}
-    {%- endif -%}
+    {{ drop_relation(old_relation) }}
   {%- endif -%}
 
-  {%- if format == 'iceberg' -%}
-    {%- set tmp_relation = make_temp_relation(target_relation) -%}
-	  {%- set build_sql = create_table_iceberg(target_relation, old_relation, tmp_relation, sql) -%}
-  {% else %}
-    {% set build_sql = create_table_as(False, target_relation, sql) -%}
-  {%- endif -%}
-
-  {% call statement("main") %}
-    {{ build_sql }}
-  {% endcall %}
+  -- build model
+  {% call statement('main') -%}
+    {{ create_table_as(False, target_relation, sql) }}
+  {%- endcall %}
 
-  -- drop tmp table in case of iceberg
-  {%- if format == 'iceberg' -%}
-  	{% do adapter.drop_relation(tmp_relation) %}
-  {%- endif -%}
-
-  -- set table properties
-  {%- if format != 'iceberg' -%}
-    {{ set_table_classification(target_relation, 'parquet') }}
-  {%- endif -%}
+  {% if format != 'iceberg' %}
+    {{ set_table_classification(target_relation) }}
+  {% endif %}
 
   {{ run_hooks(post_hooks) }}
 
diff --git a/dbt/include/athena/macros/materializations/seeds/helpers.sql b/dbt/include/athena/macros/materializations/seeds/helpers.sql
index 2ce2956..cbe916a 100644
--- a/dbt/include/athena/macros/materializations/seeds/helpers.sql
+++ b/dbt/include/athena/macros/materializations/seeds/helpers.sql
@@ -1,7 +1,7 @@
 {% macro default__reset_csv_table(model, full_refresh, old_relation, agate_table) %}
     {% set sql = "" %}
     -- No truncate in Athena so always drop CSV table and recreate
-    {{ adapter.drop_relation(old_relation) }}
+    {{ drop_relation(old_relation) }}
     {% set sql = create_csv_table(model, agate_table) %}
 
     {{ return(sql) }}


Test patch:
diff --git a/tests/unit/test_adapter.py b/tests/unit/test_adapter.py
index 5f7ac4a..b873c89 100644
--- a/tests/unit/test_adapter.py
+++ b/tests/unit/test_adapter.py
@@ -195,20 +195,61 @@ class TestAthenaAdapter:
         assert "Got an error when attempting to open a Athena connection due to foobar" in caplog.text
 
     @pytest.mark.parametrize(
-        ("s3_data_dir", "s3_data_naming", "expected"),
+        ("s3_data_dir", "s3_data_naming", "external_location", "is_temporary_table", "expected"),
         (
-            pytest.param(None, "table", "s3://my-bucket/test-dbt/tables/table/"),
-            pytest.param(None, "uuid", "s3://my-bucket/test-dbt/tables/uuid/"),
-            pytest.param(None, "table_unique", "s3://my-bucket/test-dbt/tables/table/uuid/"),
-            pytest.param(None, "schema_table", "s3://my-bucket/test-dbt/tables/schema/table/"),
-            pytest.param(None, "schema_table_unique", "s3://my-bucket/test-dbt/tables/schema/table/uuid/"),
-            pytest.param("s3://my-data-bucket/", "schema_table_unique", "s3://my-data-bucket/schema/table/uuid/"),
+            pytest.param(None, "table", None, False, "s3://my-bucket/test-dbt/tables/table", id="table naming"),
+            pytest.param(None, "uuid", None, False, "s3://my-bucket/test-dbt/tables/uuid", id="uuid naming"),
+            pytest.param(
+                None, "table_unique", None, False, "s3://my-bucket/test-dbt/tables/table/uuid", id="table_unique naming"
+            ),
+            pytest.param(
+                None,
+                "schema_table",
+                None,
+                False,
+                "s3://my-bucket/test-dbt/tables/schema/table",
+                id="schema_table naming",
+            ),
+            pytest.param(
+                None,
+                "schema_table_unique",
+                None,
+                False,
+                "s3://my-bucket/test-dbt/tables/schema/table/uuid",
+                id="schema_table_unique naming",
+            ),
+            pytest.param(
+                "s3://my-data-bucket/",
+                "schema_table_unique",
+                None,
+                False,
+                "s3://my-data-bucket/schema/table/uuid",
+                id="data_dir set",
+            ),
+            pytest.param(
+                "s3://my-data-bucket/",
+                "schema_table_unique",
+                "s3://path/to/external/",
+                False,
+                "s3://path/to/external",
+                id="external_location set and not temporary",
+            ),
+            pytest.param(
+                "s3://my-data-bucket/",
+                "schema_table_unique",
+                "s3://path/to/external/",
+                True,
+                "s3://my-data-bucket/schema/table/uuid",
+                id="external_location set and temporary",
+            ),
         ),
     )
     @patch("dbt.adapters.athena.impl.uuid4", return_value="uuid")
-    def test_s3_table_location(self, _, s3_data_dir, s3_data_naming, expected):
+    def test_s3_table_location(self, _, s3_data_dir, s3_data_naming, external_location, is_temporary_table, expected):
         self.adapter.acquire_connection("dummy")
-        assert expected == self.adapter.s3_table_location(s3_data_dir, s3_data_naming, "schema", "table")
+        assert expected == self.adapter.s3_table_location(
+            s3_data_dir, s3_data_naming, "schema", "table", external_location, is_temporary_table
+        )
 
     def test_s3_table_location_exc(self):
         self.adapter.acquire_connection("dummy")