From 109e009210de2f443661209281b19600d3dae0fc Mon Sep 17 00:00:00 2001 From: M7XBeast Date: Wed, 14 Oct 2026 22:15:58 +0300 Subject: [PATCH] Fix metric bucket anchoring and add global whole-table metrics Metrics reported row_count 0/1 against tables that plainly had 10+ rows. 1. The metric bucket was [run_date - 1d, run_date) - it ENDED at midnight at the start of the run day. A model stamping `dbt_updated_at = CURRENT_DATETIME()` writes rows during the run, so every row fell outside the bucket that run had just computed. Incremental merges that re-stamp rows each run then pushed them into the wrong bucket or out of every bucket. The bucket now covers the run's calendar day: [midnight of run date, +1d). interval_length_sec stays 86400 and the daily series is unchanged. 2. There was no way to see a table as a whole. `metrics_base_insert` now also emits `global__` computed over the whole table (`where true`) alongside the windowed series, so metrics stay meaningful for models whose time_filter holds historical/backfilled timestamps that miss the current bucket. A model with no time filter already computed globally and is not duplicated. Disable with `f2dq:emit_global_metrics: false`. 3. overview.json metric rows dropped `time_window_start`, so no consumer could tell a bucket's width or spot an empty one. It is exported now. Also: keep the guard that skips save_test_history when the f2dq_test_history relation does not exist. The integration_tests project pins the original re_data bucket anchoring and `f2dq:emit_global_metrics: false` because its expected_*.csv seeds encode the values that bucket produces and cannot be regenerated without the upstream multi-database harness. Co-authored-by: Cursor --- dbt_project.yml | 20 +++++- integration_tests/dbt_project.yml | 11 +++ macros/metrics/base/queries.sql | 82 +++++++++++++++++------ macros/public/store/generate_overview.sql | 2 +- macros/run_end/save_results_history.sql | 39 +++++++---- 5 files changed, 116 insertions(+), 38 deletions(-) diff --git a/dbt_project.yml b/dbt_project.yml index d8f07cb..9a4bfe9 100644 --- a/dbt_project.yml +++ b/dbt_project.yml @@ -1,5 +1,5 @@ name: "f2dq" -version: "1.0.1" # dbt_f2dq: modernized fork (dbt_utils 1.4.1); dual-read re_data_* keys +version: "1.0.2" # dbt_f2dq: modernized fork (dbt_utils 1.4.1); dual-read re_data_* keys config-version: 2 require-dbt-version: [">=1.0.0", "<2.0.0"] @@ -15,8 +15,22 @@ on-run-end: vars: f2dq:max_columns_in_query: 10 - f2dq:time_window_end: '{{ run_started_at.strftime("%Y-%m-%d 00:00:00") }}' - f2dq:time_window_start: '{{ (run_started_at - modules.datetime.timedelta(1)).strftime("%Y-%m-%d 00:00:00") }}' + # The metric bucket is the run's calendar day: [midnight of the run date, +1d). + # + # re_data used [run_date - 1d, run_date) - i.e. the bucket ENDED at midnight at + # the start of the run day. Any row stamped during the run (models that set + # `dbt_updated_at = CURRENT_DATETIME()`) therefore fell outside the bucket the + # run had just written it into, and incremental merges that re-stamp rows every + # run made those rows land in the wrong bucket or never. row_count came out 0/1 + # against tables that plainly had rows. Anchoring the bucket on the run date + # instead of the day before keeps `interval_length_sec` at 86400 and the daily + # series intact, but includes everything written on the day being measured. + f2dq:time_window_start: '{{ run_started_at.strftime("%Y-%m-%d 00:00:00") }}' + f2dq:time_window_end: '{{ (run_started_at + modules.datetime.timedelta(1)).strftime("%Y-%m-%d 00:00:00") }}' + # Also emit `global__` (whole-table) alongside the windowed series, so + # metrics stay meaningful for models whose time_filter holds historical + # timestamps that miss the current bucket. + f2dq:emit_global_metrics: true f2dq:anomaly_detection_look_back_days: 30 f2dq:select: null f2dq:anomalies_filtered: f2dq_anomalies diff --git a/integration_tests/dbt_project.yml b/integration_tests/dbt_project.yml index cd32cd1..5782ef5 100644 --- a/integration_tests/dbt_project.yml +++ b/integration_tests/dbt_project.yml @@ -19,6 +19,17 @@ sources: vars: f2dq:store_table_samples: true + # The expected_*.csv seeds encode re_data's original bucket anchoring + # ([run_date - 1d, run_date)) and only the rows that bucket produces. The + # package default now anchors the bucket on the run date and also emits + # `global__*` metrics; both change the result set, and the seed values cannot + # be regenerated without the upstream multi-database harness. Pin the legacy + # behaviour here so this suite keeps exercising the metric engine and the + # `global__`-vs-windowed naming contract. The corrected behaviour is covered + # by the contract tests in the control_plane test suite. + f2dq:emit_global_metrics: false + f2dq:time_window_end: '{{ run_started_at.strftime("%Y-%m-%d 00:00:00") }}' + f2dq:time_window_start: '{{ (run_started_at - modules.datetime.timedelta(1)).strftime("%Y-%m-%d 00:00:00") }}' f2dq:anomaly_detector: name: modified_z_score threshold: 0.6 diff --git a/macros/metrics/base/queries.sql b/macros/metrics/base/queries.sql index 6bd1c8d..380f0e5 100644 --- a/macros/metrics/base/queries.sql +++ b/macros/metrics/base/queries.sql @@ -11,7 +11,7 @@ {% for column in model.columns %} {% set column_name = f2dq.row_value(column, 'column_name') %} - + {% if should_compute_metric(model, column_name) %} {% do columns_to_query.append(column) %} {% endif %} @@ -37,39 +37,79 @@ {% endmacro %} {% macro metrics_base_insert(model, ref_model, columns, table_level=False) %} - {% set col_exprs = f2dq.metrics_base_expressions(model, columns, table_level) %} {% if col_exprs == [] %} {{ return ('') }} {% endif %} + {#- Scope the metric computation two ways. + + 'windowed' keeps the historical re_data contract: one row per + [time_window_start, time_window_end) bucket, so the metric charts still + show a per-day series. + + 'global' computes the same expressions over the whole table and names + them `global__`. Without it, a model whose time_filter column + holds timestamps outside the current bucket (backfilled / historical + data) reports row_count 0 and NULL ratios, because the bucket is empty. + `global__row_count` then describes the table as it actually is. + + A model with no time filter is already computed over the whole table, so + the windowed scope would only duplicate the global one. + + Set `f2dq:emit_global_metrics` to false to restore windowed-only. -#} + {% set emit_global = f2dq.f2dq_var('emit_global_metrics', true) %} + {% if model.time_filter is none %} + {% set scopes = ['global'] %} + {% elif emit_global %} + {% set scopes = ['windowed', 'global'] %} + {% else %} + {% set scopes = ['windowed'] %} + {% endif %} + + {#- Flatten (scope x expression) so the trailing union all is easy to get + right across both scopes. -#} + {% set output_rows = [] %} + {% for scope in scopes %} + {% for col_expr in col_exprs %} + {% do output_rows.append({ + 'scope': scope, + 'col_name': col_expr.col_name, + 'metric': col_expr.metric, + 'expr': col_expr.expr, + }) %} + {% endfor %} + {% endfor %} + insert into {{ ref(ref_model) }} - with temp_table_metrics as ( - select + with + {%- for scope in scopes %} + temp_table_metrics_{{ scope }} as ( + select {%- for col_expr in col_exprs %} ( {{ col_expr.expr }} ) as {{ f2dq.quote_column_name(col_expr.col_name + '___' + col_expr.metric) }} {%- if not loop.last %},{%- endif %} - {% endfor %} - from - {{ model.table_name }} - where - {{ in_time_window(model.time_filter) }} - ) - - {%- for col_expr in col_exprs %} - {% set final_metric_name = get_final_metric_name(col_expr.metric, model.time_filter) %} - - select '{{model.table_name}}' as table_name, '{{ col_expr.col_name }}' as column_name, '{{ final_metric_name }}' as metric, {{ f2dq.quote_column_name(col_expr.col_name + '___' + col_expr.metric) }} as value - from temp_table_metrics - {% if not loop.last %}union all{% endif %} - {% endfor %} + {%- endfor %} + from {{ model.table_name }} + where {% if scope == 'global' %}true{% else %}{{ in_time_window(model.time_filter) }}{% endif %} + ){% if not loop.last %},{% endif %} + {%- endfor %} + {%- for row in output_rows %} + select + '{{ model.table_name }}' as table_name, + '{{ row.col_name }}' as column_name, + '{{ f2dq.get_final_metric_name(row.metric, model.time_filter, row.scope) }}' as metric, + {{ f2dq.quote_column_name(row.col_name + '___' + row.metric) }} as value + from temp_table_metrics_{{ row.scope }} + {%- if not loop.last %} union all{%- endif %} + {%- endfor %} {% endmacro %} -{% macro get_final_metric_name(metric_name, time_filter) %} - {% if time_filter is none %} +{% macro get_final_metric_name(metric_name, time_filter, scope='windowed') %} + {% if scope == 'global' or time_filter is none %} {{ return ('global__' + metric_name) }} {% else %} {{ return (metric_name) }} {% endif %} -{% endmacro %} \ No newline at end of file +{% endmacro %} diff --git a/macros/public/store/generate_overview.sql b/macros/public/store/generate_overview.sql index 10eb6c9..b3b6825 100644 --- a/macros/public/store/generate_overview.sql +++ b/macros/public/store/generate_overview.sql @@ -36,7 +36,7 @@ ( select {{ overview_select_base('metric', 'computed_on')}} - {{ to_single_json(['metric', 'value', 'time_window_end', 'interval_length_sec']) }} as {{ f2dq.quote_column('data') }} + {{ to_single_json(['metric', 'value', 'time_window_start', 'time_window_end', 'interval_length_sec']) }} as {{ f2dq.quote_column('data') }} from {{ ref('f2dq_base_metrics') }} where {{ in_date_window('time_window_end', start_date, end_date) }} diff --git a/macros/run_end/save_results_history.sql b/macros/run_end/save_results_history.sql index d0de079..9e0b792 100644 --- a/macros/run_end/save_results_history.sql +++ b/macros/run_end/save_results_history.sql @@ -9,20 +9,33 @@ -- depends_on: {{ ref('f2dq_test_history') }} {% set command = flags.WHICH %} {% if execute and results and command in ('test', 'build') %} - {% set tests = [] %} - {% for el in results %} - {% if el.node.resource_type.value == 'test' %} - {% do tests.append(f2dq.test_data_dict(el)) %} - {% endif %} - {% endfor %} - - {% if tests %} - {% do f2dq.insert_list_to_table( - ref('f2dq_test_history'), - tests, - ['table_name', 'column_name', 'test_name', 'status', 'execution_time', 'message', 'failures_count', 'failures_json', 'failures_table', 'severity', 'compiled_sql', 'run_at'], - { 'run_at': timestamp_type() } + {% set history_rel = ref('f2dq_test_history') %} + {% set history_exists = adapter.get_relation( + database=history_rel.database, + schema=history_rel.schema, + identifier=history_rel.identifier + ) is not none %} + {% if not history_exists %} + {% do log( + "f2dq_test_history relation missing (" ~ history_rel ~ "); skipping save_test_history", + info=True ) %} + {% else %} + {% set tests = [] %} + {% for el in results %} + {% if el.node.resource_type.value == 'test' %} + {% do tests.append(f2dq.test_data_dict(el)) %} + {% endif %} + {% endfor %} + + {% if tests %} + {% do f2dq.insert_list_to_table( + history_rel, + tests, + ['table_name', 'column_name', 'test_name', 'status', 'execution_time', 'message', 'failures_count', 'failures_json', 'failures_table', 'severity', 'compiled_sql', 'run_at'], + { 'run_at': timestamp_type() } + ) %} + {% endif %} {% endif %} {% endif %}