replace dbt.create_table_as with our own implementation

Author: haritamar

integration_tests/dbt_project/macros/create_all_types_table.sql

@@ -31,9 +31,7 @@
         CURRENT_TIME() as time_col,
         CURRENT_TIMESTAMP() as timestamp_col,
     {% endset %}
-    {% set create_table_query = dbt.create_table_as(false, relation, sql_query) %}
-    {% do elementary.edr_log(create_table_query) %}
-    {% do elementary.run_query(create_table_query) %}
+    {% do elementary.edr_create_table_as(false, relation, sql_query) %}
 {% endmacro %}
 
 {% macro snowflake__create_all_types_table() %}
@@ -81,9 +79,7 @@
         [1,2,3] as array_col,
         TO_GEOGRAPHY('POINT(-122.35 37.55)') as geography_col
     {% endset %}
-    {% set create_table_query = dbt.create_table_as(false, relation, sql_query) %}
-    {% do elementary.edr_log(create_table_query) %}
-    {% do elementary.run_query(create_table_query) %}
+    {% do elementary.edr_create_table_as(false, relation, sql_query) %}
 {% endmacro %}
 
 {% macro redshift__create_all_types_table() %}
@@ -123,9 +119,7 @@
         ST_GeogFromText('SRID=4324;POLYGON((0 0,0 1,1 1,10 10,1 0,0 0))') as geography_col,
         JSON_PARSE('{"data_type": "super"}') as super_col
     {% endset %}
-    {% set create_table_query = dbt.create_table_as(false, relation, sql_query) %}
-    {% do elementary.edr_log(create_table_query) %}
-    {% do elementary.run_query(create_table_query) %}
+    {% do elementary.edr_create_table_as(false, relation, sql_query) %}
 
 {% endmacro %}
 
@@ -184,9 +178,7 @@
         'a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11'::uuid as uuid_col,
         xmlcomment('text') as xml_col
     {% endset %}
-    {% set create_table_query = dbt.create_table_as(false, relation, sql_query) %}
-    {% do elementary.edr_log(create_table_query) %}
-    {% do elementary.run_query(create_table_query) %}
+    {% do elementary.edr_create_table_as(false, relation, sql_query) %}
 {% endmacro %}
 
 {% macro default__create_all_types_table() %}

macros/edr/tests/on_run_end/handle_tests_results.sql

@@ -115,7 +115,7 @@
     {% endset %}
 
     {{ elementary.file_log("Inserting metrics into {}.".format(target_relation)) }}
-    {%- do elementary.run_query(dbt.create_table_as(True, temp_relation, test_tables_union_query)) %}
+    {%- do elementary.edr_create_table_as(true, temp_relation, test_tables_union_query) %}
     {% do elementary.run_query(insert_query) %}
 
     {% if not elementary.has_temp_table_support() %}
@@ -163,7 +163,7 @@
     {% endset %}
 
     {{ elementary.file_log("Inserting schema columns snapshot into {}.".format(target_relation)) }}
-    {%- do elementary.run_query(dbt.create_table_as(True, temp_relation, test_tables_union_query)) %}
+    {%- do elementary.edr_create_table_as(true, temp_relation, test_tables_union_query) %}
     {% do elementary.run_query(insert_query) %}
 
     {% if not elementary.has_temp_table_support() %}

macros/utils/table_operations/create_or_replace.sql

