Fix metric bucket anchoring and add global whole-table metrics
Generate Documentation / generate-docs (push) Canceled after 0s
Run package tests for all dbs / test-postgres (push) Canceled after 0s
Run package tests for all dbs / test-other-dbs (bigquery) (push) Canceled after 0s
Run package tests for all dbs / test-other-dbs (redshift) (push) Canceled after 0s
Run package tests for all dbs / test-other-dbs (snowflake) (push) Canceled after 0s
Run package tests for all dbs / clean-up-schemas (bigquery) (push) Canceled after 0s
Run package tests for all dbs / clean-up-schemas (redshift) (push) Canceled after 0s
Run package tests for all dbs / clean-up-schemas (snowflake) (push) Canceled after 0s
Generate Documentation / generate-docs (push) Canceled after 0s
Run package tests for all dbs / test-postgres (push) Canceled after 0s
Run package tests for all dbs / test-other-dbs (bigquery) (push) Canceled after 0s
Run package tests for all dbs / test-other-dbs (redshift) (push) Canceled after 0s
Run package tests for all dbs / test-other-dbs (snowflake) (push) Canceled after 0s
Run package tests for all dbs / clean-up-schemas (bigquery) (push) Canceled after 0s
Run package tests for all dbs / clean-up-schemas (redshift) (push) Canceled after 0s
Run package tests for all dbs / clean-up-schemas (snowflake) (push) Canceled after 0s
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__<metric>` 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 <cursoragent@cursor.com>
This commit is contained in:
+17
-3
@@ -1,5 +1,5 @@
|
|||||||
name: "f2dq"
|
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
|
config-version: 2
|
||||||
|
|
||||||
require-dbt-version: [">=1.0.0", "<2.0.0"]
|
require-dbt-version: [">=1.0.0", "<2.0.0"]
|
||||||
@@ -15,8 +15,22 @@ on-run-end:
|
|||||||
|
|
||||||
vars:
|
vars:
|
||||||
f2dq:max_columns_in_query: 10
|
f2dq:max_columns_in_query: 10
|
||||||
f2dq:time_window_end: '{{ run_started_at.strftime("%Y-%m-%d 00:00:00") }}'
|
# The metric bucket is the run's calendar day: [midnight of the run date, +1d).
|
||||||
f2dq:time_window_start: '{{ (run_started_at - modules.datetime.timedelta(1)).strftime("%Y-%m-%d 00:00:00") }}'
|
#
|
||||||
|
# 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__<metric>` (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:anomaly_detection_look_back_days: 30
|
||||||
f2dq:select: null
|
f2dq:select: null
|
||||||
f2dq:anomalies_filtered: f2dq_anomalies
|
f2dq:anomalies_filtered: f2dq_anomalies
|
||||||
|
|||||||
@@ -19,6 +19,17 @@ sources:
|
|||||||
|
|
||||||
vars:
|
vars:
|
||||||
f2dq:store_table_samples: true
|
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:
|
f2dq:anomaly_detector:
|
||||||
name: modified_z_score
|
name: modified_z_score
|
||||||
threshold: 0.6
|
threshold: 0.6
|
||||||
|
|||||||
@@ -37,37 +37,77 @@
|
|||||||
{% endmacro %}
|
{% endmacro %}
|
||||||
|
|
||||||
{% macro metrics_base_insert(model, ref_model, columns, table_level=False) %}
|
{% macro metrics_base_insert(model, ref_model, columns, table_level=False) %}
|
||||||
|
|
||||||
{% set col_exprs = f2dq.metrics_base_expressions(model, columns, table_level) %}
|
{% set col_exprs = f2dq.metrics_base_expressions(model, columns, table_level) %}
|
||||||
{% if col_exprs == [] %}
|
{% if col_exprs == [] %}
|
||||||
{{ return ('') }}
|
{{ return ('') }}
|
||||||
{% endif %}
|
{% 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__<metric>`. 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) }}
|
insert into {{ ref(ref_model) }}
|
||||||
with temp_table_metrics as (
|
with
|
||||||
select
|
{%- for scope in scopes %}
|
||||||
|
temp_table_metrics_{{ scope }} as (
|
||||||
|
select
|
||||||
{%- for col_expr in col_exprs %}
|
{%- for col_expr in col_exprs %}
|
||||||
( {{ col_expr.expr }} ) as {{ f2dq.quote_column_name(col_expr.col_name + '___' + col_expr.metric) }}
|
( {{ col_expr.expr }} ) as {{ f2dq.quote_column_name(col_expr.col_name + '___' + col_expr.metric) }}
|
||||||
{%- if not loop.last %},{%- endif %}
|
{%- if not loop.last %},{%- endif %}
|
||||||
{% endfor %}
|
{%- endfor %}
|
||||||
from
|
from {{ model.table_name }}
|
||||||
{{ model.table_name }}
|
where {% if scope == 'global' %}true{% else %}{{ in_time_window(model.time_filter) }}{% endif %}
|
||||||
where
|
){% if not loop.last %},{% endif %}
|
||||||
{{ in_time_window(model.time_filter) }}
|
{%- endfor %}
|
||||||
)
|
|
||||||
|
|
||||||
{%- 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 %}
|
|
||||||
|
|
||||||
|
{%- 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 %}
|
{% endmacro %}
|
||||||
|
|
||||||
{% macro get_final_metric_name(metric_name, time_filter) %}
|
{% macro get_final_metric_name(metric_name, time_filter, scope='windowed') %}
|
||||||
{% if time_filter is none %}
|
{% if scope == 'global' or time_filter is none %}
|
||||||
{{ return ('global__' + metric_name) }}
|
{{ return ('global__' + metric_name) }}
|
||||||
{% else %}
|
{% else %}
|
||||||
{{ return (metric_name) }}
|
{{ return (metric_name) }}
|
||||||
|
|||||||
@@ -36,7 +36,7 @@
|
|||||||
(
|
(
|
||||||
select
|
select
|
||||||
{{ overview_select_base('metric', 'computed_on')}}
|
{{ 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
|
from
|
||||||
{{ ref('f2dq_base_metrics') }}
|
{{ ref('f2dq_base_metrics') }}
|
||||||
where {{ in_date_window('time_window_end', start_date, end_date) }}
|
where {{ in_date_window('time_window_end', start_date, end_date) }}
|
||||||
|
|||||||
@@ -9,20 +9,33 @@
|
|||||||
-- depends_on: {{ ref('f2dq_test_history') }}
|
-- depends_on: {{ ref('f2dq_test_history') }}
|
||||||
{% set command = flags.WHICH %}
|
{% set command = flags.WHICH %}
|
||||||
{% if execute and results and command in ('test', 'build') %}
|
{% if execute and results and command in ('test', 'build') %}
|
||||||
{% set tests = [] %}
|
{% set history_rel = ref('f2dq_test_history') %}
|
||||||
{% for el in results %}
|
{% set history_exists = adapter.get_relation(
|
||||||
{% if el.node.resource_type.value == 'test' %}
|
database=history_rel.database,
|
||||||
{% do tests.append(f2dq.test_data_dict(el)) %}
|
schema=history_rel.schema,
|
||||||
{% endif %}
|
identifier=history_rel.identifier
|
||||||
{% endfor %}
|
) is not none %}
|
||||||
|
{% if not history_exists %}
|
||||||
{% if tests %}
|
{% do log(
|
||||||
{% do f2dq.insert_list_to_table(
|
"f2dq_test_history relation missing (" ~ history_rel ~ "); skipping save_test_history",
|
||||||
ref('f2dq_test_history'),
|
info=True
|
||||||
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() }
|
|
||||||
) %}
|
) %}
|
||||||
|
{% 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 %}
|
||||||
|
|
||||||
{% endif %}
|
{% endif %}
|
||||||
|
|||||||
Reference in New Issue
Block a user