diff --git a/docs/domain/metrics/specs/DESIGN.md b/docs/domain/metrics/specs/DESIGN.md index 5ab3f0b70..3be6037f4 100644 --- a/docs/domain/metrics/specs/DESIGN.md +++ b/docs/domain/metrics/specs/DESIGN.md @@ -139,6 +139,25 @@ evidence relation per source and one granularity per measure: - `source_summary`: the finest summary preserved by silver. - `derived_population`: a source entity participating in a derived metric. +All managed evidence and observation tables use the shared +`metric_evidence_table` and `metric_observations_table` dbt macros. They own +materialization, storage keys, partitioning, tags, and bounded query settings. +These settings apply uniformly; model-specific query settings are retained +only when required by model semantics. + +The tables are partitioned by calendar month from `metric_date`. Evidence is +ordered by tenant, source, entity type, entity, measure, date, and record ID; +observations omit the final record ID. Monthly partitioning is physical +storage only: it does not change metric dates, timestamps, row granularity, or +drilldown results. Each insert block may address at most 512 partitions. + +dbt builds a replacement table and exchanges it only after the build +succeeds. A failed or cancelled build therefore leaves the active table in +place, and the next build removes abandoned temporary relations. Replacement +is atomic per table, not across the complete gold DAG. Replacing evidence +invalidates active evidence cursors through the existing snapshot-expired +contract. + Definitions do not declare a separate drilldown strategy. The runtime resolves the definition's existing input roles and source measures, requires every input to use the same evidence relation, and compiles the evidence selection from diff --git a/src/ingestion/dbt/macros/metric_serving_table.sql b/src/ingestion/dbt/macros/metric_serving_table.sql new file mode 100644 index 000000000..dfbb88543 --- /dev/null +++ b/src/ingestion/dbt/macros/metric_serving_table.sql @@ -0,0 +1,50 @@ +{% macro metric_serving_query_settings(join_use_nulls=none) %} + {% set settings = { + 'max_memory_usage': 1610612736, + 'max_threads': 2, + 'max_block_size': 32768, + 'max_insert_block_size': 32768, + 'min_insert_block_size_rows': 32768, + 'min_insert_block_size_bytes': 16777216, + 'max_partitions_per_insert_block': 512, + 'max_bytes_before_external_group_by': 268435456, + 'max_bytes_before_external_sort': 268435456, + 'max_bytes_in_join': 268435456, + 'join_algorithm': 'auto' + } %} + {% if join_use_nulls is not none %} + {% do settings.update({'join_use_nulls': join_use_nulls}) %} + {% endif %} + {{ return(settings) }} +{% endmacro %} + +{% macro metric_serving_table(include_record_id, join_use_nulls=none) %} + {% set order_by = [ + 'tenant_id', + 'source_key', + 'entity_type', + 'entity_id', + 'measure_key', + 'metric_date' + ] %} + {% if include_record_id %} + {% do order_by.append('record_id') %} + {% endif %} + {{ config( + materialized='table', + engine='MergeTree', + order_by=order_by, + partition_by='toYYYYMM(metric_date)', + schema='insight', + tags=['gold'], + query_settings=metric_serving_query_settings(join_use_nulls=join_use_nulls) + ) }} +{% endmacro %} + +{% macro metric_evidence_table(join_use_nulls=none) %} + {{ metric_serving_table(true, join_use_nulls=join_use_nulls) }} +{% endmacro %} + +{% macro metric_observations_table() %} + {{ metric_serving_table(false) }} +{% endmacro %} diff --git a/src/ingestion/gold/ai_metric_evidence.sql b/src/ingestion/gold/ai_metric_evidence.sql index 082d2f560..ff13a9c28 100644 --- a/src/ingestion/gold/ai_metric_evidence.sql +++ b/src/ingestion/gold/ai_metric_evidence.sql @@ -1,17 +1,4 @@ -{{ config( - materialized='table', - engine='MergeTree', - order_by=['tenant_id', 'source_key', 'measure_key', 'entity_id', 'metric_date', 'record_id'], - schema='insight', - alias='ai_metric_evidence', - tags=['gold'], - query_settings={ - 'max_memory_usage': 1610612736, - 'max_threads': 4, - 'max_bytes_before_external_group_by': 805306368, - 'max_bytes_before_external_sort': 805306368 - } -) }} +{{ metric_evidence_table() }} -- Resolution happens HERE, once per gold build: evidence carries BOTH keys — -- `entity_id` is the canonical person id (or '' when identity does not know diff --git a/src/ingestion/gold/ai_metric_observations.sql b/src/ingestion/gold/ai_metric_observations.sql index 6dda7df25..f1765794f 100644 --- a/src/ingestion/gold/ai_metric_observations.sql +++ b/src/ingestion/gold/ai_metric_observations.sql @@ -1,17 +1,4 @@ -{{ config( - materialized='table', - engine='MergeTree', - order_by=['source_key', 'measure_key', 'entity_id', 'metric_date'], - schema='insight', - alias='ai_metric_observations', - tags=['gold'], - query_settings={ - 'max_memory_usage': 1610612736, - 'max_threads': 4, - 'max_bytes_before_external_group_by': 805306368, - 'max_bytes_before_external_sort': 805306368 - } -) }} +{{ metric_observations_table() }} SELECT tenant_id, diff --git a/src/ingestion/gold/collab_metric_evidence.sql b/src/ingestion/gold/collab_metric_evidence.sql index b8c1b7e0f..eb303a54d 100644 --- a/src/ingestion/gold/collab_metric_evidence.sql +++ b/src/ingestion/gold/collab_metric_evidence.sql @@ -1,17 +1,4 @@ -{{ config( - materialized='table', - engine='MergeTree', - order_by=['tenant_id', 'source_key', 'measure_key', 'entity_id', 'metric_date', 'record_id'], - schema='insight', - alias='collab_metric_evidence', - tags=['gold'], - query_settings={ - 'max_memory_usage': 1610612736, - 'max_threads': 4, - 'max_bytes_before_external_group_by': 805306368, - 'max_bytes_before_external_sort': 805306368 - } -) }} +{{ metric_evidence_table() }} -- Resolution happens HERE, once per gold build: evidence carries BOTH keys — -- `entity_id` is the canonical person id (or '' when identity does not know diff --git a/src/ingestion/gold/collab_metric_observations.sql b/src/ingestion/gold/collab_metric_observations.sql index 44dfb0d1b..5c3cecf02 100644 --- a/src/ingestion/gold/collab_metric_observations.sql +++ b/src/ingestion/gold/collab_metric_observations.sql @@ -1,17 +1,4 @@ -{{ config( - materialized='table', - engine='MergeTree', - order_by=['source_key', 'measure_key', 'entity_id', 'metric_date'], - schema='insight', - alias='collab_metric_observations', - tags=['gold'], - query_settings={ - 'max_memory_usage': 1610612736, - 'max_threads': 4, - 'max_bytes_before_external_group_by': 805306368, - 'max_bytes_before_external_sort': 805306368 - } -) }} +{{ metric_observations_table() }} SELECT tenant_id, diff --git a/src/ingestion/gold/git_metric_evidence.sql b/src/ingestion/gold/git_metric_evidence.sql index 819bf3922..0d08703c4 100644 --- a/src/ingestion/gold/git_metric_evidence.sql +++ b/src/ingestion/gold/git_metric_evidence.sql @@ -1,18 +1,4 @@ -{{ config( - materialized='table', - engine='MergeTree', - order_by=['tenant_id', 'source_key', 'measure_key', 'entity_id', 'metric_date', 'record_id'], - schema='insight', - alias='git_metric_evidence', - tags=['gold'], - query_settings={ - 'join_use_nulls': 1, - 'max_memory_usage': 1610612736, - 'max_threads': 4, - 'max_bytes_before_external_group_by': 805306368, - 'max_bytes_before_external_sort': 805306368 - } -) }} +{{ metric_evidence_table(join_use_nulls=1) }} -- Resolution happens HERE, once per gold build: evidence carries BOTH keys — -- `entity_id` is the canonical person id (or '' when identity does not know @@ -103,7 +89,6 @@ file_changes_source AS ( file_changes.lines_removed AS lines_removed, commits.repository_value AS repository_value, commits.repository_label AS repository_label, - commits.source_dimensions AS source_dimensions, CAST( [ tuple('file_extension', file_extension, file_extension_label), @@ -239,7 +224,7 @@ pull_requests_source AS ( AND pr_commit_emails.repo_slug = prs.repo_slug AND pr_commit_emails.pr_id = prs.pr_id ), -prs_created_source AS ( +pull_request_measures AS ( SELECT tenant_id, pr_id, @@ -247,52 +232,89 @@ prs_created_source AS ( title, author_name, assumeNotNull(entity_id) AS entity_id, - toDate(created_on) AS metric_date, - created_on AS observed_at, - state, - change_size, + toDate(pr_measure.3) AS metric_date, + pr_measure.3 AS observed_at, + pr_measure.1 AS measure_key, + pr_measure.2 AS contribution, repository_label, repository_value, source_dimensions - FROM pull_requests_source - WHERE entity_id IS NOT NULL - AND entity_id != '' - AND created_on IS NOT NULL + FROM pull_requests_source AS pull_request + ARRAY JOIN CAST(arrayConcat( + if( + created_on IS NOT NULL, + [tuple('pr_created', toFloat64(1), toDateTime64(assumeNotNull(created_on), 3))], + [] + ), + if( + created_on IS NOT NULL AND state = 'MERGED', + [tuple('pr_created_merged', toFloat64(1), toDateTime64(assumeNotNull(created_on), 3))], + [] + ), + if( + created_on IS NOT NULL AND ifNull(change_size, 0) > 0, + [tuple('pr_change_size', toFloat64(ifNull(change_size, 0)), toDateTime64(assumeNotNull(created_on), 3))], + [] + ), + if( + state = 'MERGED' AND closed_on IS NOT NULL, + [tuple('pr_merged', toFloat64(1), toDateTime64(assumeNotNull(closed_on), 3))], + [] + ), + if( + cycle_hours IS NOT NULL AND closed_on IS NOT NULL, + [tuple('pr_cycle_hours', toFloat64(assumeNotNull(cycle_hours)), toDateTime64(assumeNotNull(closed_on), 3))], + [] + ) + ) AS Array(Tuple(measure_key String, contribution Float64, observed_at DateTime64(3)))) AS pr_measure + WHERE pull_request.entity_id IS NOT NULL + AND pull_request.entity_id != '' ), -prs_merged_source AS ( +file_change_measures AS ( SELECT tenant_id, - pr_id, - pr_number, - title, - author_name, - assumeNotNull(entity_id) AS entity_id, - toDate(closed_on) AS metric_date, - closed_on AS observed_at, - cycle_hours, - repository_label, - repository_value, - source_dimensions - FROM pull_requests_source - WHERE entity_id IS NOT NULL - AND entity_id != '' - AND state = 'MERGED' - AND closed_on IS NOT NULL + entity_id, + metric_date, + file_measure.1 AS measure_key, + file_measure.2 AS value, + file_measure.3 AS dimensions + FROM file_changes_source + ARRAY JOIN CAST(arrayConcat( + if( + lines_added IS NOT NULL, + [tuple('lines_added', toFloat64(assumeNotNull(lines_added)), category_source_dimensions)], + [] + ), + if( + lines_removed IS NOT NULL, + [tuple('lines_removed', toFloat64(assumeNotNull(lines_removed)), category_source_dimensions)], + [] + ), + if( + category = 'code' AND lines_added IS NOT NULL, + [tuple('code_lines_added', toFloat64(assumeNotNull(lines_added)), file_source_dimensions)], + [] + ) + ) AS Array(Tuple( + measure_key String, + value Float64, + dimensions Array(Tuple(key String, value String, label Nullable(String))) + ))) AS file_measure ), measure_observations AS ( {{ presence_measure('commit_day', ['commits_source']) }} UNION ALL - {{ sum_measure('code_lines_added', 'file_changes_source', 'lines_added', 'file_source_dimensions', where="category = 'code'") }} - - UNION ALL - - {{ sum_measure('lines_added', 'file_changes_source', 'lines_added', 'category_source_dimensions') }} - - UNION ALL - - {{ sum_measure('lines_removed', 'file_changes_source', 'lines_removed', 'category_source_dimensions') }} + SELECT + tenant_id, + entity_id, + metric_date, + measure_key, + toNullable(sum(value)) AS value, + dimensions + FROM file_change_measures + GROUP BY tenant_id, entity_id, metric_date, measure_key, dimensions ) SELECT assumeNotNull(tenant_id) AS tenant_id, @@ -368,45 +390,12 @@ SELECT assumeNotNull(entity_id) AS entity_id, assumeNotNull(metric_date) AS metric_date, toNullable(toDateTime64(observed_at, 3)) AS observed_at, - pr_measure.1 AS measure_key, - concat(repository_value, ':pr:', toString(pr_id), ':', pr_measure.1) AS record_id, - 'pull_request' AS record_kind, - 'event' AS granularity, - if(title = '', concat('PR #', toString(pr_number)), title) AS record_label, - toNullable(toFloat64(pr_measure.2)) AS contribution, - CAST(NULL AS Nullable(String)) AS subject_key, - source_dimensions AS dimensions, - map( - 'ref', toString(pr_number), - 'title', title, - 'repository', repository_label, - 'author', author_name - ) AS details -FROM prs_created_source -ARRAY JOIN arrayConcat( - [tuple('pr_created', toFloat64(1))], - if(state = 'MERGED', [tuple('pr_created_merged', toFloat64(1))], []), - if(ifNull(change_size, 0) > 0, [tuple('pr_change_size', toFloat64(change_size))], []) -) AS pr_measure -WHERE tenant_id IS NOT NULL - AND entity_id IS NOT NULL - AND metric_date IS NOT NULL - -UNION ALL - -SELECT - assumeNotNull(tenant_id) AS tenant_id, - 'git' AS source_key, - 'person' AS entity_type, - assumeNotNull(entity_id) AS entity_id, - assumeNotNull(metric_date) AS metric_date, - toNullable(toDateTime64(observed_at, 3)) AS observed_at, - pr_measure.1 AS measure_key, - concat(repository_value, ':pr:', toString(pr_id), ':', pr_measure.1) AS record_id, + measure_key, + concat(repository_value, ':pr:', toString(pr_id), ':', measure_key) AS record_id, 'pull_request' AS record_kind, 'event' AS granularity, if(title = '', concat('PR #', toString(pr_number)), title) AS record_label, - toNullable(toFloat64(pr_measure.2)) AS contribution, + toNullable(toFloat64(contribution)) AS contribution, CAST(NULL AS Nullable(String)) AS subject_key, source_dimensions AS dimensions, map( @@ -415,11 +404,7 @@ SELECT 'repository', repository_label, 'author', author_name ) AS details -FROM prs_merged_source -ARRAY JOIN arrayConcat( - [tuple('pr_merged', toFloat64(1))], - if(cycle_hours IS NOT NULL, [tuple('pr_cycle_hours', toFloat64(cycle_hours))], []) -) AS pr_measure +FROM pull_request_measures WHERE tenant_id IS NOT NULL AND entity_id IS NOT NULL AND metric_date IS NOT NULL diff --git a/src/ingestion/gold/git_metric_observations.sql b/src/ingestion/gold/git_metric_observations.sql index 182f94d8d..e2f62cb2c 100644 --- a/src/ingestion/gold/git_metric_observations.sql +++ b/src/ingestion/gold/git_metric_observations.sql @@ -1,17 +1,4 @@ -{{ config( - materialized='table', - engine='MergeTree', - order_by=['source_key', 'measure_key', 'entity_id', 'metric_date'], - schema='insight', - alias='git_metric_observations', - tags=['gold'], - query_settings={ - 'max_memory_usage': 1610612736, - 'max_threads': 4, - 'max_bytes_before_external_group_by': 805306368, - 'max_bytes_before_external_sort': 805306368 - } -) }} +{{ metric_observations_table() }} SELECT tenant_id, diff --git a/src/ingestion/gold/task_metric_evidence.sql b/src/ingestion/gold/task_metric_evidence.sql index 57e8fd23b..1b783f43f 100644 --- a/src/ingestion/gold/task_metric_evidence.sql +++ b/src/ingestion/gold/task_metric_evidence.sql @@ -1,17 +1,4 @@ -{{ config( - materialized='table', - engine='MergeTree', - order_by=['tenant_id', 'source_key', 'measure_key', 'entity_id', 'metric_date', 'record_id'], - schema='insight', - alias='task_metric_evidence', - tags=['gold'], - query_settings={ - 'max_memory_usage': 1610612736, - 'max_threads': 4, - 'max_bytes_before_external_group_by': 805306368, - 'max_bytes_before_external_sort': 805306368 - } -) }} +{{ metric_evidence_table() }} -- Resolution happens HERE, once per gold build: evidence carries BOTH keys — -- `entity_id` is the canonical person id (or '' when identity does not know diff --git a/src/ingestion/gold/task_metric_observations.sql b/src/ingestion/gold/task_metric_observations.sql index 3e7532e95..2f7258664 100644 --- a/src/ingestion/gold/task_metric_observations.sql +++ b/src/ingestion/gold/task_metric_observations.sql @@ -1,17 +1,4 @@ -{{ config( - materialized='table', - engine='MergeTree', - order_by=['source_key', 'measure_key', 'entity_id', 'metric_date'], - schema='insight', - alias='task_metric_observations', - tags=['gold'], - query_settings={ - 'max_memory_usage': 1610612736, - 'max_threads': 4, - 'max_bytes_before_external_group_by': 805306368, - 'max_bytes_before_external_sort': 805306368 - } -) }} +{{ metric_observations_table() }} SELECT tenant_id, diff --git a/src/ingestion/gold/wiki_metric_evidence.sql b/src/ingestion/gold/wiki_metric_evidence.sql index b7b789b70..42c1cca3d 100644 --- a/src/ingestion/gold/wiki_metric_evidence.sql +++ b/src/ingestion/gold/wiki_metric_evidence.sql @@ -1,17 +1,4 @@ -{{ config( - materialized='table', - engine='MergeTree', - order_by=['tenant_id', 'source_key', 'measure_key', 'entity_id', 'metric_date', 'record_id'], - schema='insight', - alias='wiki_metric_evidence', - tags=['gold'], - query_settings={ - 'max_memory_usage': 1610612736, - 'max_threads': 4, - 'max_bytes_before_external_group_by': 805306368, - 'max_bytes_before_external_sort': 805306368 - } -) }} +{{ metric_evidence_table() }} -- Resolution happens HERE, once per gold build: evidence carries BOTH keys — -- `entity_id` is the canonical person id (or '' when identity does not know diff --git a/src/ingestion/gold/wiki_metric_observations.sql b/src/ingestion/gold/wiki_metric_observations.sql index 61ae2a693..5fd9dd9d9 100644 --- a/src/ingestion/gold/wiki_metric_observations.sql +++ b/src/ingestion/gold/wiki_metric_observations.sql @@ -1,18 +1,4 @@ -{{ config( - materialized='table', - engine='MergeTree', - order_by=['source_key', 'measure_key', 'entity_id', 'metric_date'], - schema='insight', - alias='wiki_metric_observations', - tags=['gold'], - query_settings={ - 'max_memory_usage': 3221225472, - 'max_threads': 4, - 'max_bytes_before_external_group_by': 805306368, - 'max_bytes_before_external_sort': 805306368, - 'join_algorithm': 'grace_hash,hash' - } -) }} +{{ metric_observations_table() }} SELECT tenant_id, diff --git a/src/ingestion/scripts/connectors-ddl/insight.sql b/src/ingestion/scripts/connectors-ddl/insight.sql index 12b9c27de..74544b689 100644 --- a/src/ingestion/scripts/connectors-ddl/insight.sql +++ b/src/ingestion/scripts/connectors-ddl/insight.sql @@ -23,7 +23,8 @@ CREATE TABLE IF NOT EXISTS insight.ai_metric_evidence `details` Map(String, String) ) ENGINE = MergeTree -ORDER BY (tenant_id, source_key, measure_key, entity_id, metric_date, record_id) +PARTITION BY toYYYYMM(metric_date) +ORDER BY (tenant_id, source_key, entity_type, entity_id, measure_key, metric_date, record_id) SETTINGS replicated_deduplication_window = '0', index_granularity = 8192 ; @@ -44,7 +45,8 @@ CREATE TABLE IF NOT EXISTS insight.ai_metric_observations label Nullable(String))) ) ENGINE = MergeTree -ORDER BY (source_key, measure_key, entity_id, metric_date) +PARTITION BY toYYYYMM(metric_date) +ORDER BY (tenant_id, source_key, entity_type, entity_id, measure_key, metric_date) SETTINGS replicated_deduplication_window = '0', index_granularity = 8192 ; @@ -71,7 +73,8 @@ CREATE TABLE IF NOT EXISTS insight.collab_metric_evidence `details` Map(String, String) ) ENGINE = MergeTree -ORDER BY (tenant_id, source_key, measure_key, entity_id, metric_date, record_id) +PARTITION BY toYYYYMM(metric_date) +ORDER BY (tenant_id, source_key, entity_type, entity_id, measure_key, metric_date, record_id) SETTINGS replicated_deduplication_window = '0', index_granularity = 8192 ; @@ -92,7 +95,8 @@ CREATE TABLE IF NOT EXISTS insight.collab_metric_observations label Nullable(String))) ) ENGINE = MergeTree -ORDER BY (source_key, measure_key, entity_id, metric_date) +PARTITION BY toYYYYMM(metric_date) +ORDER BY (tenant_id, source_key, entity_type, entity_id, measure_key, metric_date) SETTINGS replicated_deduplication_window = '0', index_granularity = 8192 ; @@ -119,7 +123,8 @@ CREATE TABLE IF NOT EXISTS insight.git_metric_evidence `details` Map(String, String) ) ENGINE = MergeTree -ORDER BY (tenant_id, source_key, measure_key, entity_id, metric_date, record_id) +PARTITION BY toYYYYMM(metric_date) +ORDER BY (tenant_id, source_key, entity_type, entity_id, measure_key, metric_date, record_id) SETTINGS replicated_deduplication_window = '0', index_granularity = 8192 ; @@ -140,7 +145,8 @@ CREATE TABLE IF NOT EXISTS insight.git_metric_observations label Nullable(String))) ) ENGINE = MergeTree -ORDER BY (source_key, measure_key, entity_id, metric_date) +PARTITION BY toYYYYMM(metric_date) +ORDER BY (tenant_id, source_key, entity_type, entity_id, measure_key, metric_date) SETTINGS replicated_deduplication_window = '0', index_granularity = 8192 ; @@ -213,7 +219,8 @@ CREATE TABLE IF NOT EXISTS insight.task_metric_evidence `details` Map(String, String) ) ENGINE = MergeTree -ORDER BY (tenant_id, source_key, measure_key, entity_id, metric_date, record_id) +PARTITION BY toYYYYMM(metric_date) +ORDER BY (tenant_id, source_key, entity_type, entity_id, measure_key, metric_date, record_id) SETTINGS replicated_deduplication_window = '0', index_granularity = 8192 ; @@ -234,7 +241,8 @@ CREATE TABLE IF NOT EXISTS insight.task_metric_observations label Nullable(String))) ) ENGINE = MergeTree -ORDER BY (source_key, measure_key, entity_id, metric_date) +PARTITION BY toYYYYMM(metric_date) +ORDER BY (tenant_id, source_key, entity_type, entity_id, measure_key, metric_date) SETTINGS replicated_deduplication_window = '0', index_granularity = 8192 ; @@ -288,7 +296,8 @@ CREATE TABLE IF NOT EXISTS insight.wiki_metric_evidence `details` Map(String, String) ) ENGINE = MergeTree -ORDER BY (tenant_id, source_key, measure_key, entity_id, metric_date, record_id) +PARTITION BY toYYYYMM(metric_date) +ORDER BY (tenant_id, source_key, entity_type, entity_id, measure_key, metric_date, record_id) SETTINGS replicated_deduplication_window = '0', index_granularity = 8192 ; @@ -309,7 +318,8 @@ CREATE TABLE IF NOT EXISTS insight.wiki_metric_observations label Nullable(String))) ) ENGINE = MergeTree -ORDER BY (source_key, measure_key, entity_id, metric_date) +PARTITION BY toYYYYMM(metric_date) +ORDER BY (tenant_id, source_key, entity_type, entity_id, measure_key, metric_date) SETTINGS replicated_deduplication_window = '0', index_granularity = 8192 ;