WIP - Glue integration

macros/commands/generate_schema_baseline_test.sql

@@ -1,26 +1,26 @@
-{% macro generate_schema_baseline_test(name=none, include_sources=True, include_models=False, fail_on_added=False, enforce_types=False) %}
+{% macro generate_schema_baseline_test(name=none, include_sources=True, include_models=False, fail_on_added=False, enforce_types=False, convert_to_lower=False) %}
   {% if name %}
-    {{ generate_schema_baseline_test_for_node(name, fail_on_added=fail_on_added, enforce_types=enforce_types) }}
+    {{ generate_schema_baseline_test_for_node(name, fail_on_added=fail_on_added, enforce_types=enforce_types, convert_to_lower=convert_to_lower) }}
   {% else %}
     {{ generate_schema_baseline_test_for_all_nodes(include_sources=include_sources, include_models=include_models,
-                                                   fail_on_added=fail_on_added, enforce_types=enforce_types) }}
+                                                   fail_on_added=fail_on_added, enforce_types=enforce_types, convert_to_lower=convert_to_lower) }}
   {% endif %}
 {% endmacro %}
 
-{% macro generate_schema_baseline_test_for_all_nodes(include_sources=True, include_models=False, fail_on_added=False, enforce_types=False) %}
+{% macro generate_schema_baseline_test_for_all_nodes(include_sources=True, include_models=False, fail_on_added=False, enforce_types=False, convert_to_lower=False) %}
   {% set nodes = elementary.get_nodes_from_graph() %}
   {% for node in nodes %}
     {% if node.package_name != 'elementary' and
           ((include_sources and node.resource_type == 'source') or
            (include_models and node.resource_type == 'model')) %}
       {% do print("Generating schema changes from baseline test for {} '{}':".format(node.resource_type, node.name)) %}
-      {{ generate_schema_baseline_test_for_node(node, fail_on_added=fail_on_added, enforce_types=enforce_types) }}
+      {{ generate_schema_baseline_test_for_node(node, fail_on_added=fail_on_added, enforce_types=enforce_types, convert_to_lower=convert_to_lower) }}
       {% do print('----------------------------------') %}
     {% endif %}
   {% endfor %}
 {% endmacro %}
 
-{% macro generate_schema_baseline_test_for_node(node, fail_on_added=False, enforce_types=False) %}
+{% macro generate_schema_baseline_test_for_node(node, fail_on_added=False, enforce_types=False, convert_to_lower=False) %}
   {% if node is string %}
     {% set node_name = node %}
     {% set node = elementary.get_node_by_name(node_name) %}
@@ -52,13 +52,52 @@
     {% do test_params.update({"enforce_types": "true"}) %}
   {% endif %}
 
