dbt-fusion fixes

Author: haritamar

macros/edr/dbt_artifacts/upload_dbt_artifacts.sql

@@ -51,12 +51,15 @@
     order by metadata_hash
     {% endset %}
     {% set artifacts_hashes_results = elementary.run_query(stored_artifacts_query) %}
-    {% set artifact_agate_hashes = artifacts_hashes_results.group_by("artifacts_model") %}
-    {% set artifacts_hashes = {} %}
-    {% for artifacts_model, metadata_hashes in artifact_agate_hashes.items() %}
-        {% do artifacts_hashes.update({artifacts_model: metadata_hashes.columns["metadata_hash"]}) %}
+    {% set artifacts_hashes = elementary.agate_to_dicts(artifacts_hashes_results) %}
+
+    {% set artifacts_hashes_per_model = {} %}
+    {% for artifacts_hashes_row in artifacts_hashes %}
+        {% do artifacts_hashes_per_model.setdefault(artifacts_hashes_row['artifacts_model'], []) %}
+        {% do artifacts_hashes_per_model[artifacts_hashes_row['artifacts_model']].append(artifacts_hashes_row['metadata_hash']) %}
     {% endfor %}
-    {% do return(artifacts_hashes) %}
+    
+    {% do return(artifacts_hashes_per_model) %}
 {% endmacro %}
 
 {% macro get_artifacts_hashes_for_model(model_name) %}

macros/edr/dbt_artifacts/upload_dbt_models.sql

