Feature: Add consumption and observability to dbt Artifacts

.gitignore

@@ -32,3 +32,4 @@ CLAUDE.md
 # Python bytecode cache (e.g. from running scripts/release/*.py)
 __pycache__/
 *.pyc
+specs/

integration_test_project/dbt_project.yml

@@ -20,6 +20,12 @@ vars:
   dbt_vars: ["test_dbt_vars_1", "test_dbt_vars_2", "test_dbt_vars_3"]
   dbt_artifacts_exclude_all_results: true
   is_development: '{{ env_var("IS_DEVELOPMENT", false) | as_bool }}'
+  # Consumption/observability marts: classify this project's snowflake
+  # invocations as deployment so the consumption marts have SMB to aggregate,
+  # and give the forecast a plan allowance. Only consumed by the Snowflake-gated
+  # v1 marts; inert on other targets.
+  dbt_artifacts_billing_plan: developer
+  dbt_artifacts_deployment_targets: ['snowflake']
 
 models:
   +persist_docs:

integration_test_project/tests/assert_consumption_by_model_reconciles_daily.sql

@@ -0,0 +1,28 @@
+{{ config(enabled = target.type == "snowflake") }}
+-- Monthly SMB totals must reconcile between consumption_by_model and
+-- consumption_daily. Fails (returns rows) on any mismatched month.
+with by_model as (
+    select
+        date_trunc('month', billing_month) as month
+        , sum(smb_quantity) as smb
+    from {{ ref("fct_dbt__consumption_by_model") }}
+    group by 1
+),
+
+daily as (
+    select
+        date_trunc('month', date_day) as month
+        , sum(quantity) as smb
+    from {{ ref("fct_dbt__consumption_daily") }}
+    where meter = 'smb'
+    group by 1
+)
+
+select
+    by_model.month as by_model_month
+    , daily.month as daily_month
+    , by_model.smb as by_model_smb
+    , daily.smb as daily_smb
+from by_model
+full outer join daily on by_model.month = daily.month
+where coalesce(by_model.smb, -1) != coalesce(daily.smb, -1)

integration_test_project/tests/assert_consumption_daily_has_smb.sql

@@ -0,0 +1,8 @@
+{{ config(enabled = target.type == "snowflake") }}
+-- consumption_daily must have at least one 'smb' row with quantity > 0
+-- after the harness's deployment-classed runs. Fails (returns a row) if none.
+select count(*) as smb_rows_with_quantity
+from {{ ref("fct_dbt__consumption_daily") }}
+where meter = 'smb'
+    and quantity > 0
+having count(*) = 0

integration_test_project/tests/assert_consumption_forecast_current_month.sql

@@ -0,0 +1,14 @@
+{{ config(enabled = target.type == "snowflake") }}
+-- The forecast must have exactly one 'smb' row for the current billing
+-- month, with non-null month_to_date_quantity and non-null allowance (developer
+-- plan is set in dbt_project.yml). Fails if not exactly one, or either is null.
+select
+    count(*) as n_rows
+    , count(month_to_date_quantity) as non_null_mtd
+    , count(allowance) as non_null_allowance
+from {{ ref("fct_dbt__consumption_forecast") }}
+where meter = 'smb'
+    and billing_month = date_trunc('month', current_date())
+having count(*) != 1
+    or count(month_to_date_quantity) != 1
+    or count(allowance) != 1

integration_test_project/tests/assert_dag_bottlenecks_valid.sql

@@ -0,0 +1,11 @@
+{{ config(enabled = target.type == "snowflake") }}
+-- dag_bottlenecks must be valid (possibly empty). Fails on any negative
+-- stall time.
+select
+    date_day
+    , parent_node_id
+    , total_stall_seconds
+    , max_stall_seconds
+from {{ ref("fct_dbt__dag_bottlenecks") }}
+where total_stall_seconds < 0
+    or max_stall_seconds < 0

integration_test_project/tests/assert_flaky_tests_valid.sql

@@ -0,0 +1,13 @@
+{{ config(enabled = target.type == "snowflake") }}
+-- flaky_tests must be valid (possibly empty). Fails on any out-of-range
+-- flake_rate or flips exceeding executions.
+select
+    test_node_id
+    , month
+    , flips
+    , executions
+    , flake_rate
+from {{ ref("fct_dbt__flaky_tests") }}
+where flake_rate < 0
+    or flake_rate > 1
+    or flips > executions

