implement selecting subset of tables (#77)

Author: mateuszklimek

dbt_project.yml

@@ -17,6 +17,7 @@ vars:
   re_data:time_window_end: '{{ run_started_at.strftime("%Y-%m-%d 00:00:00") }}'
   re_data:time_window_start: '{{ (run_started_at - modules.datetime.timedelta(1)).strftime("%Y-%m-%d 00:00:00") }}'
   re_data:anomaly_detection_look_back_days: 30
+  re_data:select: null
 
   re_data:alerting_z_score: 3
 

integration_tests/dbt_project.yml

@@ -23,6 +23,13 @@ vars:
     threshold: 0.6
   re_data:max_columns_in_query: 1
 
+  re_data:select:
+    - buy_events
+    - sample_table
+    - sample_without_time_filter
+    - sample_with_anomaly
+    - re_data_source_test_table
+
   re_data:metrics_base:
     table:
       - row_count

macros/meta/get_monitored.sql

@@ -5,9 +5,27 @@
     {% do both.extend(graph.sources.values()) %}
     {% set owners_config = re_data.get_owners_config() %}
 
+    {% set select_var = var('re_data:select') %}
+    {% set select_all = true %}
+
+    {% set selected_nodes = none %}
+    {% if select_var is not none %}
+        {% set select_all = false %}
+        {% set selected_nodes = dict() %}
+        {% for el in select_var %}
+            {% do selected_nodes.update({el: True}) %}
+        {% endfor %}
+    {% endif %}
+
     {% for el in both %}
         {% if el.resource_type in ['model', 'seed', 'source'] %}
             {% if el.config.get('re_data_monitored') %}
+                {% if select_all %}
+                    {% set selected = true %}
+                {% else %}
+                    {% set selected = selected_nodes.get(el.name, false) %}
+                {% endif %}
+
                 {% do monitored.append({
                     'name': re_data.name_in_db(el.identifier or el.alias or el.name),
                     'schema': re_data.name_in_db(el.schema),
@@ -17,6 +35,7 @@
                     'columns': re_data.columns_in_db(el.config.get('re_data_columns', [])),
                     'anomaly_detector': el.config.get('re_data_anomaly_detector', var('re_data:anomaly_detector', {})),
                     'owners': re_data.prepare_model_owners(el.config.get('re_data_owners', []), owners_config),
+                    'selected': selected
                     })
                 %}
             {% endif %}

macros/meta/monitored_model_queries.sql

@@ -1,10 +1,10 @@
 {% macro get_tables() %}
     select *
-    from {{ ref('re_data_monitored') }}
+    from {{ ref('re_data_selected') }}
     order by name, schema, database, time_filter
 {% endmacro %}
 
 {% macro get_schemas() %}
     select distinct schema, database
-    from {{ ref('re_data_monitored') }}
+    from {{ ref('re_data_selected') }}
 {% endmacro %}

macros/metrics/base/internal_model_template.sql

@@ -14,7 +14,7 @@
 
 {% macro re_data_last_base_metrics_thread(num) %}
     {% set part_name = 're_data_last_base_metrics_part' ~ num %}
-    {{ re_data.generate_depends(['re_data_monitored', 're_data_columns', 're_data_run_started_at', part_name]) }}
+    {{ re_data.generate_depends(['re_data_selected', 're_data_monitored', 're_data_columns', 're_data_run_started_at', part_name]) }}
 
     {{
         config(

macros/post_hook/re_data_monitored.sql

@@ -3,7 +3,7 @@
     {% do re_data.insert_list_to_table(
         this,
         monitored,
-        ['name', 'schema', 'database', 'time_filter', 'metrics', 'columns', 'anomaly_detector', 'owners']
+        ['name', 'schema', 'database', 'time_filter', 'metrics', 'columns', 'anomaly_detector', 'owners', 'selected']
     ) %}
 
     {{ return('') }}

macros/public/store/export_alerts.sql

@@ -27,7 +27,7 @@
             columns as {{ re_data.quote_column('columns') }},
             anomaly_detector as {{ re_data.quote_column('anomaly_detector') }},
             owners as {{ re_data.quote_column('owners') }}
-        from {{ ref('re_data_monitored') }}
+        from {{ ref('re_data_selected') }}
     {% endset %}
     {% set query_result = run_query(monitored_query) %}
     {% set monitored_file_path = monitored_path or 'target/re_data/monitored.json' %}

macros/samples/internal_model_template.sql

@@ -6,7 +6,7 @@
 
 
 {% macro re_data_last_table_samples() %}
-    {{ re_data.generate_depends(['re_data_monitored', 're_data_columns', 're_data_run_started_at', 're_data_last_table_samples_part']) }}
+    {{ re_data.generate_depends(['re_data_selected', 're_data_monitored', 're_data_columns', 're_data_run_started_at', 're_data_last_table_samples_part']) }}
 
     {{
         config(

models/alerts/re_data_anomalies.sql

@@ -29,7 +29,7 @@ select
     {{ re_data.generate_metric_value_text('z.metric', 'z.last_value') }} as last_value_text
 from
     {{ ref('re_data_z_score')}} z 
-left join {{ ref('re_data_monitored') }} m 
+left join {{ ref('re_data_selected') }} m 
 on {{ split_and_return_nth_value('table_name', '.', 1) }} = m.database
 and {{ split_and_return_nth_value('table_name', '.', 2) }} = m.schema
 and {{ split_and_return_nth_value('table_name', '.', 3) }} = m.name

models/alerts/re_data_schema_changes.sql

@@ -9,6 +9,7 @@
 -- depends_on: {{ ref('re_data_run_started_at') }}
 -- depends_on: {{ ref('re_data_columns_over_time') }}
 -- depends_on: {{ ref('re_data_monitored') }}
+-- depends_on: {{ ref('re_data_selected') }}
 
 {% if execute and not re_data.in_compile() %}
     {% set last_data_points %} 
@@ -54,7 +55,7 @@
         select * from {{ ref('re_data_columns_over_time')}}
         where detected_time = '{{ most_recent_time }}'
         and table_name in (
-            select {{ full_table_name('name', 'schema', 'database') }} from {{ ref('re_data_monitored')}}
+            select {{ full_table_name('name', 'schema', 'database') }} from {{ ref('re_data_selected')}}
         )
     ),
 
@@ -63,7 +64,7 @@
         select * from {{ ref('re_data_columns_over_time')}}
         where detected_time = '{{ prev_most_recent}}'
         and table_name in (
-            select {{ full_table_name('name', 'schema', 'database') }} from {{ ref('re_data_monitored')}}
+            select {{ full_table_name('name', 'schema', 'database') }} from {{ ref('re_data_selected')}}
         )
     ),