@@ -68,7 +68,7 @@
     {% set flatten_model_metadata_dict = {
         'unique_id': node_dict.get('unique_id'),
         'alias': node_dict.get('alias'),
-        'checksum': checksum_dict.get('checksum'),
+        'checksum': checksum_dict.get('checksum') if checksum_dict is mapping else none,
         'materialization': config_dict.get('materialized'),
         'tags': elementary.filter_none_and_sort(tags),
         'meta': meta_dict,

macros/edr/dbt_artifacts/upload_dbt_tests.sql

@@ -188,7 +188,7 @@
     {% set test_type = 'generic' %}
     {%- if test_namespace == 'dbt_expectations' -%}
         {% set test_type = 'expectation' %}
-    {%- elif 'tests/generic' in test_path or 'macros/' in test_path -%}
+    {%- elif 'tests/generic' in test_path or 'macros/' in test_path or 'generic_tests/' in test_path -%}
         {% set test_type = 'generic' %}
     {%- elif 'tests/' in test_path -%}
         {% set test_type = 'singular' %}

macros/edr/tests/on_run_end/handle_tests_results.sql

@@ -28,34 +28,38 @@
 {% macro get_result_enriched_elementary_test_results(cached_elementary_test_results, cached_elementary_test_failed_row_counts, render_result_rows=false) %}
   {% set elementary_test_results = [] %}
 
-  {% for result in results | selectattr('node.resource_type', '==', 'test') %}
-    {% set result = elementary.get_run_result_dict(result) %}
-    {% set elementary_test_results_rows = cached_elementary_test_results.get(result.node.unique_id) %}
-    {% set elementary_test_failed_row_count = cached_elementary_test_failed_row_counts.get(result.node.unique_id) %}
-
-    {# Materializing the test failed and therefore was not added to the cache. #}
-    {% if not elementary_test_results_rows %}
-      {% set flattened_test = elementary.flatten_test(result.node) %}
-      {% set elementary_test_results_rows = [elementary.get_dbt_test_result_row(flattened_test)] %}
-    {% endif %}
-
-    {% for elementary_test_results_row in elementary_test_results_rows %}
-      {% set failures = elementary_test_results_row.get("failures", result.failures) %}
-
-      {# For Elementary anomaly tests, we actually save more than one result per test, in that case the dbt status will be "fail"
-         even if one such result failed and the rest succeeded. To handle this, we make sure to mark the status as "pass" for these 
-         results if the number of failed rows is 0.
-         We don't want to do this for every test though - because otherwise it can break configurations like warn_if=0 #}
-      {% set status = "pass" if failures == 0 and elementary_test_results_row.get("test_type") == "anomaly_detection" else result.status %}
-
-      {% do elementary_test_results_row.update({'status': status, 'failures': failures, 'invocation_id': invocation_id, 
-                                                'failed_row_count': elementary_test_failed_row_count}) %}
-      {% do elementary_test_results_row.setdefault('test_results_description', result.message) %}
-      {% if render_result_rows %}
-        {% do elementary_test_results_row.update({"result_rows": elementary.render_result_rows(elementary_test_results_row.result_rows)}) %}
+  {% for result in results %}
+    {% set node = elementary.get_node(result.unique_id) %}
+    {% if node.resource_type == 'test' %}
+      {% do print('result is' ~ result | tojson) %}
+      {% set result = elementary.get_run_result_dict(result) %}
+      {% set elementary_test_results_rows = cached_elementary_test_results.get(result.node.unique_id) %}
+      {% set elementary_test_failed_row_count = cached_elementary_test_failed_row_counts.get(result.node.unique_id) %}
+
+      {# Materializing the test failed and therefore was not added to the cache. #}
+      {% if not elementary_test_results_rows %}
+        {% set flattened_test = elementary.flatten_test(node) %}
+        {% set elementary_test_results_rows = [elementary.get_dbt_test_result_row(flattened_test)] %}
       {% endif %}
-      {% do elementary_test_results.append(elementary_test_results_row) %}
-    {% endfor %}
+
+      {% for elementary_test_results_row in elementary_test_results_rows %}
+        {% set failures = elementary_test_results_row.get("failures", result.failures) %}
+
+        {# For Elementary anomaly tests, we actually save more than one result per test, in that case the dbt status will be "fail"
+          even if one such result failed and the rest succeeded. To handle this, we make sure to mark the status as "pass" for these 
+          results if the number of failed rows is 0.
+          We don't want to do this for every test though - because otherwise it can break configurations like warn_if=0 #}
+        {% set status = "pass" if failures == 0 and elementary_test_results_row.get("test_type") == "anomaly_detection" else result.status %}
+
+        {% do elementary_test_results_row.update({'status': status, 'failures': failures, 'invocation_id': invocation_id, 
+                                                  'failed_row_count': elementary_test_failed_row_count}) %}
+        {% do elementary_test_results_row.setdefault('test_results_description', result.message) %}
+        {% if render_result_rows %}
+          {% do elementary_test_results_row.update({"result_rows": elementary.render_result_rows(elementary_test_results_row.result_rows)}) %}
+        {% endif %}
+        {% do elementary_test_results.append(elementary_test_results_row) %}
+      {% endfor %}
+    {% endif %}
   {% endfor %}
 
   {% do return(elementary_test_results) %}
@@ -174,13 +178,15 @@
 {% macro pop_test_result_rows(elementary_test_results) %}
   {% set result_rows = [] %}
   {% for test_result in elementary_test_results %}
-    {% for result_row in test_result.pop('result_rows', []) %}
-      {% do result_rows.append({
-        "elementary_test_results_id": test_result.id,
-        "detected_at": test_result.detected_at,
-        "result_row": result_row
-      }) %}
-    {% endfor %}
+    {% if 'result_rows' in test_result %}
+      {% for result_row in test_result.pop('result_rows') %}
+        {% do result_rows.append({
+          "elementary_test_results_id": test_result.id,
+          "detected_at": test_result.detected_at,
+          "result_row": result_row
+        }) %}
+      {% endfor %}
+    {% endif %}
   {% endfor %}
   {% do return(result_rows) %}
 {% endmacro %}

macros/utils/graph/get_result_node.sql

@@ -1,12 +1,13 @@
 {% macro get_result_node(identifier, package_name='elementary') %}
   {% for result in results %}
-    {% if result.node.identifier == identifier %}
+    {% set node = elementary.get_node(result.unique_id) %}
+    {% if node.identifier == identifier %}
       {% if package_name %}
-        {% if result.node.package_name == package_name %}
-          {{ return(result.node) }}
+        {% if node.package_name == package_name %}
+          {{ return(node) }}
         {% endif %}
       {% else %}
-        {{ return(result.node) }}
+        {{ return(node) }}
       {% endif %}
     {% endif %}
   {% endfor %}