feat(athena): add build_with_subquery config to skip tmp table staging

Signed-off-by: Daisuke Taniwaki <daisuketaniwaki@gmail.com>

Author: dtaniwaki

dbt-athena/.changes/unreleased/Features-20260406-225458.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: build_with_subquery config option to skip tmp table staging for incremental merge and append strategies
+time: 2026-04-06T22:54:58.739286+09:00
+custom:
+    Author: dtaniwaki
+    Issue: "1829"

dbt-athena/src/dbt/adapters/athena/impl.py

@@ -104,6 +104,9 @@ class AthenaConfig(AdapterConfig):
         force_batch: Skip creating the table as ctas and run the operation directly in batch insert mode.
         unique_tmp_table_suffix: Enforce the use of a unique id as tmp table suffix instead of __dbt_tmp.
         temp_schema: Define in which schema to create temporary tables used in incremental runs.
+        build_with_subquery: Use a subquery directly instead of staging data into __dbt_tmp.
+            Creates an empty tmp table (WITH NO DATA) for schema comparison, then applies via subquery.
+            Supported for iceberg merge and append strategies. Incompatible with force_batch.
     """
 
     work_group: Optional[str] = None
@@ -127,6 +130,7 @@ class AthenaConfig(AdapterConfig):
     force_batch: bool = False
     unique_tmp_table_suffix: bool = False
     temp_schema: Optional[str] = None
+    build_with_subquery: bool = False
 
 
 class AthenaAdapter(SQLAdapter):

dbt-athena/src/dbt/include/athena/macros/materializations/models/incremental/helpers.sql

@@ -53,6 +53,7 @@
     target_relation,
     existing_relation,
     force_batch,
+    source_sql=none,
     statement_name="main"
   )
 %}
@@ -65,18 +66,32 @@
     {% if force_batch %}
         {% do batch_incremental_insert(tmp_relation, target_relation, dest_cols_csv) %}
     {% else %}
-      {%- set insert_full -%}
-          insert into {{ target_relation }} ({{ dest_cols_csv }})
-              (
-                 select {{ dest_cols_csv }}
-                 from {{ tmp_relation }}
-              );
-      {%- endset -%}
+      {%- if source_sql is not none -%}
+        {%- set insert_full -%}
+            insert into {{ target_relation }} ({{ dest_cols_csv }})
+                (
+                   select {{ dest_cols_csv }}
+                   from ({{ source_sql }}) _dbt_sbq
+                );
+        {%- endset -%}
+      {%- else -%}
+        {%- set insert_full -%}
+            insert into {{ target_relation }} ({{ dest_cols_csv }})
+                (
+                   select {{ dest_cols_csv }}
+                   from {{ tmp_relation }}
+                );
+        {%- endset -%}
+      {%- endif -%}
 
       {%- set query_result =  adapter.run_query_with_partitions_limit_catching(insert_full) -%}
       {%- do log('QUERY RESULT: ' ~ query_result) -%}
       {%- if query_result == 'TOO_MANY_OPEN_PARTITIONS' -%}
+        {%- if source_sql is not none -%}
+          {% do exceptions.raise_compiler_error('Runtime error: TOO_MANY_OPEN_PARTITIONS encountered with build_with_subquery=True. Disable build_with_subquery to enable automatic batching.') %}
+        {%- else -%}
           {% do batch_incremental_insert(tmp_relation, target_relation, dest_cols_csv) %}
+        {%- endif -%}
       {%- endif -%}
     {%- endif -%}
 

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

@@ -10,6 +10,7 @@
   {% set lf_grants = config.get('lf_grants') %}
   {% set partitioned_by = config.get('partitioned_by') %}
   {% set force_batch = config.get('force_batch', False) | as_bool -%}
+  {% set build_with_subquery = config.get('build_with_subquery', False) | as_bool -%}
   {% set unique_tmp_table_suffix = config.get('unique_tmp_table_suffix', False) | as_bool -%}
   {% set temp_schema = config.get('temp_schema') %}
   {% set target_relation = this.incorporate(type='table') %}
@@ -146,22 +147,43 @@
 
   -- Append Strategy --
   {% elif strategy == 'append' %}
-    {% if old_tmp_relation is not none %}
-      {% do drop_relation(old_tmp_relation) %}
+    {% if build_with_subquery %}
+      {% if model_language == 'python' %}
+        {% do exceptions.raise_compiler_error('build_with_subquery is not supported with Python models.') %}
+      {% endif %}
+      {% if force_batch %}
+        {% do exceptions.raise_compiler_error('build_with_subquery is incompatible with force_batch. Batching requires data in the tmp table.') %}
+      {% endif %}
+
+      {% if old_tmp_relation is not none %}
+        {% do drop_relation(old_tmp_relation) %}
+      {% endif %}
+      {%- set empty_sql = 'SELECT * FROM (' ~ compiled_code ~ ') _dbt_sbq WITH NO DATA' -%}
+      {% do run_query(create_table_as(True, tmp_relation, empty_sql)) %}
+
+      {% set build_sql = incremental_insert(
+          on_schema_change, tmp_relation, target_relation, existing_relation, false, source_sql=compiled_code
+        )
+      %}
+      {% do to_drop.append(tmp_relation) %}
+    {% else %}
+      {% if old_tmp_relation is not none %}
+        {% do drop_relation(old_tmp_relation) %}
+      {% endif %}
+      {% set query_result = safe_create_table_as(True, tmp_relation, compiled_code, model_language, force_batch) -%}
+      {%- if model_language == 'python' -%}
+        {% call statement('create_table', language=model_language) %}
+          {{ query_result }}
+        {% endcall %}
+      {%- endif -%}
+      {% set build_sql = incremental_insert(
+          on_schema_change, tmp_relation, target_relation, existing_relation, force_batch
+        )
+      %}
+      {% do to_drop.append(tmp_relation) %}
     {% endif %}
-    {% set query_result = safe_create_table_as(True, tmp_relation, compiled_code, model_language, force_batch) -%}
-    {%- if model_language == 'python' -%}
-      {% call statement('create_table', language=model_language) %}
-        {{ query_result }}
-      {% endcall %}
-    {%- endif -%}
-    {% set build_sql = incremental_insert(
-        on_schema_change, tmp_relation, target_relation, existing_relation, force_batch
-      )
-    %}
-    {% do to_drop.append(tmp_relation) %}
 
-  -- Iceberge Merge Stategy --
+  -- Iceberg Merge Strategy --
   {% elif strategy == 'merge' and table_type == 'iceberg' %}
     {% set unique_key = config.get('unique_key') %}
     {% set incremental_predicates = config.get('incremental_predicates') %}
@@ -186,29 +208,63 @@
         {% do exceptions.raise_compiler_error(inc_predicates_not_list) %}
       {% endif %}
     {% endif %}
-    {% if old_tmp_relation is not none %}
-      {% do drop_relation(old_tmp_relation) %}
+
+    {% if build_with_subquery %}
+      -- Validate build_with_subquery constraints
+      {% if model_language == 'python' %}
+        {% do exceptions.raise_compiler_error('build_with_subquery is not supported with Python models.') %}
+      {% endif %}
+      {% if force_batch %}
+        {% do exceptions.raise_compiler_error('build_with_subquery is incompatible with force_batch. Batching requires data in the tmp table.') %}
+      {% endif %}
+
+      -- Create empty tmp table for schema comparison (no data scan)
+      {% if old_tmp_relation is not none %}
+        {% do drop_relation(old_tmp_relation) %}
+      {% endif %}
+      {%- set empty_sql = 'SELECT * FROM (' ~ compiled_code ~ ') _dbt_sbq WITH NO DATA' -%}
+      {% do run_query(create_table_as(True, tmp_relation, empty_sql)) %}
+
+      {% set build_sql = iceberg_merge(
+          on_schema_change=on_schema_change,
+          tmp_relation=tmp_relation,
+          target_relation=target_relation,
+          unique_key=unique_key,
+          incremental_predicates=incremental_predicates,
+          existing_relation=existing_relation,
+          delete_condition=delete_condition,
+          update_condition=update_condition,
+          insert_condition=insert_condition,
+          force_batch=false,
+          source_sql=compiled_code,
+        )
+      %}
+      {% do to_drop.append(tmp_relation) %}
+    {% else %}
+      {% if old_tmp_relation is not none %}
+        {% do drop_relation(old_tmp_relation) %}
+      {% endif %}
+      {% set query_result = safe_create_table_as(True, tmp_relation, compiled_code, model_language, force_batch) -%}
+      {%- if model_language == 'python' -%}
+        {% call statement('create_table', language=model_language) %}
+          {{ query_result }}
+        {% endcall %}
+      {%- endif -%}
+      {% set build_sql = iceberg_merge(
+          on_schema_change=on_schema_change,
+          tmp_relation=tmp_relation,
+          target_relation=target_relation,
+          unique_key=unique_key,
+          incremental_predicates=incremental_predicates,
+          existing_relation=existing_relation,
+          delete_condition=delete_condition,
+          update_condition=update_condition,
+          insert_condition=insert_condition,
+          force_batch=force_batch,
+        )
+      %}
+      {% do to_drop.append(tmp_relation) %}
     {% endif %}
-    {% set query_result = safe_create_table_as(True, tmp_relation, compiled_code, model_language, force_batch) -%}
-    {%- if model_language == 'python' -%}
-      {% call statement('create_table', language=model_language) %}
-        {{ query_result }}
-      {% endcall %}
-    {%- endif -%}
-    {% set build_sql = iceberg_merge(
-        on_schema_change=on_schema_change,
-        tmp_relation=tmp_relation,
-        target_relation=target_relation,
-        unique_key=unique_key,
-        incremental_predicates=incremental_predicates,
-        existing_relation=existing_relation,
-        delete_condition=delete_condition,
-        update_condition=update_condition,
-        insert_condition=insert_condition,
-        force_batch=force_batch,
-      )
-    %}
-    {% do to_drop.append(tmp_relation) %}
   {% endif %}
 
   {% call statement("main", language=model_language) %}

dbt-athena/src/dbt/include/athena/macros/materializations/models/incremental/merge.sql

@@ -77,6 +77,7 @@
     update_condition,
     insert_condition,
     force_batch,
+    source_sql=none,
     statement_name="main"
   )
 %}
@@ -143,9 +144,15 @@
     {%- if force_batch -%}
       {% do batch_iceberg_merge(tmp_relation, target_relation, merge_part, dest_cols_csv) %}
     {%- else -%}
-      {%- set src_part -%}
-          merge into {{ target_relation }} as target using {{ tmp_relation }} as src
-      {%- endset -%}
+      {%- if source_sql is not none -%}
+        {%- set src_part -%}
+            merge into {{ target_relation }} as target using ({{ source_sql }}) as src
+        {%- endset -%}
+      {%- else -%}
+        {%- set src_part -%}
+            merge into {{ target_relation }} as target using {{ tmp_relation }} as src
+        {%- endset -%}
+      {%- endif -%}
       {%- set merge_full -%}
           {{ src_part }}
           {{ merge_part }}
@@ -154,7 +161,11 @@
       {%- set query_result =  adapter.run_query_with_partitions_limit_catching(merge_full) -%}
       {%- do log('QUERY RESULT: ' ~ query_result) -%}
       {%- if query_result == 'TOO_MANY_OPEN_PARTITIONS' -%}
-        {% do batch_iceberg_merge(tmp_relation, target_relation, merge_part, dest_cols_csv) %}
+        {%- if source_sql is not none -%}
+          {% do exceptions.raise_compiler_error('Runtime error: TOO_MANY_OPEN_PARTITIONS encountered with build_with_subquery=True. Disable build_with_subquery to enable automatic batching.') %}
+        {%- else -%}
+          {% do batch_iceberg_merge(tmp_relation, target_relation, merge_part, dest_cols_csv) %}
+        {%- endif -%}
       {%- endif -%}
     {%- endif -%}