add commit option

Author: borjavb

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

@@ -26,9 +26,9 @@
   {%- set incremental_substrategy = config.get('incremental_substrategy', 'copy_partitions' if copy_partitions else 'merge') -%}
 
   {% if strategy in ['insert_overwrite', 'microbatch'] %}
-    {% if incremental_substrategy not in ['merge', 'delete+insert', 'copy_partitions'] %}
+    {% if incremental_substrategy not in ['merge', 'commit+delete+insert', 'delete+insert', 'copy_partitions'] %}
       {% set wrong_fn -%}
-      The 'incremental_substrategy' option has to be either 'merge' (default), 'delete+insert' or 'copy_partitions'.
+      The 'incremental_substrategy' option has to be either 'merge' (default), 'commit+delete+insert', 'delete+insert' or 'copy_partitions'.
       {%- endset %}
       {% do exceptions.raise_compiler_error(wrong_fn) %}
     {% endif %}

dbt/include/bigquery/macros/materializations/incremental_strategy/insert_overwrite.sql

@@ -87,8 +87,11 @@
         #}
 
         {% if incremental_substrategy == 'delete+insert' %}
-          -- 1. run insert_overwrite with delete+insert transaction strategy optimisation
+          -- 1. run insert_overwrite with delete+insert (without a transaction) strategy optimisation
           {{ bq_get_insert_overwrite_with_delete_and_insert_sql(target_relation, source_sql, dest_columns, [predicate], include_sql_header = not tmp_relation_exists) }};
+        {% elif incremental_substrategy == 'commit+delete+insert' %}
+          -- 1. run insert_overwrite with delete+insert (with a transaction) strategy optimisation
+          {{ bq_get_insert_overwrite_with_delete_and_insert_sql(target_relation, source_sql, dest_columns, [predicate], include_sql_header = not tmp_relation_exists, transactional=True) }};
         {% else %}
           -- 1. run insert_overwrite with merge strategy optimisation
           {{ get_insert_overwrite_merge_sql(target_relation, source_sql, dest_columns, [predicate], include_sql_header = not tmp_relation_exists) }};
@@ -162,8 +165,11 @@
       );
 
       {% if  incremental_substrategy == 'delete+insert' %}
-        -- 3. run insert_overwrite with the delete+insert transaction strategy optimisation
+        -- 3. run insert_overwrite with the delete+insert (without a transaction) strategy optimisation
         {{ bq_get_insert_overwrite_with_delete_and_insert_sql(target_relation, source_sql, dest_columns, [predicate]) }};
+      {% elif incremental_substrategy == 'commit+delete+insert' %}
+        -- 3. run insert_overwrite with the delete+insert (with a transaction) strategy optimisation
+        {{ bq_get_insert_overwrite_with_delete_and_insert_sql(target_relation, source_sql, dest_columns, [predicate], transactional=True) }};
       {% else %}
         -- 3. run insert_overwrite with the merge strategy optimisation
         {{ get_insert_overwrite_merge_sql(target_relation, source_sql, dest_columns, [predicate]) }};
@@ -177,39 +183,39 @@
 
 
 
-{% macro bq_get_insert_overwrite_with_delete_and_insert_sql(target, source, dest_columns, predicates, include_sql_header) -%}
+{% macro bq_get_insert_overwrite_with_delete_and_insert_sql(target, source, dest_columns, predicates, include_sql_header, transactional=False) -%}
     {%- set predicates = [] if predicates is none else [] + predicates -%}
     {%- set dest_cols_csv = get_quoted_csv(dest_columns | map(attribute="name")) -%}
     {%- set sql_header = config.get('sql_header', none) -%}
 
     {{ sql_header if sql_header is not none and include_sql_header }}
 
+    {% if transactional %}
+    -- We can rely on a multi-statement transaction to enable atomicity
+    -- If something goes south, nothing is committed
+    -- DELETE + INSERT allow for isolation lock
     begin
-        begin transaction;
-
-            -- (as of Nov 2024)
-            -- DELETE operations are free if the partition is a DATE
-            -- * Not free if the partitions are granular (hourly, monthly)
-            --   or some other conditions like subqueries and so on.
-            delete from {{ target }} as DBT_INTERNAL_DEST
-            where true
-            {%- if predicates %}
-                {% for predicate in predicates %}
-                    and {{ predicate }}
-                {% endfor %}
-            {%- endif -%};
-
-
-            insert into {{ target }} ({{ dest_cols_csv }})
-            (
-                select {{ dest_cols_csv }}
-                from {{ source }}
-            );
+    begin transaction ;
+    {% endif %}
+        -- DELETE operations are free https://cloud.google.com/bigquery/docs/using-dml-with-partitioned-tables#using_dml_delete_to_delete_partitions
+        delete from {{ target }} as DBT_INTERNAL_DEST
+        where true
+        {% if predicates %} and {{ predicates | join(' and ') }} {% endif %};
 
