From b4ed4c2dd3b955d371c5cd19c177e9beadace4a3 Mon Sep 17 00:00:00 2001 From: Sindre Grindheim Date: Mon, 9 Nov 2026 16:20:02 +0100 Subject: [PATCH 1/6] fix: store source freshness age for dbt Fusion results dbt-core exposes the time since the last load as `age` on the freshness result. dbt Fusion has no `age` and exposes it as `max_loaded_at_time_ago_in_s`, so `process_freshness_result` stored null for every Fusion row. Fall back to that field when `age` is missing. Co-Authored-By: Claude Opus 5.5 --- macros/edr/dbt_artifacts/upload_source_freshness.sql | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/macros/edr/dbt_artifacts/upload_source_freshness.sql b/macros/edr/dbt_artifacts/upload_source_freshness.sql index a2a7987e5..1e5b8ef42 100644 --- a/macros/edr/dbt_artifacts/upload_source_freshness.sql +++ b/macros/edr/dbt_artifacts/upload_source_freshness.sql @@ -35,13 +35,23 @@ } ) %} {% endif %} + {# + dbt-core exposes the time since the last load as `age`; dbt-fusion has no + `age` and exposes it as `max_loaded_at_time_ago_in_s` (the sources.json name). + #} + {% set max_loaded_at_time_ago_in_s = result_dict.get("age") %} + {% if max_loaded_at_time_ago_in_s is none %} + {% set max_loaded_at_time_ago_in_s = result_dict.get( + "max_loaded_at_time_ago_in_s" + ) %} + {% endif %} {% do return( { "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, + "max_loaded_at_time_ago_in_s": max_loaded_at_time_ago_in_s, "criteria": result_dict.node.get("freshness", {}), "adapter_response": result_dict.adapter_response, "timing": result_dict.timing, From 6f71d0441551d5fe9e4d058d1609f1adb9eb6e64 Mon Sep 17 00:00:00 2001 From: Sindre Grindheim Date: Tue, 10 Nov 2026 10:15:13 +0100 Subject: [PATCH 2/6] test: assert source freshness results store max_loaded_at_time_ago_in_s Co-Authored-By: Claude Opus 5.5 --- integration_tests/tests/test_dbt_artifacts/test_artifacts.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/integration_tests/tests/test_dbt_artifacts/test_artifacts.py b/integration_tests/tests/test_dbt_artifacts/test_artifacts.py index 4b64fc5f2..ace52b24d 100644 --- a/integration_tests/tests/test_dbt_artifacts/test_artifacts.py +++ b/integration_tests/tests/test_dbt_artifacts/test_artifacts.py @@ -191,11 +191,14 @@ def test_source_freshness_results(test_id: str, dbt_project: DbtProject): dbt_project, {"source_freshness_run_project_hooks": True} ): dbt_project.dbt_runner.source_freshness() - dbt_project.read_table( + results = dbt_project.read_table( "dbt_source_freshness_results", where=f"unique_id = 'source.elementary_tests.test_source.{test_id}'", raise_if_empty=True, ) + assert all( + result["max_loaded_at_time_ago_in_s"] is not None for result in results + ) def test_timings(dbt_project: DbtProject): From 4aecb190a79113c6d7f457e06f4bd571013c9c45 Mon Sep 17 00:00:00 2001 From: Sindre Grindheim Date: Tue, 10 Nov 2026 10:32:08 +0100 Subject: [PATCH 3/6] test: cover the dbt Fusion fallback for the source freshness age Move the age lookup into get_freshness_result_age and test it with dbt-core-shaped (`age`) and dbt-Fusion-shaped (`max_loaded_at_time_ago_in_s`) results, since test_source_freshness_results is skipped on Fusion and never reaches the fallback. Co-Authored-By: Claude Opus 5.5 --- .../macros/test_get_freshness_result_age.sql | 4 ++ .../test_dbt_artifacts/test_artifacts.py | 1 + .../test_get_freshness_result_age.py | 37 +++++++++++++++++++ .../dbt_artifacts/upload_source_freshness.sql | 26 +++++++------ 4 files changed, 57 insertions(+), 11 deletions(-) create mode 100644 integration_tests/dbt_project/macros/test_get_freshness_result_age.sql create mode 100644 integration_tests/tests/test_dbt_artifacts/test_get_freshness_result_age.py diff --git a/integration_tests/dbt_project/macros/test_get_freshness_result_age.sql b/integration_tests/dbt_project/macros/test_get_freshness_result_age.sql new file mode 100644 index 000000000..2aa4c9de7 --- /dev/null +++ b/integration_tests/dbt_project/macros/test_get_freshness_result_age.sql @@ -0,0 +1,4 @@ +{% macro test_get_freshness_result_age(result_dict) %} + {% set result = elementary.get_freshness_result_age(result_dict) %} + {{ return(result) }} +{% endmacro %} diff --git a/integration_tests/tests/test_dbt_artifacts/test_artifacts.py b/integration_tests/tests/test_dbt_artifacts/test_artifacts.py index ace52b24d..ca8f1190c 100644 --- a/integration_tests/tests/test_dbt_artifacts/test_artifacts.py +++ b/integration_tests/tests/test_dbt_artifacts/test_artifacts.py @@ -145,6 +145,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): + """dbt source freshness uploads its results, including the time since the last load.""" database_property, schema_property = get_database_and_schema_properties( dbt_project.target ) diff --git a/integration_tests/tests/test_dbt_artifacts/test_get_freshness_result_age.py b/integration_tests/tests/test_dbt_artifacts/test_get_freshness_result_age.py new file mode 100644 index 000000000..4a58a94f6 --- /dev/null +++ b/integration_tests/tests/test_dbt_artifacts/test_get_freshness_result_age.py @@ -0,0 +1,37 @@ +import json + +import pytest +from dbt_project import DbtProject + + +@pytest.mark.parametrize( + "result_dict,expected_output", + [ + # dbt-core + ({"age": 120.5}, 120.5), + ({"age": 0}, 0), + # dbt-fusion: no `age`, the value is in max_loaded_at_time_ago_in_s + ({"max_loaded_at_time_ago_in_s": 88.0}, 88.0), + ({"age": None, "max_loaded_at_time_ago_in_s": 88.0}, 88.0), + ({}, None), + ], +) +def test_get_freshness_result_age( + dbt_project: DbtProject, result_dict, expected_output +): + """get_freshness_result_age reads `age` (dbt-core) and falls back to + `max_loaded_at_time_ago_in_s` (dbt-fusion).""" + result = dbt_project.dbt_runner.run_operation( + "elementary_tests.test_get_freshness_result_age", + macro_args={"result_dict": result_dict}, + ) + # When the macro returns None, log_macro_results doesn't log anything, + # so run_operation returns an empty list + if not result: + actual_output = None + else: + actual_output = json.loads(result[0]) + assert actual_output == expected_output, ( + f"get_freshness_result_age({result_dict!r}) returned {actual_output!r}, " + f"expected {expected_output!r}" + ) diff --git a/macros/edr/dbt_artifacts/upload_source_freshness.sql b/macros/edr/dbt_artifacts/upload_source_freshness.sql index 1e5b8ef42..b3a18610c 100644 --- a/macros/edr/dbt_artifacts/upload_source_freshness.sql +++ b/macros/edr/dbt_artifacts/upload_source_freshness.sql @@ -35,23 +35,15 @@ } ) %} {% endif %} - {# - dbt-core exposes the time since the last load as `age`; dbt-fusion has no - `age` and exposes it as `max_loaded_at_time_ago_in_s` (the sources.json name). - #} - {% set max_loaded_at_time_ago_in_s = result_dict.get("age") %} - {% if max_loaded_at_time_ago_in_s is none %} - {% set max_loaded_at_time_ago_in_s = result_dict.get( - "max_loaded_at_time_ago_in_s" - ) %} - {% endif %} {% do return( { "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": max_loaded_at_time_ago_in_s, + "max_loaded_at_time_ago_in_s": elementary.get_freshness_result_age( + result_dict + ), "criteria": result_dict.node.get("freshness", {}), "adapter_response": result_dict.adapter_response, "timing": result_dict.timing, @@ -61,6 +53,18 @@ ) %} {% endmacro %} +{# + dbt-core exposes the time since the last load as `age`; dbt-fusion has no + `age` and exposes it as `max_loaded_at_time_ago_in_s` (the sources.json name). +#} +{% macro get_freshness_result_age(result_dict) %} + {% set age = result_dict.get("age") %} + {% if age is none %} + {% set age = result_dict.get("max_loaded_at_time_ago_in_s") %} + {% endif %} + {% do return(age) %} +{% endmacro %} + {% macro flatten_source_freshness(node_dict) %} {% set compile_timing = {} %} {% set execute_timing = {} %} From 8481d511e3575bef650807147434e62d9daf5f50 Mon Sep 17 00:00:00 2001 From: Sindre Grindheim Date: Tue, 10 Nov 2026 11:08:45 +0100 Subject: [PATCH 4/6] test: point the freshness test's source at the seed's schema Under pytest-xdist, seeds land in `target.schema` plus the worker suffix (`schema_name_suffix`), but the source pointed at `target.schema`. The freshness check then failed with a runtime error and stored a row without `max_loaded_at_time_ago_in_s`, which the old row-exists check didn't notice. Add SCHEMA_NAME_SUFFIX like DbtProject's own sources do. Co-Authored-By: Claude Opus 5.5 --- integration_tests/tests/test_dbt_artifacts/test_artifacts.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/integration_tests/tests/test_dbt_artifacts/test_artifacts.py b/integration_tests/tests/test_dbt_artifacts/test_artifacts.py index ca8f1190c..b01d076c0 100644 --- a/integration_tests/tests/test_dbt_artifacts/test_artifacts.py +++ b/integration_tests/tests/test_dbt_artifacts/test_artifacts.py @@ -2,7 +2,7 @@ import pytest from dbt_flags import set_flags -from dbt_project import DbtProject +from dbt_project import SCHEMA_NAME_SUFFIX, DbtProject from dbt_utils import get_database_and_schema_properties TEST_MODEL = "one" @@ -156,7 +156,7 @@ def test_source_freshness_results(test_id: str, dbt_project: DbtProject): ) source_def = { "name": "test_source", - "schema": f"{{{{ target.{schema_property} }}}}", + "schema": f"{{{{ target.{schema_property} }}}}{SCHEMA_NAME_SUFFIX}", "tables": [ { "name": test_id, From f6033d6720eab95998eed4cba0f2823e5d7ed183 Mon Sep 17 00:00:00 2001 From: Sindre Grindheim Date: Tue, 10 Nov 2026 11:32:27 +0100 Subject: [PATCH 5/6] test: cast the freshness test's loaded_at_field with T-SQL syntax SQL Server and Fabric have no `::` cast, so the freshness check failed with "Type UPDATE_TIME is not a defined system type" and stored a runtime-error row without the age. Co-Authored-By: Claude Opus 5.5 --- .../tests/test_dbt_artifacts/test_artifacts.py | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/integration_tests/tests/test_dbt_artifacts/test_artifacts.py b/integration_tests/tests/test_dbt_artifacts/test_artifacts.py index b01d076c0..42f812810 100644 --- a/integration_tests/tests/test_dbt_artifacts/test_artifacts.py +++ b/integration_tests/tests/test_dbt_artifacts/test_artifacts.py @@ -149,11 +149,14 @@ def test_source_freshness_results(test_id: str, dbt_project: DbtProject): database_property, schema_property = get_database_and_schema_properties( dbt_project.target ) - 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')" - ) + if dbt_project.target == "dremio": + loaded_at_field = ( + "TO_TIMESTAMP(SUBSTRING(UPDATE_TIME, 0, 23), 'YYYY-MM-DD HH24:MI:SS.FFF')" + ) + elif dbt_project.is_tsql: + loaded_at_field = 'CAST("UPDATE_TIME" AS datetime2(6))' + else: + loaded_at_field = '"UPDATE_TIME"::timestamp' source_def = { "name": "test_source", "schema": f"{{{{ target.{schema_property} }}}}{SCHEMA_NAME_SUFFIX}", From ac5c8547503021fb50e575a1ea2de41a34b98f2b Mon Sep 17 00:00:00 2001 From: Sindre Grindheim Date: Tue, 10 Nov 2026 12:33:22 +0100 Subject: [PATCH 6/6] test: revert the changes to test_source_freshness_results The new assertion exposed that the test's freshness check fails with a runtime error on several targets (schema suffix, T-SQL cast, and more on Trino, Dremio and Spark). That is outside this fix, which test_get_freshness_result_age covers on every target. Co-Authored-By: Claude Opus 5.5 --- .../test_dbt_artifacts/test_artifacts.py | 23 +++++++------------ 1 file changed, 8 insertions(+), 15 deletions(-) diff --git a/integration_tests/tests/test_dbt_artifacts/test_artifacts.py b/integration_tests/tests/test_dbt_artifacts/test_artifacts.py index 42f812810..4b64fc5f2 100644 --- a/integration_tests/tests/test_dbt_artifacts/test_artifacts.py +++ b/integration_tests/tests/test_dbt_artifacts/test_artifacts.py @@ -2,7 +2,7 @@ import pytest from dbt_flags import set_flags -from dbt_project import SCHEMA_NAME_SUFFIX, DbtProject +from dbt_project import DbtProject from dbt_utils import get_database_and_schema_properties TEST_MODEL = "one" @@ -145,21 +145,17 @@ 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): - """dbt source freshness uploads its results, including the time since the last load.""" database_property, schema_property = get_database_and_schema_properties( dbt_project.target ) - if dbt_project.target == "dremio": - loaded_at_field = ( - "TO_TIMESTAMP(SUBSTRING(UPDATE_TIME, 0, 23), 'YYYY-MM-DD HH24:MI:SS.FFF')" - ) - elif dbt_project.is_tsql: - loaded_at_field = 'CAST("UPDATE_TIME" AS datetime2(6))' - else: - loaded_at_field = '"UPDATE_TIME"::timestamp' + 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')" + ) source_def = { "name": "test_source", - "schema": f"{{{{ target.{schema_property} }}}}{SCHEMA_NAME_SUFFIX}", + "schema": f"{{{{ target.{schema_property} }}}}", "tables": [ { "name": test_id, @@ -195,14 +191,11 @@ def test_source_freshness_results(test_id: str, dbt_project: DbtProject): dbt_project, {"source_freshness_run_project_hooks": True} ): dbt_project.dbt_runner.source_freshness() - results = dbt_project.read_table( + dbt_project.read_table( "dbt_source_freshness_results", where=f"unique_id = 'source.elementary_tests.test_source.{test_id}'", raise_if_empty=True, ) - assert all( - result["max_loaded_at_time_ago_in_s"] is not None for result in results - ) def test_timings(dbt_project: DbtProject):