dbt-data-reliability/macros/utils/table_operations/replace_table_data.sql at d0fb760d664977a03a89ec0465d4feae22caf0f5 · elementary-data/dbt-data-reliability · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
{% macro replace_table_data(relation, rows) %}
    {{ return(adapter.dispatch("replace_table_data", "elementary")(relation, rows)) }}
{% endmacro %}

{# Default (Bigquery & Snowflake) - upload data to a temp table, and then atomically replace the table with a new one #}
{% macro default__replace_table_data(relation, rows) %}
    {% set intermediate_relation = elementary.create_intermediate_relation(
        relation, rows, temporary=True
    ) %}
    {% do elementary.edr_create_table_as(
        False, relation, "select * from {}".format(intermediate_relation)
    ) %}
    {% do adapter.drop_relation(intermediate_relation) %}
{% endmacro %}

{# Databricks - truncate and insert (non-atomic) #}
{% macro databricks__replace_table_data(relation, rows) %}
    {% do dbt.truncate_relation(relation) %}
    {% do elementary.insert_rows(
        relation,
        rows,
        should_commit=false,
        chunk_size=elementary.get_config_var("dbt_artifacts_chunk_size"),
    ) %}
{% endmacro %}

{# Spark - truncate and insert (non-atomic) #}
{% macro spark__replace_table_data(relation, rows) %}
    {% call statement("truncate_relation") -%}
        delete from {{ relation }} where 1 = 1
    {%- endcall %}
    {% do elementary.insert_rows(
        relation,
        rows,
        should_commit=false,
        chunk_size=elementary.get_config_var("dbt_artifacts_chunk_size"),
    ) %}
{% endmacro %}

{# FabricSpark - delegate to Spark (non-atomic) #}
{% macro fabricspark__replace_table_data(relation, rows) %}
    {{ return(elementary.spark__replace_table_data(relation, rows)) }}
{% endmacro %}

{# Dremio - truncate and insert (non-atomic) #}
{% macro dremio__replace_table_data(relation, rows) %}
    {% do dbt.truncate_relation(relation) %}
    {% do elementary.insert_rows(
        relation,
        rows,
        should_commit=false,
        chunk_size=elementary.get_config_var("dbt_artifacts_chunk_size"),
    ) %}
{% endmacro %}

{# In Postgres / Redshift we do not want to replace the table, because that will cause views without
   late binding to be deleted. So instead we atomically replace the data in a transaction #}
{% macro postgres__replace_table_data(relation, rows) %}
    {% set intermediate_relation = elementary.create_intermediate_relation(
        relation, rows, temporary=True
    ) %}

    {% set query %}
        begin transaction;
        delete from {{ relation }};   -- truncate supported in Redshift transactions, but causes an immediate commit
        insert into {{ relation }} select * from {{ intermediate_relation }};
        commit;
    {% endset %}
    {% do elementary.run_query(query) %}

    {% do adapter.drop_relation(intermediate_relation) %}
{% endmacro %}

{% macro athena__replace_table_data(relation, rows) %}
    {% call statement("truncate_relation") -%} delete from {{ relation }} {%- endcall %}
    {% do elementary.insert_rows(
        relation,
        rows,
        should_commit=false,
        chunk_size=elementary.get_config_var("dbt_artifacts_chunk_size"),
    ) %}
{% endmacro %}

{% macro trino__replace_table_data(relation, rows) %}
    {% set intermediate_relation = elementary.create_intermediate_relation(
        relation, rows, temporary=True
    ) %}
    {% do elementary.run_query(
        adapter.dispatch("create_table_as")(
            False,
            relation,
            "select * from {}".format(intermediate_relation),
            replace=true,
        )
    ) %}
    {% do adapter.drop_relation(intermediate_relation) %}
{% endmacro %}

{# DuckDB - truncate and insert with commit to survive ROLLBACK on in-memory databases #}
{% macro duckdb__replace_table_data(relation, rows) %}
    {% do dbt.truncate_relation(relation) %}
    {% do elementary.insert_rows(
        relation,
        rows,
        should_commit=true,
        chunk_size=elementary.get_config_var("dbt_artifacts_chunk_size"),
    ) %}
{% endmacro %}