fix delete+insert materialization behaviour with specified location

.changes/unreleased/Fixes-20250220-121852.yaml

@@ -0,0 +1,7 @@
+kind: Fixes
+body: fix delete+insert materialization behaviour with specified location
+time: 2025-02-20T12:18:52.200467772+03:00
+custom:
+    Author: yakovlevvs
+    Issue: "467"
+    PR: "468"

dbt/include/trino/macros/adapters.sql

@@ -82,7 +82,7 @@
   {{ return(sql) }}
 {% endmacro %}
 
-{% macro properties() %}
+{% macro properties(temporary=False) %}
   {%- set _properties = config.get('properties') -%}
   {%- set table_format = config.get('table_format') -%}
   {%- set file_format = config.get('file_format') -%}
@@ -117,6 +117,19 @@
     {%- endif -%}
   {%- endif -%}
 
+  {%- if temporary -%}
+    {%- if _properties -%}
+      {%- if _properties.location -%}
+          {%- if _properties.location[0] == "'" and _properties.location[-1] == "'" -%}
+            {%- set location = _properties.location[1:-1] -%}
+          {%- else -%}
+            {%- set location = _properties.location -%}
+          {%- endif -%}
+          {%- do _properties.update({'location': "'" ~ location ~ "__dbt_tmp'"}) -%}
+      {%- endif -%}
+    {%- endif -%}
+  {%- endif -%}
+
   {%- if _properties is not none -%}
       WITH (
           {%- for key, value in _properties.items() -%}
@@ -154,7 +167,7 @@
     {{ get_assert_columns_equivalent(sql) }}
     {%- set sql = get_select_subquery(sql) %}
     {{ comment(model.get('description')) }}
-    {{ properties() }}
+    {{ properties(temporary) }}
   ;
 
   insert into {{ relation }}
@@ -167,7 +180,7 @@
 
     create{{ or_replace }} table {{ relation }}
       {{ comment(model.get('description')) }}
-      {{ properties() }}
+      {{ properties(temporary) }}
     as (
       {{ sql }}
     );

tests/functional/adapter/materialization/test_incremental_delete_insert.py

@@ -1,3 +1,5 @@
+from dbt.tests.util import run_dbt_and_capture
+
 import pytest
 from dbt.tests.adapter.incremental.test_incremental_unique_id import (
     BaseIncrementalUniqueKey,
@@ -68,6 +70,38 @@
 
 """
 
+models__location_specified = """
+{{
+    config(
+        materialized='incremental',
+        incremental_strategy='delete+insert',
+        unique_key=['state', 'county', 'city'],
+        properties= {
+            "format": "'PARQUET'",
+            "format_version": "2",
+            "location": "'s3a://datalake/model'"
+        }
+    )
+}}
+
+select
+    'CT' as state,
+    'Hartford' as county,
+    'Hartford' as city,
+    cast('2022-02-14' as date) as last_visit_date
+union all
+select 'MA','Suffolk','Boston',DATE '2020-02-12'
+union all
+select 'NJ','Mercer','Trenton',DATE '2022-01-01'
+union all
+select 'NY','Kings','Brooklyn',DATE '2021-04-02'
+union all
+select 'NY','New York','Manhattan',DATE '2021-04-01'
+union all
+select 'PA','Philadelphia','Philadelphia',DATE '2021-05-21'
+
+"""
+
 models__expected__unique_key_list__inplace_overwrite_sql = """
 {{
     config(
@@ -146,3 +180,25 @@ def project_config_update(self):
             "models": {"+on_table_exists": "drop", "+incremental_strategy": "delete+insert"},
             "seeds": {"incremental": {"seed": {"+column_types": {"some_date": "date"}}}},
         }
+
+@pytest.mark.iceberg
+class TestIcebergIncrementalDeleteInsertWithLocation():
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "model.sql": models__location_specified,
+        }
+
+    def test_table_properties(self, project):
+
+        # Create model with properties
+        results, logs = run_dbt_and_capture(["--debug", "run"], expect_pass=True)
+        assert len(results) == 1
+        assert "location = 's3a://datalake/model'" in logs
+
+        # Temporary table is created on the second run
+        # So, now we check if the second run is successful and location
+        # is patched correctly
+        results, logs = run_dbt_and_capture(["--debug", "run"], expect_pass=True)
+        assert "location = 's3a://datalake/model__dbt_tmp'" in logs

tests/functional/adapter/test_table_properties.py

@@ -32,6 +32,7 @@ def project_config_update(self):
                 "+properties": {
                     "format": "'PARQUET'",
                     "format_version": "2",
+                    "location": "'s3a://datalake/model'"
                 },
             },
         }
@@ -47,6 +48,7 @@ def test_table_properties(self, project):
         assert "WITH (" in logs
         assert "format = 'PARQUET'" in logs
         assert "format_version = 2" in logs
+        assert "location = 's3a://datalake/model'" in logs
 
 
 @pytest.mark.iceberg