-  {# Common yaml for sources and models #}
-  {% set common_yaml %}
+  {# Full yaml for sources and models #}
+  {% set full_yaml %}
+  {%- if node.resource_type == 'source' %}
+    {{generate_schema_baseline_test_for_source(node, columns, test_params, convert_to_lower)}}
+  {% else %}
+    {{generate_schema_baseline_test_for_model(node, columns, test_params, convert_to_lower)}}
+  {% endif -%}
+  {% endset %}
+
+  {% do print(full_yaml) %}
+{% endmacro %}
+
+{% macro generate_schema_baseline_test_for_source(node, columns, test_params, convert_to_lower) %}
+sources:
+  - name: {{ node.source_name }}
+    tables:
+      - name: {{ node.name }}
+        columns:
+        {%- for column in columns %}
+          - name: {{ column.name }}
+            {%- if convert_to_lower %}
+            data_type: {{ column.dtype|lower }}
+            {% else %}
+            data_type: {{ column.dtype }}
+            {% endif -%}
+        {% endfor %}
+        tests:
+          - elementary.schema_changes_from_baseline
+          {%- if test_params %}:
+            {%- for param, param_val in test_params.items() %}
+              {{param}}: {{param_val}}
+            {%- endfor -%}
+          {% endif -%}  
+{% endmacro %}
+
+{% macro generate_schema_baseline_test_for_model(node, columns, test_params, convert_to_lower) %}
+models:
   - name: {{ node.name }}
     columns:
     {%- for column in columns %}
       - name: {{ column.name }}
+        {%- if convert_to_lower %}
+        data_type: {{ column.dtype|lower }}
+        {% else %}
         data_type: {{ column.dtype }}
+        {% endif -%}
     {% endfor %}
     tests:
       - elementary.schema_changes_from_baseline
@@ -67,19 +106,4 @@
           {{param}}: {{param_val}}
         {%- endfor -%}
       {% endif -%}
-  {% endset %}
-
-  {% set full_yaml %}
-  {%- if node.resource_type == 'source' %}
-sources:
-  - name: {{ node.source_name }}
-    tables:
-      {{- common_yaml }}
-  {% else %}
-models:
-  {{- common_yaml }}
-  {% endif -%}
-  {% endset %}
-
-  {% do print(full_yaml) %}
 {% endmacro %}

macros/edr/materializations/model/incremental.sql

@@ -61,3 +61,14 @@
   {% do elementary.cache_metrics(metrics) %}
   {% do return(relations) %}
 {% endmaterialization %}
+
+{% materialization incremental, adapter="glue", supported_languages=["sql", "python"] %}
+  {% set relations = dbt.materialization_incremental_glue.call_macro() %}
+  {% if not elementary.is_elementary_enabled() %}
+    {% do return(relations) %}
+  {% endif %}
+
+  {% set metrics = elementary.query_metrics() %}
+  {% do elementary.cache_metrics(metrics) %}
+  {% do return(relations) %}
+{% endmaterialization %}

macros/edr/materializations/model/metrics.sql

@@ -1,7 +1,7 @@
 {% macro query_table_metrics() %}
   {% set query %}
     select
-      {{ modules.datetime.datetime.utcnow().timestamp() }} as build_timestamp,
+      cast({{ modules.datetime.datetime.utcnow().timestamp() }} as float) as build_timestamp,
       count(*) as row_count
     from {{ this }}
   {% endset %}

macros/edr/materializations/model/table.sql

@@ -61,3 +61,14 @@
   {% do elementary.cache_metrics(metrics) %}
   {% do return(relations) %}
 {% endmaterialization %}
+
+{% materialization table, adapter="glue" %}
+  {% set relations = dbt.materialization_table_glue() %}
+  {% if not elementary.is_elementary_enabled() %}
+    {% do return(relations) %}
+  {% endif %}
+
+  {% set metrics = elementary.query_metrics() %}
+  {% do elementary.cache_metrics(metrics) %}
+  {% do return(relations) %}
+{% endmaterialization %}

macros/utils/cross_db_utils/quote_column.sql

@@ -11,6 +11,8 @@
 {%- macro edr_quote_column(column_name) -%}
     {% if adapter.quote(column_name[1:-1]) == column_name %}
         {{ return(column_name) }}
+    {% elif target.type == 'glue' %}
+        {{ return(column_name) }}
     {% else %}
         {% set quoted_column = adapter.quote(column_name) %}
         {{ return(quoted_column) }}

macros/utils/table_operations/create_table_like.sql

@@ -1,4 +1,8 @@
 {% macro create_table_like(relation, like_relation, temporary=False, like_columns=none) %}
+    {{ return(adapter.dispatch('create_table_like', 'elementary')(relation, like_relation, temporary, like_columns)) }}
+{% endmacro %}
+
+{% macro default__create_table_like(relation, like_relation, temporary=False, like_columns=none) %}
     {% set empty_table_query %}
         SELECT
         {% if like_columns %}
@@ -13,3 +17,19 @@
     {% endset %}
     {% do elementary.run_query(dbt.create_table_as(temporary, relation, empty_table_query)) %}
 {% endmacro %}
+
+{% macro glue__create_table_like(relation, like_relation, temporary=False, like_columns=none) %}
+    {% set empty_table_query %}
+        SELECT
+        {% if like_columns %}
+            {% for column in like_columns %}
+                {{ column }}{{ ", " if not loop.last }}
+            {% endfor %}
+        {% else %}
+            *
+        {% endif %}
+        FROM {{ like_relation }}
+        WHERE 1 = 0
+    {% endset %}
+    {% do dbt.glue_exec_query(dbt.create_table_as(temporary, relation, empty_table_query)) %}
+{% endmacro %}

macros/utils/table_operations/insert_rows.sql

@@ -22,13 +22,21 @@
       {% set queries_len = insert_rows_queries | length %}
       {% for insert_query in insert_rows_queries %}
         {% do elementary.file_log("[{}/{}] Running insert query.".format(loop.index, queries_len)) %}
-        {% do elementary.run_query(insert_query) %}
+        {% if target.type == 'glue' %}
+          {% do dbt.glue_exec_query(insert_query) %}
+        {% else %}
+          {% do elementary.run_query(insert_query) %}
+        {% endif %}
       {% endfor %}
     {% elif insert_rows_method == 'chunk' %}
       {% set rows_chunks = elementary.split_list_to_chunks(rows, chunk_size) %}
       {% for rows_chunk in rows_chunks %}
         {% set insert_rows_query = elementary.get_chunk_insert_query(table_relation, columns, rows_chunk) %}
-        {% do elementary.run_query(insert_rows_query) %}
+        {% if target.type == 'glue' %}
+          {% do dbt.glue_exec_query(insert_rows_query) %}
+        {% else %}
+          {% do elementary.run_query(insert_rows_query) %}
+        {% endif %}
       {% endfor %}
     {% else %}
       {% do exceptions.raise_compiler_error("Specified invalid value for 'insert_rows_method' var.") %}
@@ -57,7 +65,7 @@
           {% do rendered_column_values.append(column_value) %}
         {% else %}
           {% set column_value = elementary.insensitive_get_dict_value(row, column.name) %}
-          {% do rendered_column_values.append(elementary.render_value(column_value)) %}
+          {% do rendered_column_values.append(elementary.render_value(column_value, column.data_type)) %}
         {% endif %}
       {% endfor %}
       {% set row_sql = "({})".format(rendered_column_values | join(",")) %}
@@ -92,7 +100,7 @@
             {% for row in rows -%}
                 ({%- for column in columns -%}
                     {%- set column_value = elementary.insensitive_get_dict_value(row, column.name, none) -%}
-                    {{ elementary.render_value(column_value) }}
+                    {{ elementary.render_value(column_value, column.data_type) }}
                     {{- "," if not loop.last else "" -}}
                  {%- endfor -%}) {{- "," if not loop.last else "" -}}
             {%- endfor -%}
@@ -116,10 +124,16 @@
     {{- return(string_value | replace("'", "''")) -}}
 {%- endmacro -%}
 
-{%- macro render_value(value) -%}
+{%- macro glue__escape_special_chars(string_value) -%}
+    {{- return(string_value | replace("'", '"')) -}}
+{%- endmacro -%}
+
+{%- macro render_value(value, data_type=none) -%}
     {%- if value is defined and value is not none -%}
         {%- if value is number -%}
             {{- value -}}
+        {%- elif target.type == 'glue' and data_type=='timestamp' -%}
+            cast('{{- value -}}' as timestamp)
         {%- elif value is string -%}
             '{{- elementary.escape_special_chars(value) -}}'
         {%- elif value is mapping or value is sequence -%}

macros/utils/table_operations/make_temp_relation.sql

@@ -53,3 +53,12 @@
     {% do return(tmp_relation) %}
 {% endmacro %}
 
+{% macro glue__make_temp_table_relation(base_relation, suffix) %}
+    {% set tmp_identifier = elementary.table_name_with_suffix(base_relation.identifier, suffix) %}
+    {% set tmp_relation = api.Relation.create(
+        identifier=tmp_identifier,
+        schema=base_relation.schema,
+        database=base_relation.database,
+        type='table') %}
+    {% do return(tmp_relation) %}
+{% endmacro %}

macros/utils/table_operations/replace_table_data.sql

@@ -38,3 +38,15 @@
 
     {% do adapter.drop_relation(intermediate_relation) %}
 {% endmacro %}
+
+{# Glue - truncate and insert (non-atomic) #}
+{% macro glue__replace_table_data(relation, rows) %}
+    {% set intermediate_relation = elementary.create_intermediate_relation(relation, rows, temporary=True) %}
+    {% do dbt.glue_exec_query(dbt.get_insert_overwrite_sql(intermediate_relation, relation)) %}
+
+    {% set query %}
+        drop table if exists {{ intermediate_relation }}
+    {% endset %}
+
+    {% do dbt.glue_exec_query(query) %}
+{% endmacro %}

models/edr/data_monitoring/data_monitoring/data_monitoring_metrics.sql

@@ -1,6 +1,7 @@
 {{
   config(
     materialized='incremental',
+    incremental_strategy='insert_overwrite',
     unique_key='id',
     on_schema_change='append_new_columns',
     full_refresh=elementary.get_config_var('elementary_full_refresh'),

models/edr/data_monitoring/schema_changes/schema_columns_snapshot.sql

@@ -1,6 +1,7 @@
 {{
   config(
     materialized='incremental',
+    incremental_strategy='insert_overwrite',
     unique_key = 'column_state_id',
     on_schema_change = 'append_new_columns',
     full_refresh=elementary.get_config_var('elementary_full_refresh'),

models/edr/dbt_artifacts/dbt_exposures.sql

@@ -1,6 +1,7 @@
 {{
   config(
     materialized='incremental',
+    incremental_strategy='insert_overwrite',
     transient=False,
     post_hook='{{ elementary.upload_dbt_exposures() }}',
     unique_key='unique_id',

models/edr/dbt_artifacts/dbt_invocations.sql

@@ -1,6 +1,7 @@
 {{
   config(
     materialized = 'incremental',
+    incremental_strategy='insert_overwrite',
     transient=False,
     unique_key = 'invocation_id',
     on_schema_change = 'append_new_columns',

models/edr/dbt_artifacts/dbt_metrics.sql

@@ -1,6 +1,7 @@
 {{
   config(
     materialized='incremental',
+    incremental_strategy='insert_overwrite',
     transient=False,
     post_hook='{{ elementary.upload_dbt_metrics() }}',
     unique_key='unique_id',

models/edr/dbt_artifacts/dbt_models.sql

@@ -1,6 +1,7 @@
 {{
   config(
     materialized='incremental',
+    incremental_strategy='insert_overwrite',
     transient=False,
     post_hook='{{ elementary.upload_dbt_models() }}',
     unique_key='unique_id',

models/edr/dbt_artifacts/dbt_run_results.sql

@@ -1,6 +1,7 @@
 {{
   config(
     materialized = 'incremental',
+    incremental_strategy='insert_overwrite',
     transient=False,
     unique_key = 'model_execution_id',
     on_schema_change = 'append_new_columns',

models/edr/dbt_artifacts/dbt_seeds.sql

@@ -1,6 +1,7 @@
 {{
   config(
     materialized='incremental',
+    incremental_strategy='insert_overwrite',
     transient=False,
     post_hook='{{ elementary.upload_dbt_seeds() }}',
     unique_key='unique_id',

models/edr/dbt_artifacts/dbt_snapshots.sql

@@ -1,6 +1,7 @@
 {{
   config(
     materialized='incremental',
+    incremental_strategy='insert_overwrite',
     transient=False,
     post_hook='{{ elementary.upload_dbt_snapshots() }}',
     unique_key='unique_id',

models/edr/dbt_artifacts/dbt_sources.sql

@@ -1,6 +1,7 @@
 {{
   config(
     materialized='incremental',
+    incremental_strategy='insert_overwrite',
     transient=False,
     post_hook='{{ elementary.upload_dbt_sources() }}',
     unique_key='unique_id',

models/edr/dbt_artifacts/dbt_tests.sql

@@ -1,6 +1,7 @@
 {{
   config(
     materialized='incremental',
+    incremental_strategy='insert_overwrite',
     transient=False,
     post_hook='{{ elementary.upload_dbt_tests() }}',
     unique_key='unique_id',

models/edr/run_results/dbt_source_freshness_results.sql

@@ -1,6 +1,7 @@
 {{
   config(
     materialized = 'incremental',
+    incremental_strategy='insert_overwrite',
     unique_key = 'source_freshness_execution_id',
     on_schema_change = 'append_new_columns',
     full_refresh=elementary.get_config_var('elementary_full_refresh'),

models/edr/run_results/elementary_test_results.sql

@@ -1,6 +1,7 @@
 {{
   config(
     materialized = 'incremental',
+    incremental_strategy='insert_overwrite',
     unique_key = 'id',
     on_schema_change = 'append_new_columns',
     full_refresh=elementary.get_config_var('elementary_full_refresh'),

models/edr/run_results/test_result_rows.sql

@@ -1,6 +1,7 @@
 {{
   config(
     materialized = 'incremental',
+    incremental_strategy='insert_overwrite',
     unique_key = 'elementary_test_results_id',
     on_schema_change = 'append_new_columns',
     full_refresh=elementary.get_config_var('elementary_full_refresh'),