feat: unique table suffix to allow parallel incremental executions

dbt/adapters/athena/impl.py

@@ -419,6 +419,10 @@ def clean_up_table(self, relation: AthenaRelation) -> None:
         if table_location := self.get_glue_table_location(relation):
             self.delete_from_s3(table_location)
 
+    @available
+    def generate_unique_temporary_table_suffix(self, suffix_initial="__dbt_tmp", length=8):
+        return f"{suffix_initial}_{str(uuid4())[:length]}"
+
     def quote(self, identifier: str) -> str:
         return f"{self.quote_character}{identifier}{self.quote_character}"
 

dbt/include/athena/macros/materializations/models/incremental/incremental.sql

@@ -11,10 +11,11 @@
   {% set force_batch = config.get('force_batch', False) | as_bool -%}
   {% set target_relation = this.incorporate(type='table') %}
   {% set existing_relation = load_relation(this) %}
-  {% set old_tmp_relation = adapter.get_relation(identifier=target_relation.identifier ~ '__dbt_tmp',
+  {% set tmp_table_suffix = adapter.generate_unique_temporary_table_suffix() %}
+  {% set old_tmp_relation = adapter.get_relation(identifier=target_relation.identifier ~ tmp_table_suffix,
                                              schema=schema,
                                              database=database) %}
-  {% set tmp_relation = make_temp_relation(target_relation, '__dbt_tmp') %}
+  {% set tmp_relation = make_temp_relation(target_relation, suffix=tmp_table_suffix) %}
 
   -- If no partitions are used with insert_overwrite, we fall back to append mode.
   {% if partitioned_by is none and strategy == 'insert_overwrite' %}