dbt-f2dq/macros/metrics/base/queries.sql

{% macro metrics_base_compute_for_thread(thread_value, ref_model) %}
    {%- set tables =  run_query(f2dq.get_tables()) %}
    {%- for mtable in tables %}
        -- we are splitting computing metrics to 4 different threads
        {% set for_loop_mod = (loop.index % 4) %}
        {% if for_loop_mod == thread_value %}
            {% set model = get_model_config(mtable) %}

            {% set columns_to_query = [] %}
            {% set size = 0 %}

            {% 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 %}

                {% set columns_size = columns_to_query| length %}

                {% if columns_size == var('f2dq:max_columns_in_query') %}
                    {%- set insert_stats_query = f2dq.metrics_base_insert(model, ref_model, columns_to_query) -%}

                    {% if insert_stats_query %}
                        {% do run_query(insert_stats_query) %}
                    {% endif %}
                    {% do columns_to_query.clear() %}
                {% endif %}
            {% endfor %}

            {%- set insert_stats_query = f2dq.metrics_base_insert(model, ref_model, columns_to_query, table_level=True) -%}
            {% do run_query(insert_stats_query) %}

            {{ dbt_utils.log_info('[f2dq_log] - finished computing metrics for:' ~ model.model_name) }}
        {% endif %}
    {% endfor %}
{% 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__<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) }}
    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 {% 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, scope='windowed') %}
    {% if scope == 'global' or time_filter is none %}
        {{ return ('global__' + metric_name) }}
    {% else %}
        {{ return (metric_name) }}
    {% endif %}
{% endmacro %}