integration_test_project/tests/assert_lineage_edges_cover_models.sql

@@ -0,0 +1,30 @@
+{{ config(enabled = target.type == "snowflake") }}
+-- lineage_edges must be non-empty, and every model in the latest graph
+-- state that declares dependencies must appear as a child of >= 1 edge.
+-- Fails (returns rows) for any model-with-deps that has no edge.
+with latest_models as (
+    select
+        node_id
+        , depends_on_nodes
+        , row_number() over (
+            partition by node_id order by run_started_at desc
+        ) as run_rank
+    from {{ ref("stg_dbt__models") }}
+),
+
+models_with_deps as (
+    select node_id
+    from latest_models
+    where run_rank = 1
+        and array_size(depends_on_nodes) > 0
+),
+
+edge_children as (
+    select distinct child_node_id
+    from {{ ref("dim_dbt__lineage_edges") }}
+)
+
+select models_with_deps.node_id as model_missing_edge
+from models_with_deps
+left join edge_children on models_with_deps.node_id = edge_children.child_node_id
+where edge_children.child_node_id is null

integration_test_project/tests/assert_model_performance_valid.sql

@@ -0,0 +1,15 @@
+{{ config(enabled = target.type == "snowflake") }}
+-- model_performance must return rows and be internally consistent -- no
+-- median_runtime without a success, and the mart must not be empty. (A null
+-- median on a day whose only successes were full refreshes is CORRECT by
+-- design, so we assert consistency rather than blanket non-null.)
+select 'median_without_success' as issue
+from {{ ref("fct_dbt__model_performance") }}
+where median_runtime is not null
+    and success_count = 0
+
+union all
+
+select 'empty_mart' as issue
+from (select count(*) as row_count from {{ ref("fct_dbt__model_performance") }})
+where row_count = 0

integration_test_project/tests/assert_run_health_daily_has_invocations.sql

@@ -0,0 +1,6 @@
+{{ config(enabled = target.type == "snowflake") }}
+-- run_health_daily must have at least one row with invocations > 0.
+select count(*) as days_with_invocations
+from {{ ref("fct_dbt__run_health_daily") }}
+where invocations > 0
+having count(*) = 0

macros/consumption/classify_invocation_billing.sql

