Set unique temp table suffix to allow parallel incremental executions (…

…#811)

Signed-off-by: huang xingyi <hxy911122@gmail.com>
Co-authored-by: Ben Cassell <ben.cassell@databricks.com>

CHANGELOG.md

@@ -1,8 +1,12 @@
 ## dbt-databricks 1.8.7 (TBD)
 
+### Features
+
+- Add config for generating unique tmp table names for enabling parralel replace-where (thanks @huangxingyi-git!) ([811](https://github.com/databricks/dbt-databricks/pull/811))
+
 ### Fixes
 
-- Stop setting cluster by to None. If you want to drop liquid clustering, you will need to full-refresh ([806]https://github.com/databricks/dbt-databricks/pull/806)
+- Stop setting cluster by to None. If you want to drop liquid clustering, you will need to full-refresh ([806](https://github.com/databricks/dbt-databricks/pull/806))
 
 ## dbt-databricks 1.8.6 (September 18, 2024)
 

dbt/adapters/databricks/impl.py

@@ -23,6 +23,7 @@
 from typing import TYPE_CHECKING
 from typing import TypeVar
 from typing import Union
+from uuid import uuid4
 
 from dbt.adapters.base import AdapterConfig
 from dbt.adapters.base import PythonJobHelper
@@ -103,6 +104,7 @@ class DatabricksConfig(AdapterConfig):
     databricks_tags: Optional[Dict[str, str]] = None
     tblproperties: Optional[Dict[str, str]] = None
     zorder: Optional[Union[List[str], str]] = None
+    unique_tmp_table_suffix: bool = False
 
 
 def check_not_found_error(errmsg: str) -> bool:
@@ -685,6 +687,10 @@ def get_config_from_model(self, model: RelationConfig) -> DatabricksRelationConf
                 f"Materialization {model.config.materialized} is not supported."
             )
 
+    @available
+    def generate_unique_temporary_table_suffix(self, suffix_initial: str = "__dbt_tmp") -> str:
+        return f"{suffix_initial}_{str(uuid4())}"
+
 
 @dataclass(frozen=True)
 class RelationAPIBase(ABC, Generic[DatabricksRelationConfig]):

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

@@ -5,6 +5,7 @@
   {%- set grant_config = config.get('grants') -%}
   {%- set tblproperties = config.get('tblproperties') -%}
   {%- set tags = config.get('databricks_tags') -%}
+  {%- set unique_tmp_table_suffix = config.get('unique_tmp_table_suffix', False) | as_bool -%}
 
   {%- set file_format = dbt_databricks_validate_get_file_format(raw_file_format) -%}
   {%- set incremental_strategy = dbt_databricks_validate_get_incremental_strategy(raw_strategy, file_format) -%}
@@ -18,7 +19,11 @@
   {%- set on_schema_change = incremental_validate_on_schema_change(config.get('on_schema_change'), default='ignore') -%}
   {%- set target_relation = this.incorporate(type='table') -%}
   {%- set existing_relation = adapter.get_relation(database=this.database, schema=this.schema, identifier=this.identifier, needs_information=True) -%}
-
+  {%- if unique_tmp_table_suffix == True and raw_strategy == 'replace_where' and raw_file_format == 'delta' -%}
+    {%- set temp_relation_suffix = adapter.generate_unique_temporary_table_suffix() -%}
+  {%- else -%}
+    {%- set temp_relation_suffix = '__dbt_tmp' -%}
+  {%- endif -%}
 
   {#-- Set Overwrite Mode to STATIC for initial replace --#}
   {%- if incremental_strategy == 'insert_overwrite' and should_full_refresh() -%}
@@ -69,7 +74,7 @@
     {%- set _existing_config = adapter.get_relation_config(existing_relation) -%}
     {%- set model_config = adapter.get_config_from_model(config.model) -%}
     {%- set _configuration_changes = model_config.get_changeset(_existing_config) -%}
-    {%- set temp_relation = databricks__make_temp_relation(target_relation, as_table=language != 'sql') -%}
+    {%- set temp_relation = databricks__make_temp_relation(target_relation, suffix=temp_relation_suffix, as_table=language != 'sql') -%}
     {%- call statement('create_temp_relation', language=language) -%}
       {{ create_table_as(True, temp_relation, compiled_code, language) }}
     {%- endcall -%}