Fix: drop temp tables in python model materialization v2 (#1026)

Author: ericj-db

CHANGELOG.md

@@ -7,6 +7,7 @@
 ### Fixes
 
 - Fix bug with multiple not_null constraints defined on the model level ([1008](https://github.com/databricks/dbt-databricks/pull/1008))
+- Fix bug with temp tables not being dropped after python model is materialized ([1010](https://github.com/databricks/dbt-databricks/issues/1010))
 
 ## dbt-databricks 1.10.1 (Apr 29, 2025)
 

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

@@ -73,6 +73,10 @@
     {% do apply_grants(target_relation, grant_config, should_revoke) %}
     {% do optimize(target_relation) %}
 
+    {% if language == 'python' %}
+      {{ drop_relation_if_exists(intermediate_relation) }}
+    {% endif %}
+
     {{ run_post_hooks() }}
 
   {% else %}

dbt/include/databricks/macros/materializations/table.sql

@@ -34,6 +34,10 @@
 
     {% set should_revoke = should_revoke(existing_relation, full_refresh_mode=True) %}
     {{ apply_grants(target_relation, grant_config, should_revoke) }}
+
+    {% if language == 'python' %}
+      {{ drop_relation_if_exists(intermediate_relation) }}
+    {% endif %}
 
     {{ run_post_hooks() }}
   {% else %}

tests/functional/adapter/incremental/fixtures.py

@@ -549,6 +549,7 @@ def model(dbt, spark):
 models:
   - name: tags
     config:
+      unique_tmp_table_suffix: true
       tags: ["python"]
       databricks_tags:
         c: e

tests/functional/adapter/python_model/fixtures.py

@@ -89,6 +89,7 @@ def model(dbt, spark):
 def model(dbt, spark):
     dbt.config(
         materialized='table',
+        unique_tmp_table_suffix=True
     )
     data = [[1,2]] * 10
     return spark.createDataFrame(data, schema=['test1', 'test3'])
@@ -190,3 +191,26 @@ def model(dbt, spark):
 2,"Fang"
 3,"Elbert"
 """
+
+simple_incremental_python_model = """
+import pandas
+
+def model(dbt, spark):
+    dbt.config(
+        materialized='incremental',
+    )
+    data = [[1,2]] * 5
+    return spark.createDataFrame(data, schema=['test', 'test2'])
+"""
+
+simple_incremental_python_model_v2 = """
+import pandas
+
+def model(dbt, spark):
+    dbt.config(
+        materialized='incremental',
+        unique_tmp_table_suffix=True,
+    )
+    data = [[1,2]] * 10
+    return spark.createDataFrame(data, schema=['test', 'test2'])
+"""

tests/functional/adapter/python_model/test_python_model.py

@@ -8,9 +8,22 @@
     BasePythonIncrementalTests,
     BasePythonModelTests,
 )
+from tests.functional.adapter.fixtures import MaterializationV2Mixin
 from tests.functional.adapter.python_model import fixtures as override_fixtures
 
 
+def verify_temp_tables_cleaned(project):
+    verify_temp_table_cleaned(project, "*__dbt_stg")
+    verify_temp_table_cleaned(project, "*__dbt_tmp")
+
+
+def verify_temp_table_cleaned(project, suffix):
+    tmp_tables = project.run_sql(
+        "SHOW TABLES IN {database}.{schema} LIKE '" + f"{suffix}'", fetch="all"
+    )
+    assert len(tmp_tables) == 0
+
+
 @pytest.mark.python
 @pytest.mark.skip_profile("databricks_uc_sql_endpoint")
 class TestPythonModel(BasePythonModelTests):
@@ -207,3 +220,39 @@ def test_workflow_run(self, project):
             "SELECT * FROM {database}.{schema}.my_workflow_model", fetch="all"
         )
         assert len(sql_results) == 10
+
+
+@pytest.mark.python
+@pytest.mark.skip_profile("databricks_uc_sql_endpoint")
+class TestChangingSchemaV2(MaterializationV2Mixin):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"simple_python_model.py": override_fixtures.simple_python_model}
+
+    def test_changing_unique_tmp_table_suffix(self, project):
+        util.run_dbt(["run"])
+        util.write_file(
+            override_fixtures.simple_python_model_v2,
+            project.project_root + "/models",
+            "simple_python_model.py",
+        )
+        util.run_dbt(["run"])
+        verify_temp_tables_cleaned(project)
+
+
+@pytest.mark.python
+@pytest.mark.skip_profile("databricks_uc_sql_endpoint")
+class TestChangingSchemaIncrementalV2(MaterializationV2Mixin):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"incremental_model.py": override_fixtures.simple_incremental_python_model}
+
+    def test_changing_unique_tmp_table_suffix(self, project):
+        util.run_dbt(["run"])
+        util.write_file(
+            override_fixtures.simple_incremental_python_model_v2,
+            project.project_root + "/models",
+            "incremental_model.py",
+        )
+        util.run_dbt(["run"])
+        verify_temp_tables_cleaned(project)