@@ -0,0 +1,87 @@
+{#-
+    classify_invocation_billing()
+
+    Returns a scalar-per-row SQL `case` expression that classifies each
+    invocation as 'deployment' or 'development'. dbt bills only orchestrated
+    deployment builds, so every consumption metric filters on this.
+
+    Consumed inside the new consumption models' CTEs only -- it references the
+    columns dbt_cloud_job_id, target_name and env_vars, which must be in scope
+    (they are on stg_dbt__invocations). This macro does NOT modify any existing
+    staging/source contract.
+
+    Precedence (first match wins):
+      1. dbt_cloud_job_id is not null                         -> deployment
+      2. lower(target_name) in deployment_targets (lowered)   -> deployment
+      3. optional: env var named by dbt_artifacts_deployment_env_var is
+         present and truthy in env_vars                       -> deployment
+      fallback                                                -> development
+
+    Escape hatch: dbt_artifacts_count_all_invocations = true classes every
+    invocation as deployment.
+
+    Rule 3 needs adapter-specific JSON access, so it lives only in snowflake__
+    (env_vars is stored via type_json() = OBJECT on Snowflake, accessed with
+    the variant path env_vars:"NAME"). It is emitted only when the var is set,
+    so the default path parses no JSON. default__ raises a clear compile error
+    if rule 3 is requested on an adapter that does not implement it yet.
+-#}
+
+{% macro classify_invocation_billing() %}
+    {{ return(adapter.dispatch('classify_invocation_billing', 'dbt_artifacts')()) }}
+{% endmacro %}
+
+
+{#- Shared predicate for rule 2: lower(target_name) in the (lowered) deployment
+    target list. Emits `false` for an empty list so `in ()` is never produced. -#}
+{% macro _billing_deployment_targets_predicate() %}
+    {%- set targets = var('dbt_artifacts_deployment_targets', ['prod', 'production', 'ci']) -%}
+    {%- if targets | length == 0 -%}
+        false
+    {%- else -%}
+        lower(target_name) in (
+            {%- for t in targets -%}
+                '{{ t | lower }}'{% if not loop.last %}, {% endif %}
+            {%- endfor -%}
+        )
+    {%- endif -%}
+{% endmacro %}
+
+
+{% macro default__classify_invocation_billing() %}
+    {%- if var('dbt_artifacts_count_all_invocations', false) -%}
+        cast('deployment' as {{ dbt.type_string() }})
+    {%- else -%}
+        {#- Raise only at run time (execute == true), never during parse, so a
+            non-Snowflake consumer's `dbt parse` is unaffected even with this var
+            set. The consumption models that call this are Snowflake-gated anyway. -#}
+        {%- if var('dbt_artifacts_deployment_env_var', none) is not none and execute -%}
+            {{ exceptions.raise_compiler_error(
+                "dbt_artifacts_deployment_env_var (billing classification rule 3) requires adapter-specific JSON support and is only implemented for Snowflake in v1. Unset the var or run on Snowflake."
+            ) }}
+        {%- endif -%}
+        case
+            when dbt_cloud_job_id is not null then 'deployment'
+            when {{ dbt_artifacts._billing_deployment_targets_predicate() }} then 'deployment'
+            else 'development'
+        end
+    {%- endif -%}
+{% endmacro %}
+
+
+{% macro snowflake__classify_invocation_billing() %}
+    {%- if var('dbt_artifacts_count_all_invocations', false) -%}
+        cast('deployment' as {{ dbt.type_string() }})
+    {%- else -%}
+        {%- set env_var_name = var('dbt_artifacts_deployment_env_var', none) -%}
+        case
+            when dbt_cloud_job_id is not null then 'deployment'
+            when {{ dbt_artifacts._billing_deployment_targets_predicate() }} then 'deployment'
+            {%- if env_var_name is not none %}
+            when coalesce(env_vars:"{{ env_var_name }}"::string, '') not in ('', 'false', 'False', '0')
+                then 'deployment'
+            {%- endif %}
+            else 'development'
+        end
+    {%- endif -%}
+{% endmacro %}

macros/consumption/get_smb_allowance.sql

@@ -0,0 +1,28 @@
+{#-
+    get_smb_allowance()
+
+    Resolves the monthly SMB allowance at compile time, returning a Python
+    number or none. Resolution order:
+      1. var('dbt_artifacts_smb_allowance')  -- explicit, wins
+      2. plan default from var('dbt_artifacts_billing_plan')
+         (developer: 3000, starter: 15000, enterprise: 100000)
+      3. none  -- forecast projections still populate; breach columns stay null
+
+    Plan allowances are published defaults that WILL drift; they are overridable
+    and documented here in one place. Returns none for an unknown plan name.
+-#}
+
+{% macro get_smb_allowance() %}
+    {%- set explicit = var("dbt_artifacts_smb_allowance", none) -%}
+    {%- if explicit is not none -%}
+        {{ return(explicit) }}
+    {%- endif -%}
+
+    {%- set plan = var("dbt_artifacts_billing_plan", none) -%}
+    {%- set plan_allowances = {"developer": 3000, "starter": 15000, "enterprise": 100000} -%}
+    {%- if plan is not none and plan in plan_allowances -%}
+        {{ return(plan_allowances[plan]) }}
+    {%- endif -%}
+
+    {{ return(none) }}
+{% endmacro %}

macros/database_specific_helpers/datetime_helpers.sql

@@ -0,0 +1,31 @@
+{#-
+    cast_to_utc_date(column_expr)
+
+    Converts a stored run timestamp to its UTC calendar date. Shared
+    infrastructure for every daily-grain mart in the consumption and
+    observability features.
+
+    Storage assumption (Snowflake): dbt captures run_started_at (and the
+    execution-level *_started_at / *_completed_at columns) in UTC, and the
+    package stores them via type_timestamp(), which on Snowflake is
+    TIMESTAMP_NTZ -- a wall-clock value carrying no timezone. Casting a
+    TIMESTAMP_NTZ to date simply truncates to the date part, so the result is
+    invariant to the session TIMEZONE setting and is already the correct UTC
+    date. convert_timezone() is deliberately NOT used: on an NTZ input it
+    assumes the value is in the session timezone and would reintroduce session
+    dependence, corrupting the date near midnight.
+-#}
+
+{% macro cast_to_utc_date(column_expr) %}
+    {{ return(adapter.dispatch('cast_to_utc_date', 'dbt_artifacts')(column_expr)) }}
+{% endmacro %}
+
+{% macro default__cast_to_utc_date(column_expr) %}
+    cast({{ column_expr }} as date)
+{% endmacro %}
+
+{% macro snowflake__cast_to_utc_date(column_expr) %}
+    {#- run_started_at is TIMESTAMP_NTZ holding dbt's UTC timestamp; a direct
+        date cast yields the UTC calendar date and is session-TZ invariant. -#}
+    cast({{ column_expr }} as date)
+{% endmacro %}

macros/database_specific_helpers/flatten_json_array.sql

@@ -0,0 +1,44 @@
+{#-
+    flatten_json_array(array_column, alias)
+
+    Renders a lateral table-function FROM-clause fragment that explodes an
+    array-valued column into one row per element. Intended for use in the
+    FROM clause after a comma:
+
+        from my_model
+        , {{ dbt_artifacts.flatten_json_array('my_model.depends_on_nodes', 'dep') }}
+
+    The per-element scalar is then accessed as `{{ alias }}.value` (cast as
+    needed). On Snowflake the package's depends_on_nodes columns are native
+    ARRAY (type_array()), so the input is flattened directly -- no parse_json.
+
+    Snowflake only in v1. Other adapters raise a clear compile error; the model
+    that uses this helper is itself gated to Snowflake. Cross-adapter overrides
+    (BigQuery unnest, Postgres jsonb_array_elements_text, Trino unnest, Spark
+    explode, SQL Server openjson) are planned.
+-#}
+
+{% macro flatten_json_array(array_column, alias) %}
+    {{ return(adapter.dispatch('flatten_json_array', 'dbt_artifacts')(array_column, alias)) }}
+{% endmacro %}
+
+{% macro default__flatten_json_array(array_column, alias) %}
+    {#- The models using this helper are Snowflake-gated, so on other adapters
+        this branch is reached only during `dbt parse` (execute == false), never
+        at run time. Emit a parse-safe placeholder so parsing succeeds on every
+        adapter, and raise a clear error only if a model is actually built on an
+        unsupported adapter (execute == true). This keeps the package backwards-
+        compatible: a non-Snowflake consumer's `dbt parse`/`dbt run` is unaffected. -#}
+    {%- if execute -%}
+        {{ exceptions.raise_compiler_error(
+            "dbt_artifacts.flatten_json_array() is only implemented for Snowflake in v1 (adapter '"
+            ~ target.type
+            ~ "' is not supported yet). The models that use it are Snowflake-gated, so this should not be reachable; cross-adapter support is planned."
+        ) }}
+    {%- endif -%}
+    (select cast(null as {{ dbt.type_string() }}) as value) as {{ alias }}
+{% endmacro %}
+
+{% macro snowflake__flatten_json_array(array_column, alias) %}
+    lateral flatten(input => {{ array_column }}) as {{ alias }}
+{% endmacro %}

macros/database_specific_helpers/statistical_helpers.sql

@@ -0,0 +1,36 @@
+{#-
+    Statistical helpers for the performance-baseline math (observability marts).
+
+    Both macros are aggregate expressions intended for use in GROUP BY queries
+    (e.g. per-day, per-node medians). The default__ implementations use the
+    ANSI ordered-set aggregate percentile_cont(fraction) within group
+    (order by ...), which Snowflake supports natively as an aggregate, so no
+    snowflake__ override is required. Non-Snowflake overrides (SQL Server
+    window-only syntax, Spark approx_percentile) are planned.
+-#}
+
+{#- MEDIAN -#}
+
+{% macro median(column_expr) %}
+    {{ return(adapter.dispatch('median', 'dbt_artifacts')(column_expr)) }}
+{% endmacro %}
+
+{% macro default__median(column_expr) %}
+    percentile_cont(0.5) within group (order by {{ column_expr }})
+{% endmacro %}
+
+{#- PERCENTILE (arbitrary fraction in [0, 1]) -#}
+
+{% macro percentile(column_expr, fraction) %}
+    {{ return(adapter.dispatch('percentile', 'dbt_artifacts')(column_expr, fraction)) }}
+{% endmacro %}
+
+{% macro default__percentile(column_expr, fraction) %}
+    percentile_cont({{ fraction }}) within group (order by {{ column_expr }})
+{% endmacro %}
+
+{#- P95 -- thin sugar over percentile(col, 0.95); not dispatched. -#}
+
+{% macro p95(column_expr) %}
+    {{ dbt_artifacts.percentile(column_expr, 0.95) }}
+{% endmacro %}