ADAP-1051 - Temp Table Drop Fix (#1076)

* Fix for ADAP-1051

* adding changie files

* adding changie files

* Revert "adding changie files"

This reverts commit f51ca0c.

* Add test cases

* Revert adapters.sql

* Update body for changie

---------

Co-authored-by: Doug Beatty <[email protected]>
Co-authored-by: Mike Alfare <[email protected]>
Co-authored-by: Teresa Martyny <[email protected]>
Co-authored-by: Mila Page <[email protected]>

.changes/unreleased/Fixes-20240120-180818.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Drop intermediate objects created in BigQuery for incremental models
+time: 2024-01-20T18:08:18.817915-06:00
+custom:
+  Author: vinit2107
+  Issue: "1036"

dbt/include/bigquery/macros/materializations/incremental.sql

@@ -151,10 +151,6 @@
       {{ build_sql }}
     {% endcall %}
 
-    {%- if language == 'python' and tmp_relation -%}
-      {{ adapter.drop_relation(tmp_relation) }}
-    {%- endif -%}
-
   {% endif %}
 
   {{ run_hooks(post_hooks) }}
@@ -166,6 +162,10 @@
 
   {% do persist_docs(target_relation, model) %}
 
+  {%- if tmp_relation_exists -%}
+    {{ adapter.drop_relation(tmp_relation) }}
+  {%- endif -%}
+
   {{ return({'relations': [target_relation]}) }}
 
 {%- endmaterialization %}

tests/functional/test_drop_temp_relation.py

@@ -0,0 +1,60 @@
+import pytest
+from google.api_core.exceptions import NotFound
+from dbt.adapters.bigquery.relation import BigQueryRelation
+from dbt.tests.util import run_dbt, get_connection, relation_from_name
+
+
+_INCREMENTAL_MODEL = """
+{{
+    config(
+        materialized="incremental",
+        on_schema_change="sync_all_columns"
+    )
+}}
+    select 20 as id, cast('2020-01-01 01:00:00' as datetime) as date_hour union all
+    select 40 as id, cast('2020-01-01 02:00:00' as datetime) as date_hour
+"""
+
+_INCREMENTAL_MODEL_YAML = """version: 2
+models:
+- name: test_drop_relation
+  columns:
+  - name: id
+    type: int64
+  - name: date_hour
+    type: datetime
+"""
+
+
+class BaseIncrementalModelConfig:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "test_drop_relation.sql": _INCREMENTAL_MODEL,
+            "schema.yml": _INCREMENTAL_MODEL_YAML,
+        }
+
+
+class TestIncrementalModel(BaseIncrementalModelConfig):
+    def test_incremental_model_succeeds(self, project):
+        """
+        Steps:
+        1. Create the model
+        2. Merge into the model using __dbt_tmp table
+        3. Assert raises NotFound exception
+        """
+        results = run_dbt(["run"])
+        assert len(results) == 1
+        results = run_dbt(["run"])
+        assert len(results) == 1
+        relation: BigQueryRelation = relation_from_name(
+            project.adapter, "test_drop_relation__dbt_tmp"
+        )
+        adapter = project.adapter
+        with pytest.raises(NotFound):
+            with get_connection(project.adapter) as conn:
+                conn.handle.get_table(
+                    adapter.connections.get_bq_table(
+                        relation.database, relation.schema, relation.table
+                    )
+                )