-        commit transaction;
+        -- INSERT the data
+        insert into {{ target }} ({{ dest_cols_csv }})
+        (
+            select {{ dest_cols_csv }}
+            from {{ source }}
+        )
+    {% if transactional %}
+    -- leaving the trailing ; out of the if as the calling macro already adds a leading ; to this output
+    ; commit transaction;
 
     exception when error then
-        raise using message = FORMAT("Error: %s", @@error.message);
+        -- If things go south, abort and rollback
+        raise using message = FORMAT("dbt error while commit+delete+instert: %s", @@error.message);
         rollback transaction;
     end
+    {% endif %}
 {% endmacro %}

tests/functional/adapter/incremental/incremental_strategy_fixtures.py

@@ -714,6 +714,55 @@
 select * from data
 """.lstrip()
 
+overwrite_static_day_commit_delete_and_insert_sub_strategy_sql = """
+{% set partitions_to_replace = [
+  "'2020-01-01'",
+  "'2020-01-02'",
+] %}
+
+{{
+    config(
+        materialized="incremental",
+        incremental_strategy="insert_overwrite",
+        incremental_substrategy='commit+delete+insert',
+        cluster_by="id",
+        partition_by={
+            "field": "date_time",
+            "data_type": "datetime",
+            "granularity": "day"
+        },
+        partitions=partitions_to_replace,
+        on_schema_change="sync_all_columns"
+    )
+}}
+
+
+with data as (
+
+    {% if not is_incremental() %}
+
+        select 1 as id, cast('2020-01-01' as datetime) as date_time union all
+        select 2 as id, cast('2020-01-01' as datetime) as date_time union all
+        select 3 as id, cast('2020-01-01' as datetime) as date_time union all
+        select 4 as id, cast('2020-01-01' as datetime) as date_time
+
+    {% else %}
+
+        -- we want to overwrite the 4 records in the 2020-01-01 partition
+        -- with the 2 records below, but add two more in the 2020-01-02 partition
+        select 10 as id, cast('2020-01-01' as datetime) as date_time union all
+        select 20 as id, cast('2020-01-01' as datetime) as date_time union all
+        select 30 as id, cast('2020-01-02' as datetime) as date_time union all
+        select 40 as id, cast('2020-01-02' as datetime) as date_time
+
+    {% endif %}
+
+)
+
+select * from data
+""".lstrip()
+
+
 
 overwrite_static_day_merge_sub_strategy_sql = """
 {% set partitions_to_replace = [

tests/functional/adapter/incremental/test_incremental_strategies.py

@@ -28,6 +28,7 @@
     overwrite_day_with_time_partition_datetime_sql,
     overwrite_static_day_sql,
     overwrite_static_day_delete_and_insert_sub_strategy_sql,
+    overwrite_static_day_commit_delete_and_insert_sub_strategy_sql,
     overwrite_static_day_merge_sub_strategy_sql,
     overwrite_static_day_copy_partitions_sub_strategy_sql,
 )
@@ -54,6 +55,7 @@ def models(self):
             "incremental_overwrite_day_with_time_partition_datetime.sql": overwrite_day_with_time_partition_datetime_sql,
             "incremental_overwrite_static_substrategy_day.sql": overwrite_static_day_sql,
             "incremental_overwrite_static_substrategy_day_with_deleteinsert.sql": overwrite_static_day_delete_and_insert_sub_strategy_sql,
+            "incremental_overwrite_static_substrategy_day_with_commitdeleteinsert.sql": overwrite_static_day_commit_delete_and_insert_sub_strategy_sql,
             "incremental_overwrite_static_substrategy_day_with_merge.sql": overwrite_static_day_merge_sub_strategy_sql,
             "incremental_overwrite_static_substrategy_day_with_copy_partitions.sql": overwrite_static_day_copy_partitions_sub_strategy_sql,
         }
@@ -95,6 +97,10 @@ def test__bigquery_assert_incremental_configurations_apply_the_right_strategy(se
                 "incremental_overwrite_static_substrategy_day_with_deleteinsert",
                 "incremental_overwrite_day_expected",
             ),
+            (
+                "incremental_overwrite_static_substrategy_day_with_commitdeleteinsert",
+                "incremental_overwrite_day_expected",
+            ),
             (
                 "incremental_overwrite_static_substrategy_day_with_merge",
                 "incremental_overwrite_day_expected",