@@ -4,39 +4,31 @@
 
 {# Snowflake and Bigquery #}
 {% macro default__create_or_replace(temporary, relation, sql_query) %}
-    {% do elementary.run_query(dbt.create_table_as(temporary, relation, sql_query)) %}
+    {% do elementary.edr_create_table_as(temporary, relation, sql_query) %}
 {% endmacro %}
 
 {% macro redshift__create_or_replace(temporary, relation, sql_query) %}
-    {% do dbt.drop_relation_if_exists(relation) %}
-    {% do elementary.run_query(dbt.create_table_as(temporary, relation, sql_query)) %}
-    {% do adapter.commit() %}
+    {% do elementary.edr_create_table_as(temporary, relation, sql_query, drop_first=true, should_commit=true) %}
 {% endmacro %}
 
 {% macro postgres__create_or_replace(temporary, relation, sql_query) %}
     {% do elementary.run_query("BEGIN") %}
-    {% do dbt.drop_relation_if_exists(relation) %}
-    {% do elementary.run_query(dbt.create_table_as(temporary, relation, sql_query)) %}
+    {% do elementary.edr_create_table_as(temporary, relation, sql_query, drop_first=true) %}
     {% do elementary.run_query("COMMIT") %}
 {% endmacro %}
 
 {% macro spark__create_or_replace(temporary, relation, sql_query) %}
-    {% do dbt.drop_relation_if_exists(relation) %}
-    {% do elementary.run_query(dbt.create_table_as(temporary, relation, sql_query)) %}
-    {% do adapter.commit() %}
+    {% do elementary.edr_create_table_as(temporary, relation, sql_query, drop_first=true, should_commit=true) %}
 {% endmacro %}
 
 {% macro athena__create_or_replace(temporary, relation, sql_query) %}
-    {% do dbt.drop_relation_if_exists(relation) %}
-    {% do elementary.run_query(dbt.create_table_as(temporary, relation, sql_query)) %}
+    {% do elementary.edr_create_table_as(temporary, relation, sql_query, drop_first=true) %}
 {% endmacro %}
 
 {% macro trino__create_or_replace(temporary, relation, sql_query) %}
-    {% do dbt.drop_relation_if_exists(relation) %}
-    {% do elementary.run_query(dbt.create_table_as(temporary, relation, sql_query)) %}
+    {% do elementary.edr_create_table_as(temporary, relation, sql_query, drop_first=true) %}
 {% endmacro %}
 
 {% macro clickhouse__create_or_replace(temporary, relation, sql_query) %}
-    {% do dbt.drop_relation_if_exists(relation) %}
-    {% do elementary.run_query(dbt.create_table_as(temporary, relation, sql_query)) %}
+    {% do elementary.edr_create_table_as(temporary, relation, sql_query, drop_first=true) %}
 {% endmacro %}

macros/utils/table_operations/create_table_as.sql

@@ -0,0 +1,21 @@
+{% macro edr_create_table_as(temporary, relation, sql_query, drop_first=false, should_commit=false) %}
+  {# This macro contains a simplified implementation that replaces our usage of 
+     dbt.create_table_as and serves our needs.
+     This version also runs the query rather than return the SQL.
+  #}
+
+  {% if drop_first %}
+    {% do dbt.drop_relation_if_exists(relation) %}
+  {% endif %}
+
+  {% set create_query %} 
+    create or replace {% if temporary %} temporary {% endif %} table {{ relation }}
+    as {{ sql_query }}
+  {% endset %}
+
+  {% do elementary.run_query(create_query) %}
+
+  {% if should_commit %}
+    {% do adapter.commit() %}    
+  {% endif %}
+{% endmacro %}

macros/utils/table_operations/create_table_like.sql

@@ -11,5 +11,5 @@
         FROM {{ like_relation }}
         WHERE 1 = 0
     {% endset %}
-    {% do elementary.run_query(dbt.create_table_as(temporary, relation, empty_table_query)) %}
+    {% do elementary.edr_create_table_as(temporary, relation, empty_table_query) %}
 {% endmacro %}

macros/utils/table_operations/create_temp_table.sql

@@ -8,12 +8,7 @@
                                                                                identifier=table_name,
                                                                                type='table') -%}
     {% set temp_table_relation = elementary.make_temp_table_relation(temp_table_relation) %}
-    {% if temp_table_exists %}
-        {% do adapter.drop_relation(temp_table_relation) %}
-        {% do elementary.run_query(dbt.create_table_as(True, temp_table_relation, sql_query)) %}
-    {% else %}
-        {% do elementary.run_query(dbt.create_table_as(True, temp_table_relation, sql_query)) %}
-    {% endif %}
+    {% do elementary.edr_create_table_as(True, temp_table_relation, sql_query, drop_first=temp_table_exists) %}
     {{ return(temp_table_relation) }}{% endmacro %}
 
 {% macro snowflake__create_temp_table(database_name, schema_name, table_name, sql_query) %}

macros/utils/table_operations/replace_table_data.sql

@@ -5,7 +5,7 @@
 {# 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.run_query(dbt.create_table_as(False, relation, 'select * from {}'.format(intermediate_relation))) %}
+    {% do elementary.edr_create_table_as(False, relation, 'select * from {}'.format(intermediate_relation)) %}
     {% do adapter.drop_relation(intermediate_relation) %}
 {% endmacro %}