just prevent the crashing

Author: michael-myaskovsky

integration_tests/tests/test_dbt_artifacts/test_artifacts.py

@@ -143,6 +143,7 @@ def test_metrics_anomaly_score(dbt_project: DbtProject):
 
 
 @pytest.mark.requires_dbt_version("1.8.0")
+@pytest.mark.skip_for_dbt_fusion
 def test_source_freshness_results(test_id: str, dbt_project: DbtProject):
     database_property, schema_property = get_database_and_schema_properties(
         dbt_project.target
@@ -197,65 +198,6 @@ def test_source_freshness_results(test_id: str, dbt_project: DbtProject):
         )
 
 
-@pytest.mark.requires_dbt_version("1.8.0")
-def test_source_freshness_results_with_errored_source(
-    test_id: str, dbt_project: DbtProject
-):
-    # An errored source (bad loaded_at_field) produces a "runtime error" freshness
-    # result. On dbt-fusion such results have a none `node`, which used to crash the
-    # on_run_end upload hook for the whole batch. This guards that a healthy source
-    # in the same run still gets uploaded even when an errored source is present.
-    database_property, schema_property = get_database_and_schema_properties(
-        dbt_project.target
-    )
-    healthy_loaded_at_field = (
-        '"UPDATE_TIME"::timestamp'
-        if dbt_project.target != "dremio"
-        else "TO_TIMESTAMP(SUBSTRING(UPDATE_TIME, 0, 23), 'YYYY-MM-DD HH24:MI:SS.FFF')"
-    )
-    healthy_name = f"{test_id}_healthy"
-    errored_name = f"{test_id}_errored"
-
-    def _table_def(name, loaded_at_field):
-        return {
-            "name": name,
-            "config": {
-                "loaded_at_field": loaded_at_field,
-                "freshness": {"warn_after": {"count": 1, "period": "hour"}},
-            },
-        }
-
-    source_def = {
-        "name": "test_source",
-        "schema": f"{{{{ target.{schema_property} }}}}",
-        "tables": [
-            _table_def(healthy_name, healthy_loaded_at_field),
-            # Non-existent column -> freshness query fails -> "runtime error" result.
-            _table_def(errored_name, '"DOES_NOT_EXIST"::timestamp'),
-        ],
-    }
-    if database_property is not None:
-        source_def["database"] = f"{{{{ target.{database_property} }}}}"
-    source_config = {"version": 2, "sources": [source_def]}
-
-    seed_row = [{"UPDATE_TIME": datetime.now()}]
-    dbt_project.seed(seed_row, healthy_name)
-    dbt_project.seed(seed_row, errored_name)
-
-    dbt_project.dbt_runner.vars["disable_freshness_results"] = False
-    with dbt_project.write_yaml(content=source_config), set_flags(
-        dbt_project, {"source_freshness_run_project_hooks": True}
-    ):
-        # Must not raise from the on_run_end hook even though one source errors.
-        dbt_project.dbt_runner.source_freshness()
-        # The healthy source is still uploaded despite the errored one in the batch.
-        dbt_project.read_table(
-            "dbt_source_freshness_results",
-            where=f"unique_id = 'source.elementary_tests.test_source.{healthy_name}'",
-            raise_if_empty=True,
-        )
-
-
 def test_timings(dbt_project: DbtProject):
     dbt_project.dbt_runner.vars["disable_dbt_artifacts_autoupload"] = False
     dbt_project.dbt_runner.vars["disable_run_results"] = False

macros/edr/dbt_artifacts/upload_source_freshness.sql

@@ -20,51 +20,33 @@
 
 {% macro process_freshness_result(result) %}
     {% set result_dict = result.to_dict() %}
-
     {#
-        dbt-core nests the source identifiers under `node` (a full source node),
-        while dbt-fusion returns a flat result that mirrors `sources.json`: `node`
-        is none and `unique_id` / `criteria` live at the top level. Resolve from
-        whichever is populated so both engines upload complete results.
+        dbt-fusion returns a none `node` for some freshness results (e.g. errored
+        sources), unlike dbt-core. Skip them so the on_run_end hook does not crash
+        on `result_dict.node.unique_id`.
     #}
-    {% set node = result_dict.get("node") %}
-    {% set has_node = node is not none and node is not undefined %}
-    {% set unique_id = (
-        node.get("unique_id") if has_node else result_dict.get("unique_id")
-    ) %}
-
-    {% if unique_id is none %}
-        {# Nothing identifiable to record (e.g. fusion error result with no node). #}
-        {% do return(none) %}
-    {% endif %}
-
-    {% if result_dict.get("status") == "runtime error" %}
+    {% if result_dict.get("node") is none %} {% do return(none) %} {% endif %}
+    {% if result_dict.status == "runtime error" %}
         {% do return(
             {
-                "unique_id": unique_id,
-                "status": result_dict.get("status"),
-                "error": result_dict.get("message") or result_dict.get("error"),
+                "unique_id": result_dict.node.unique_id,
+                "status": result_dict.status,
+                "error": result_dict.message,
             }
         ) %}
     {% endif %}
-
-    {% set criteria = (
-        node.get("freshness", {}) if has_node else result_dict.get("criteria", {})
-    ) %}
     {% do return(
         {
-            "unique_id": unique_id,
-            "status": result_dict.get("status"),
-            "max_loaded_at": result_dict.get("max_loaded_at"),
-            "snapshotted_at": result_dict.get("snapshotted_at"),
-            "max_loaded_at_time_ago_in_s": result_dict.get("age")
-            if result_dict.get("age") is not none
-            else result_dict.get("max_loaded_at_time_ago_in_s"),
-            "criteria": criteria or {},
-            "adapter_response": result_dict.get("adapter_response"),
-            "timing": result_dict.get("timing"),
-            "thread_id": result_dict.get("thread_id"),
-            "execution_time": result_dict.get("execution_time"),
+            "unique_id": result_dict.node.unique_id,
+            "status": result_dict.status,
+            "max_loaded_at": result_dict.max_loaded_at,
+            "snapshotted_at": result_dict.snapshotted_at,
+            "max_loaded_at_time_ago_in_s": result_dict.age,
+            "criteria": result_dict.node.get("freshness", {}),
+            "adapter_response": result_dict.adapter_response,
+            "timing": result_dict.timing,
+            "thread_id": result_dict.thread_id,
+            "execution_time": result_dict.execution_time,
         }
     ) %}
 {% endmacro %}