diff --git a/bin/clickhouse-metrics-init b/bin/clickhouse-metrics-init
deleted file mode 100755
index d057be56c1a5..000000000000
--- a/bin/clickhouse-metrics-init
+++ /dev/null
@@ -1,5 +0,0 @@
-#!/usr/bin/env bash
-# temporary script to init metrics clickhouse schema before we've got migrations setup properly
-bin/wait-for-docker
-
-cat bin/clickhouse-metrics.sql | docker-compose exec -T clickhouse clickhouse-client
diff --git a/bin/clickhouse-metrics.sql b/bin/clickhouse-metrics.sql
deleted file mode 100644
index e11b35d0f666..000000000000
--- a/bin/clickhouse-metrics.sql
+++ /dev/null
@@ -1,384 +0,0 @@
--- temporary sql to initialise metrics tables for local development
--- will be removed once we have migrations set up
--- modelled after clickhouse-logs.sql following the same patterns
-CREATE OR REPLACE TABLE metrics1
-(
- `time_bucket` DateTime MATERIALIZED toStartOfDay(timestamp) CODEC(DoubleDelta, ZSTD(1)),
- `uuid` String CODEC(ZSTD(1)),
- `team_id` Int32 CODEC(ZSTD(1)),
- `trace_id` String CODEC(ZSTD(1)),
- `span_id` String CODEC(ZSTD(1)),
- `trace_flags` Int32 CODEC(ZSTD(1)),
- `timestamp` DateTime64(6) CODEC(DoubleDelta, ZSTD(1)),
- `observed_timestamp` DateTime64(6) CODEC(DoubleDelta, ZSTD(1)),
- `created_at` DateTime64(6) MATERIALIZED now() CODEC(DoubleDelta, ZSTD(1)),
- `service_name` LowCardinality(String) CODEC(ZSTD(1)),
- `metric_name` LowCardinality(String) CODEC(ZSTD(1)),
- `metric_type` LowCardinality(String) CODEC(ZSTD(1)),
- `value` Float64 CODEC(Gorilla, ZSTD(1)),
- `count` UInt64 DEFAULT 1 CODEC(T64, ZSTD(1)),
- `histogram_bounds` Array(Float64) CODEC(ZSTD(1)),
- `histogram_counts` Array(UInt64) CODEC(ZSTD(1)),
- `unit` LowCardinality(String) CODEC(ZSTD(1)),
- `aggregation_temporality` LowCardinality(String) CODEC(ZSTD(1)),
- `is_monotonic` Bool DEFAULT false CODEC(ZSTD(1)),
- `resource_attributes` Map(LowCardinality(String), String) CODEC(ZSTD(1)),
- `resource_fingerprint` UInt64 MATERIALIZED cityHash64(resource_attributes) CODEC(DoubleDelta, ZSTD(1)),
- `instrumentation_scope` String CODEC(ZSTD(1)),
- `attributes_map_str` Map(LowCardinality(String), String) CODEC(ZSTD(1)),
- `attributes_map_float` Map(LowCardinality(String), Float64) CODEC(ZSTD(1)),
- `time_minute` DateTime ALIAS toStartOfMinute(timestamp),
- `attributes` Map(String, String) ALIAS mapApply((k, v) -> (left(k, -5), v), attributes_map_str),
- INDEX idx_metric_name_set metric_name TYPE set(100) GRANULARITY 1,
- INDEX idx_metric_type_set metric_type TYPE set(10) GRANULARITY 1,
- INDEX idx_attributes_str_keys mapKeys(attributes_map_str) TYPE bloom_filter(0.01) GRANULARITY 1,
- INDEX idx_attributes_str_values mapValues(attributes_map_str) TYPE bloom_filter(0.001) GRANULARITY 1,
- INDEX idx_observed_minmax observed_timestamp TYPE minmax GRANULARITY 1,
- PROJECTION projection_aggregate_counts
- (
- SELECT
- team_id,
- time_bucket,
- toStartOfMinute(timestamp),
- service_name,
- metric_name,
- metric_type,
- resource_fingerprint,
- count() AS event_count,
- sum(value) AS total_value,
- min(value) AS min_value,
- max(value) AS max_value
- GROUP BY
- team_id,
- time_bucket,
- toStartOfMinute(timestamp),
- service_name,
- metric_name,
- metric_type,
- resource_fingerprint
- )
-)
-ENGINE = MergeTree
-PARTITION BY toDate(timestamp)
-PRIMARY KEY (team_id, time_bucket, service_name, metric_name, resource_fingerprint, timestamp)
-ORDER BY (team_id, time_bucket, service_name, metric_name, resource_fingerprint, timestamp)
-SETTINGS
- index_granularity_bytes = 104857600,
- index_granularity = 8192,
- ttl_only_drop_parts = 1;
-
-create or replace TABLE metrics AS metrics1 ENGINE = Distributed('posthog', 'default', 'metrics1');
-
--- Raw metrics as a TSDB series/samples split (NOT the fat one-row-per-event
--- shape logs/traces use). Mirrors posthog/clickhouse/metrics/metric_events.py
--- (Replicated in prod). metric_series stores each label set ONCE; metric_samples
--- is tiny (fingerprint + timestamp + value + trace_id), joined on
--- series_fingerprint at query time.
-CREATE OR REPLACE TABLE metric_series1
-(
- `team_id` Int32,
- `metric_name` LowCardinality(String),
- `series_fingerprint` UInt64 CODEC(DoubleDelta),
- `metric_type` LowCardinality(String),
- `unit` LowCardinality(String),
- `aggregation_temporality` LowCardinality(String),
- `is_monotonic` Bool DEFAULT false,
- `service_name` LowCardinality(String),
- `resource_attributes` Map(LowCardinality(String), String),
- `attributes` Map(LowCardinality(String), String),
- `last_seen` DateTime64(6) CODEC(DoubleDelta),
- INDEX idx_service_set service_name TYPE set(1000) GRANULARITY 1,
- INDEX idx_attr_keys mapKeys(attributes) TYPE bloom_filter(0.01) GRANULARITY 1,
- INDEX idx_attr_values mapValues(attributes) TYPE bloom_filter(0.01) GRANULARITY 1
-)
-ENGINE = ReplacingMergeTree(last_seen)
-ORDER BY (team_id, metric_name, series_fingerprint)
-TTL toDateTime(last_seen) + INTERVAL 90 DAY DELETE
-SETTINGS index_granularity = 8192;
-
-create or replace TABLE metric_series AS metric_series1 ENGINE = Distributed('posthog', 'default', 'metric_series1');
-
-CREATE OR REPLACE TABLE metric_samples1
-(
- `team_id` Int32,
- `metric_name` LowCardinality(String),
- `series_fingerprint` UInt64 CODEC(DoubleDelta),
- `timestamp` DateTime64(6) CODEC(DoubleDelta),
- `value` Float64 CODEC(Gorilla),
- `count` UInt64 DEFAULT 1,
- `histogram_bounds` Array(Float64),
- `histogram_counts` Array(UInt64),
- `trace_id` String,
- `span_id` String,
- `trace_flags` Int32,
- INDEX idx_trace_id_bf trace_id TYPE bloom_filter(0.01) GRANULARITY 1
-)
-ENGINE = MergeTree
-PARTITION BY toDate(timestamp)
-ORDER BY (team_id, metric_name, series_fingerprint, timestamp)
-TTL toDateTime(timestamp) + INTERVAL 30 DAY
-SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1;
-
-create or replace TABLE metric_samples AS metric_samples1 ENGINE = Distributed('posthog', 'default', 'metric_samples1');
-
--- Attribute discovery table (reuses logs pattern)
-create or replace table default.metric_attributes
-(
- `team_id` Int32,
- `time_bucket` DateTime64(0),
- `service_name` LowCardinality(String),
- `resource_fingerprint` UInt64 DEFAULT 0,
- `attribute_key` LowCardinality(String),
- `attribute_value` String,
- `attribute_type` LowCardinality(String),
- `attribute_count` SimpleAggregateFunction(sum, UInt64),
- INDEX idx_attribute_key attribute_key TYPE bloom_filter(0.01) GRANULARITY 1,
- INDEX idx_attribute_value attribute_value TYPE bloom_filter(0.01) GRANULARITY 1,
- INDEX idx_attribute_key_n3 attribute_key TYPE ngrambf_v1(3, 32768, 3, 0) GRANULARITY 1,
- INDEX idx_attribute_value_n3 attribute_value TYPE ngrambf_v1(3, 32768, 3, 0) GRANULARITY 1
-)
-ENGINE = AggregatingMergeTree
-PARTITION BY toDate(time_bucket)
-ORDER BY (team_id, attribute_type, time_bucket, resource_fingerprint, attribute_key, attribute_value);
-
--- MV: resource attributes -> metric_attributes
-drop view if exists metric_to_resource_attributes;
-CREATE MATERIALIZED VIEW metric_to_resource_attributes TO metric_attributes
-(
- `team_id` Int32,
- `time_bucket` DateTime64(0),
- `service_name` LowCardinality(String),
- `resource_fingerprint` UInt64,
- `attribute_key` LowCardinality(String),
- `attribute_value` String,
- `attribute_type` LowCardinality(String),
- `attribute_count` SimpleAggregateFunction(sum, UInt64)
-)
-AS SELECT
- team_id,
- time_bucket,
- service_name,
- resource_fingerprint,
- attribute_key,
- attribute_value,
- 'resource' as attribute_type,
- attribute_count
-FROM
-(
- SELECT
- team_id AS team_id,
- toStartOfInterval(timestamp, toIntervalMinute(10)) AS time_bucket,
- service_name AS service_name,
- resource_fingerprint,
- arrayJoin(resource_attributes) AS attribute,
- attribute.1 AS attribute_key,
- attribute.2 AS attribute_value,
- sumSimpleState(1) AS attribute_count
- FROM metrics1
- GROUP BY
- team_id,
- time_bucket,
- service_name,
- resource_fingerprint,
- attribute
-);
-
--- MV: metric attributes -> metric_attributes
-drop view if exists metric_to_metric_attributes;
-CREATE MATERIALIZED VIEW metric_to_metric_attributes TO metric_attributes
-(
- `team_id` Int32,
- `time_bucket` DateTime64(0),
- `service_name` LowCardinality(String),
- `resource_fingerprint` UInt64,
- `attribute_key` LowCardinality(String),
- `attribute_value` String,
- `attribute_type` LowCardinality(String),
- `attribute_count` SimpleAggregateFunction(sum, UInt64)
-)
-AS SELECT
- team_id,
- time_bucket,
- service_name,
- resource_fingerprint,
- attribute_key,
- attribute_value,
- 'metric' as attribute_type,
- attribute_count
-FROM
-(
- SELECT
- team_id AS team_id,
- toStartOfInterval(timestamp, toIntervalMinute(10)) AS time_bucket,
- service_name AS service_name,
- resource_fingerprint,
- arrayJoin(mapFilter((k, v) -> ((length(k) < 256) AND (length(v) < 256)), attributes)) AS attribute,
- attribute.1 AS attribute_key,
- attribute.2 AS attribute_value,
- sumSimpleState(1) AS attribute_count
- FROM metrics1
- GROUP BY
- team_id,
- time_bucket,
- service_name,
- resource_fingerprint,
- attribute
-);
-
--- Kafka engine table reading AVRO from clickhouse_metrics topic
-CREATE OR REPLACE TABLE kafka_metrics_avro
-(
- `uuid` String,
- `trace_id` String,
- `span_id` String,
- `trace_flags` Nullable(Int32),
- `timestamp` DateTime64(6),
- `observed_timestamp` DateTime64(6),
- `service_name` Nullable(String),
- `metric_name` Nullable(String),
- `metric_type` Nullable(String),
- `value` Nullable(Float64),
- `count` Nullable(Int64),
- `histogram_bounds` Array(Float64),
- `histogram_counts` Array(Int64),
- `unit` Nullable(String),
- `aggregation_temporality` Nullable(String),
- `is_monotonic` Nullable(UInt8),
- `resource_attributes` Map(String, String),
- `instrumentation_scope` Nullable(String),
- `attributes` Map(String, String),
- `series_fingerprint` Nullable(Int64)
-)
-ENGINE = Kafka('kafka:9092', 'clickhouse_metrics', 'clickhouse-metrics-avro', 'Avro')
-SETTINGS
- kafka_skip_broken_messages = 100,
- kafka_security_protocol = 'PLAINTEXT',
- kafka_thread_per_consumer = 1,
- kafka_num_consumers = 1,
- kafka_poll_timeout_ms=15000,
- kafka_poll_max_batch_size=10,
- kafka_max_block_size=10;
-
--- MV: kafka_metrics_avro -> metrics1
-drop table if exists kafka_metrics_avro_mv;
-
-CREATE MATERIALIZED VIEW kafka_metrics_avro_mv TO metrics1
-AS SELECT
- uuid,
- trace_id,
- span_id,
- ifNull(trace_flags, 0) as trace_flags,
- timestamp,
- observed_timestamp,
- ifNull(service_name, '') as service_name,
- ifNull(metric_name, '') as metric_name,
- ifNull(metric_type, '') as metric_type,
- ifNull(value, 0) as value,
- toUInt64(ifNull(count, 1)) as count,
- histogram_bounds,
- arrayMap(x -> toUInt64(x), histogram_counts) as histogram_counts,
- ifNull(unit, '') as unit,
- ifNull(aggregation_temporality, '') as aggregation_temporality,
- ifNull(is_monotonic, 0) as is_monotonic,
- mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), resource_attributes)) AS resource_attributes,
- ifNull(instrumentation_scope, '') as instrumentation_scope,
- mapSort(mapApply((k,v) -> (concat(k, '__str'), JSONExtractString(v)), attributes)) as attributes_map_str,
- mapSort(mapFilter((k, v) -> isNotNull(v), mapApply((k,v) -> (concat(k, '__float'), toFloat64OrNull(JSONExtract(v, 'String'))), attributes))) as attributes_map_float,
- toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) as team_id
-FROM kafka_metrics_avro settings min_insert_block_size_rows=0, min_insert_block_size_bytes=0;
-
--- Two MVs on the existing kafka_metrics_avro: every clickhouse_metrics point fans
--- into metric_series (labels, deduped) and metric_samples (the tiny row), alongside
--- kafka_metrics_avro_mv -> metrics1. series_fingerprint is assigned ONCE at ingest
--- (capture-logs) and shipped in the Avro payload; both MVs read it verbatim — they do
--- NOT recompute it. ClickHouse never computes the identity (no cityHash64 over the
--- maps), so the two tables cannot disagree and the hash cannot collapse. The Avro
--- `long` carries the u64 bits as signed; reinterpretAsUInt64 restores them.
--- Rows with a NULL series_fingerprint (a producer that predates the ingest change, or
--- a rollback) are dropped, not coerced to 0: coalescing to a shared id would collapse
--- every such series onto one ReplacingMergeTree row with arbitrary labels — silent join
--- corruption. Dropped rows are recoverable by replaying the topic once ingest is live.
-drop table if exists kafka_metrics_avro_to_metric_series;
-CREATE MATERIALIZED VIEW kafka_metrics_avro_to_metric_series TO metric_series1
-AS SELECT
- toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) as team_id,
- ifNull(metric_name, '') as metric_name,
- reinterpretAsUInt64(assumeNotNull(series_fingerprint)) as series_fingerprint,
- ifNull(metric_type, '') as metric_type,
- ifNull(unit, '') as unit,
- ifNull(aggregation_temporality, '') as aggregation_temporality,
- ifNull(is_monotonic, 0) as is_monotonic,
- ifNull(service_name, '') as service_name,
- mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), resource_attributes)) AS resource_attributes,
- mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), attributes)) AS attributes,
- timestamp as last_seen
-FROM kafka_metrics_avro
-WHERE kafka_metrics_avro.series_fingerprint IS NOT NULL
-settings min_insert_block_size_rows=0, min_insert_block_size_bytes=0;
-
-drop table if exists kafka_metrics_avro_to_metric_samples;
-CREATE MATERIALIZED VIEW kafka_metrics_avro_to_metric_samples TO metric_samples1
-AS SELECT
- toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) as team_id,
- ifNull(metric_name, '') as metric_name,
- reinterpretAsUInt64(assumeNotNull(series_fingerprint)) as series_fingerprint,
- timestamp,
- ifNull(value, 0) as value,
- toUInt64(ifNull(count, 1)) as count,
- histogram_bounds,
- arrayMap(x -> toUInt64(x), histogram_counts) as histogram_counts,
- trace_id,
- span_id,
- ifNull(trace_flags, 0) as trace_flags
-FROM kafka_metrics_avro
-WHERE kafka_metrics_avro.series_fingerprint IS NOT NULL
-settings min_insert_block_size_rows=0, min_insert_block_size_bytes=0;
-
--- Kafka consumer lag tracking
-create or replace table metrics_kafka_metrics
-(
- `_partition` UInt32,
- `_topic` String,
- `max_offset` SimpleAggregateFunction(max, UInt64),
- `max_observed_timestamp` SimpleAggregateFunction(max, DateTime64(9)),
- `max_timestamp` SimpleAggregateFunction(max, DateTime64(9)),
- `max_created_at` SimpleAggregateFunction(max, DateTime64(9)),
- `max_lag` SimpleAggregateFunction(max, UInt64)
-)
-ENGINE = MergeTree
-ORDER BY (_topic, _partition);
-
-drop view if exists kafka_metrics_avro_kafka_metrics_mv;
-CREATE MATERIALIZED VIEW kafka_metrics_avro_kafka_metrics_mv TO metrics_kafka_metrics
-AS
- SELECT
- _partition,
- _topic,
- maxSimpleState(_offset) as max_offset,
- maxSimpleState(observed_timestamp) as max_observed_timestamp,
- maxSimpleState(timestamp) as max_timestamp,
- maxSimpleState(now()) as max_created_at,
- maxSimpleState(now() - observed_timestamp) as max_lag
- FROM kafka_metrics_avro
- group by _partition, _topic;
-
--- Read aliases in the `posthog` database.
--- The product connection uses CLICKHOUSE_DATABASE=posthog and resolves the
--- bare names `metrics`, `metric_attributes`, `metrics_kafka_metrics` there,
--- while everything above lives in `default` (this script runs unqualified
--- through docker clickhouse-client). Without these aliases the product reads
--- empty same-named tables in `posthog` while data accumulates in `default`.
-DROP TABLE IF EXISTS posthog.metrics1_to_metric_attributes;
-DROP TABLE IF EXISTS posthog.metrics1_to_resource_attributes;
-DROP TABLE IF EXISTS posthog.metrics1;
-CREATE OR REPLACE TABLE posthog.metrics AS default.metrics1 ENGINE = Distributed('posthog', 'default', 'metrics1');
-DROP TABLE IF EXISTS posthog.metric_attributes;
-CREATE TABLE posthog.metric_attributes AS default.metric_attributes ENGINE = Distributed('posthog', 'default', 'metric_attributes');
-DROP TABLE IF EXISTS posthog.metrics_kafka_metrics;
-CREATE TABLE posthog.metrics_kafka_metrics AS default.metrics_kafka_metrics ENGINE = Distributed('posthog', 'default', 'metrics_kafka_metrics');
-DROP TABLE IF EXISTS posthog.metric_series;
-CREATE OR REPLACE TABLE posthog.metric_series AS default.metric_series1 ENGINE = Distributed('posthog', 'default', 'metric_series1');
-DROP TABLE IF EXISTS posthog.metric_samples;
-CREATE OR REPLACE TABLE posthog.metric_samples AS default.metric_samples1 ENGINE = Distributed('posthog', 'default', 'metric_samples1');
-
-select 'clickhouse metrics tables initialised successfully!';
diff --git a/bin/verify-metrics-pipe b/bin/verify-metrics-pipe
index 5fedf52d9e07..c3ebdaf9b7a5 100755
--- a/bin/verify-metrics-pipe
+++ b/bin/verify-metrics-pipe
@@ -59,7 +59,7 @@ Check, in order:
curl -s localhost:6743/_metrics | head # logs-ingestion
curl -s localhost:6744/_metrics | head # metrics-ingestion
4. Is capture-logs receiving? docker logs capture-logs | tail
- 5. Were the CH tables initialized? bin/clickhouse-metrics-init
+ 5. Were the CH tables created? bin/migrate (migration 0308_metrics_kafka_ingest)
6. Give it ~30s after start (15s scrape + ingest lag), then re-run.
MSG
exit 1
diff --git a/docker/clickhouse/config.d/default.xml b/docker/clickhouse/config.d/default.xml
index 235bde0d8fee..549254a8dcd2 100644
--- a/docker/clickhouse/config.d/default.xml
+++ b/docker/clickhouse/config.d/default.xml
@@ -102,6 +102,9 @@
+
+
+
diff --git a/docker/clickhouse/config.d/multinode/ai_events_node.xml b/docker/clickhouse/config.d/multinode/ai_events_node.xml
index 05191ee5791e..78a0a4c1aeb5 100644
--- a/docker/clickhouse/config.d/multinode/ai_events_node.xml
+++ b/docker/clickhouse/config.d/multinode/ai_events_node.xml
@@ -159,6 +159,9 @@
+
+
+
diff --git a/docker/clickhouse/config.d/multinode/aux_node.xml b/docker/clickhouse/config.d/multinode/aux_node.xml
index a8730d6b20b1..2ab965852166 100644
--- a/docker/clickhouse/config.d/multinode/aux_node.xml
+++ b/docker/clickhouse/config.d/multinode/aux_node.xml
@@ -159,6 +159,9 @@
+
+
+
diff --git a/docker/clickhouse/config.d/multinode/data_node.xml b/docker/clickhouse/config.d/multinode/data_node.xml
index 5024b6333fda..a76c555acaea 100644
--- a/docker/clickhouse/config.d/multinode/data_node.xml
+++ b/docker/clickhouse/config.d/multinode/data_node.xml
@@ -159,6 +159,9 @@
+
+
+
diff --git a/docker/clickhouse/config.d/multinode/ingestion_events_node.xml b/docker/clickhouse/config.d/multinode/ingestion_events_node.xml
index 9ef53cbc34da..2ecc11f465c1 100644
--- a/docker/clickhouse/config.d/multinode/ingestion_events_node.xml
+++ b/docker/clickhouse/config.d/multinode/ingestion_events_node.xml
@@ -159,6 +159,9 @@
+
+
+
diff --git a/docker/clickhouse/config.d/multinode/ingestion_medium_node.xml b/docker/clickhouse/config.d/multinode/ingestion_medium_node.xml
index ec7592466603..68fce379a2f8 100644
--- a/docker/clickhouse/config.d/multinode/ingestion_medium_node.xml
+++ b/docker/clickhouse/config.d/multinode/ingestion_medium_node.xml
@@ -159,6 +159,9 @@
+
+
+
diff --git a/docker/clickhouse/config.d/multinode/ingestion_small_node.xml b/docker/clickhouse/config.d/multinode/ingestion_small_node.xml
index dc051256f618..42db9871d47a 100644
--- a/docker/clickhouse/config.d/multinode/ingestion_small_node.xml
+++ b/docker/clickhouse/config.d/multinode/ingestion_small_node.xml
@@ -159,6 +159,9 @@
+
+
+
diff --git a/docker/clickhouse/config.d/multinode/logs_node.xml b/docker/clickhouse/config.d/multinode/logs_node.xml
index 34a81ffbb6e0..4851507f2309 100644
--- a/docker/clickhouse/config.d/multinode/logs_node.xml
+++ b/docker/clickhouse/config.d/multinode/logs_node.xml
@@ -159,6 +159,9 @@
+
+
+
diff --git a/docker/clickhouse/config.d/multinode/ops_node.xml b/docker/clickhouse/config.d/multinode/ops_node.xml
index fb22cf292f6f..4fbbb8e8bc76 100644
--- a/docker/clickhouse/config.d/multinode/ops_node.xml
+++ b/docker/clickhouse/config.d/multinode/ops_node.xml
@@ -159,6 +159,9 @@
+
+
+
diff --git a/docker/clickhouse/config.d/multinode/sessions_node.xml b/docker/clickhouse/config.d/multinode/sessions_node.xml
index 7edd307022b7..edc60226fd7a 100644
--- a/docker/clickhouse/config.d/multinode/sessions_node.xml
+++ b/docker/clickhouse/config.d/multinode/sessions_node.xml
@@ -159,6 +159,9 @@
+
+
+
diff --git a/hogli.yaml b/hogli.yaml
index 803cb42d54fa..142fd33173fe 100644
--- a/hogli.yaml
+++ b/hogli.yaml
@@ -966,10 +966,6 @@ tools:
bin_script: clickhouse-logs-init
description: Initialize ClickHouse logs schema (temporary, pre-migration)
hidden: true
- clickhouse:metrics:init:
- bin_script: clickhouse-metrics-init
- description: Initialize ClickHouse metrics schema (temporary, pre-migration)
- hidden: true
clickhouse:multinode:start:
cmd: tools/infra-scripts/clickhouse-multinode/start-multinode-clickhouse
description: Boot the multinode ClickHouse smoke-test stack (one server per logical cluster) for local migration verification
diff --git a/posthog/clickhouse/hcl/README.md b/posthog/clickhouse/hcl/README.md
index 1c5af35fe5d3..790a33205684 100644
--- a/posthog/clickhouse/hcl/README.md
+++ b/posthog/clickhouse/hcl/README.md
@@ -83,9 +83,9 @@ reconciling a new role.
Every role `manifest.hcl` composes for the gate's env is dumped and gated — for `local-multi` that is
each node the multinode stack runs, one per published port in `dump-live.sh`'s `ROLES`, compared
against its `golden/local-multi/.hcl` (`aux` is filed as `auxiliary`, see `golden_name` in
-`lib.sh`). Every logs node composes `roles/logs/{base,traces,traces_kafka_metrics}`; the local one
-adds a self-contained `roles/logs/local` (extracted from the live node) for the legacy `logs32`
-family it still runs, and skips the cloud-only metrics ingest in `roles/logs/cloud`.
+`lib.sh`). Every logs node composes `roles/logs/{base,traces,traces_kafka_metrics,metrics}`; the
+local one adds a self-contained `roles/logs/local` (extracted from the live node) for the legacy
+`logs32` family it still runs.
`node_roles` is **derived**: an object in `roles/shared/` appears in every node's composition →
`node_roles` = every role the manifest declares; an object under `roles/ops/` appears only in the ops
diff --git a/posthog/clickhouse/hcl/golden/local-multi/logs.hcl b/posthog/clickhouse/hcl/golden/local-multi/logs.hcl
index 0ff746f89c84..588003d4a873 100644
--- a/posthog/clickhouse/hcl/golden/local-multi/logs.hcl
+++ b/posthog/clickhouse/hcl/golden/local-multi/logs.hcl
@@ -1,4 +1,78 @@
database "posthog" {
+ table "kafka_metrics_avro" {
+ column "uuid" {
+ type = "String"
+ }
+ column "trace_id" {
+ type = "String"
+ }
+ column "span_id" {
+ type = "String"
+ }
+ column "trace_flags" {
+ type = "Nullable(Int32)"
+ }
+ column "timestamp" {
+ type = "DateTime64(6)"
+ }
+ column "observed_timestamp" {
+ type = "DateTime64(6)"
+ }
+ column "service_name" {
+ type = "Nullable(String)"
+ }
+ column "metric_name" {
+ type = "Nullable(String)"
+ }
+ column "metric_type" {
+ type = "Nullable(String)"
+ }
+ column "value" {
+ type = "Nullable(Float64)"
+ }
+ column "count" {
+ type = "Nullable(Int64)"
+ }
+ column "histogram_bounds" {
+ type = "Array(Float64)"
+ }
+ column "histogram_counts" {
+ type = "Array(Int64)"
+ }
+ column "unit" {
+ type = "Nullable(String)"
+ }
+ column "aggregation_temporality" {
+ type = "Nullable(String)"
+ }
+ column "is_monotonic" {
+ type = "Nullable(UInt8)"
+ }
+ column "resource_attributes" {
+ type = "Map(String, String)"
+ }
+ column "instrumentation_scope" {
+ type = "Nullable(String)"
+ }
+ column "attributes" {
+ type = "Map(String, String)"
+ }
+ column "series_fingerprint" {
+ type = "Nullable(Int64)"
+ }
+ engine "kafka" {
+ broker_list = "warpstream_metrics"
+ topic_list = "kafka_topic_list = 'clickhouse_metrics'"
+ group_name = "kafka_group_name = 'clickhouse-metrics-avro-new'"
+ format = "kafka_format = 'Avro'"
+ num_consumers = 8
+ skip_broken_messages = 100
+ poll_timeout_ms = 3000
+ poll_max_batch_size = 1000
+ thread_per_consumer = true
+ }
+ }
+
table "kafka_trace_spans_avro" {
column "uuid" {
type = "String"
@@ -1140,6 +1214,64 @@ SQL
}
}
+ table "metric_attributes" {
+ order_by = ["team_id", "attribute_type", "time_bucket", "resource_fingerprint", "attribute_key", "attribute_value"]
+ partition_by = "toDate(time_bucket)"
+ settings = {
+ deduplicate_merge_projection_mode = "drop"
+ index_granularity = "8192"
+ }
+ column "team_id" {
+ type = "Int32"
+ }
+ column "time_bucket" {
+ type = "DateTime64(0)"
+ }
+ column "service_name" {
+ type = "LowCardinality(String)"
+ }
+ column "resource_fingerprint" {
+ type = "UInt64"
+ default = "0"
+ }
+ column "attribute_key" {
+ type = "LowCardinality(String)"
+ }
+ column "attribute_value" {
+ type = "String"
+ }
+ column "attribute_count" {
+ type = "SimpleAggregateFunction(sum, UInt64)"
+ }
+ column "attribute_type" {
+ type = "LowCardinality(String)"
+ }
+ index "idx_attribute_key" {
+ expr = "attribute_key"
+ type = "bloom_filter(0.01)"
+ granularity = 1
+ }
+ index "idx_attribute_value" {
+ expr = "attribute_value"
+ type = "bloom_filter(0.01)"
+ granularity = 1
+ }
+ index "idx_attribute_key_n3" {
+ expr = "attribute_key"
+ type = "ngrambf_v1(3, 32768, 3, 0)"
+ granularity = 1
+ }
+ index "idx_attribute_value_n3" {
+ expr = "attribute_value"
+ type = "ngrambf_v1(3, 32768, 3, 0)"
+ granularity = 1
+ }
+ engine "replicated_aggregating_merge_tree" {
+ zoo_path = "/clickhouse/tables/noshard/posthog.metric_attributes"
+ replica_name = "{replica}"
+ }
+ }
+
table "metric_samples" {
column "team_id" {
type = "Int32"
@@ -1349,110 +1481,382 @@ SQL
}
}
- table "query_log_archive" {
- column "hostname" {
- type = "LowCardinality(String)"
- }
- column "user" {
- type = "LowCardinality(String)"
- }
- column "query_id" {
- type = "String"
+ table "metrics" {
+ column "time_bucket" {
+ type = "DateTime"
+ materialized = "toStartOfDay(timestamp)"
}
- column "initial_query_id" {
+ column "uuid" {
type = "String"
}
- column "is_initial_query" {
- type = "UInt8"
+ column "team_id" {
+ type = "Int32"
}
- column "type" {
- type = "Enum8('QueryStart'=1, 'QueryFinish'=2, 'ExceptionBeforeStart'=3, 'ExceptionWhileProcessing'=4)"
+ column "trace_id" {
+ type = "String"
}
- column "event_date" {
- type = "Date"
+ column "span_id" {
+ type = "String"
}
- column "event_time" {
- type = "DateTime"
+ column "trace_flags" {
+ type = "Int32"
}
- column "event_time_microseconds" {
+ column "timestamp" {
type = "DateTime64(6)"
}
- column "query_start_time" {
- type = "DateTime"
- }
- column "query_start_time_microseconds" {
+ column "observed_timestamp" {
type = "DateTime64(6)"
}
- column "query_duration_ms" {
- type = "UInt64"
+ column "created_at" {
+ type = "DateTime64(6)"
+ materialized = "now()"
}
- column "read_rows" {
- type = "UInt64"
+ column "service_name" {
+ type = "LowCardinality(String)"
}
- column "read_bytes" {
- type = "UInt64"
+ column "metric_name" {
+ type = "LowCardinality(String)"
}
- column "written_rows" {
- type = "UInt64"
+ column "metric_type" {
+ type = "LowCardinality(String)"
}
- column "written_bytes" {
- type = "UInt64"
+ column "value" {
+ type = "Float64"
+ codec = "Gorilla(8)"
}
- column "result_rows" {
- type = "UInt64"
+ column "count" {
+ type = "UInt64"
+ default = "1"
+ codec = "T64"
}
- column "result_bytes" {
- type = "UInt64"
+ column "histogram_bounds" {
+ type = "Array(Float64)"
}
- column "memory_usage" {
- type = "UInt64"
+ column "histogram_counts" {
+ type = "Array(UInt64)"
}
- column "peak_threads_usage" {
- type = "UInt64"
+ column "unit" {
+ type = "LowCardinality(String)"
}
- column "current_database" {
+ column "aggregation_temporality" {
type = "LowCardinality(String)"
}
- column "query" {
- type = "String"
+ column "is_monotonic" {
+ type = "Bool"
+ default = "false"
}
- column "formatted_query" {
+ column "resource_attributes" {
+ type = "Map(LowCardinality(String), String)"
+ }
+ column "resource_fingerprint" {
+ type = "UInt64"
+ materialized = "cityHash64(resource_attributes)"
+ }
+ column "instrumentation_scope" {
type = "String"
}
- column "normalized_query_hash" {
- type = "UInt64"
+ column "attributes_map_str" {
+ type = "Map(LowCardinality(String), String)"
}
- column "query_kind" {
- type = "LowCardinality(String)"
+ column "attributes_map_float" {
+ type = "Map(LowCardinality(String), Float64)"
}
- column "exception_code" {
- type = "Int32"
+ column "time_minute" {
+ type = "DateTime"
+ alias = "toStartOfMinute(timestamp)"
}
- column "exception" {
- type = "String"
+ column "attributes" {
+ type = "Map(String, String)"
+ alias = "mapApply((k, v) -> (left(k, -5), v), attributes_map_str)"
}
- column "stack_trace" {
+ engine "distributed" {
+ cluster_name = "posthog_single_shard"
+ remote_database = "posthog"
+ remote_table = "metrics1"
+ }
+ }
+
+ table "metrics1" {
+ order_by = ["team_id", "time_bucket", "service_name", "metric_name", "resource_fingerprint", "timestamp"]
+ partition_by = "toDate(timestamp)"
+ settings = {
+ index_granularity = "8192"
+ index_granularity_bytes = "104857600"
+ ttl_only_drop_parts = "1"
+ }
+ column "time_bucket" {
+ type = "DateTime"
+ materialized = "toStartOfDay(timestamp)"
+ }
+ column "uuid" {
type = "String"
}
column "team_id" {
- type = "Int64"
+ type = "Int32"
}
- column "log_comment" {
- type = "JSON(max_dynamic_paths=256, access_method LowCardinality(String), alert_config_id String, api_key_label String, api_key_mask String, batch_export_id String, chargeable Bool, client_query_id String, cohort_id Int64, `dagster.job_name` String, `dagster.run_id` String, `dagster.tags.owner` String, dashboard_id Int64, experiment_feature_flag_key String, experiment_id Int64, feature LowCardinality(String), id String, insight_id Int64, is_impersonated Bool, kind LowCardinality(String), name String, org_id String, person_on_events_mode LowCardinality(String), product LowCardinality(String), query_type LowCardinality(String), request_name String, route_id String, service_name String, session_id String, table_id String, team_id Int64, `temporal.activity_id` String, `temporal.activity_type` String, `temporal.attempt` Int64, `temporal.workflow_id` String, `temporal.workflow_namespace` String, `temporal.workflow_run_id` String, `temporal.workflow_type` String, user_id Int64, warehouse_query Bool, workflow LowCardinality(String), workload LowCardinality(String), SKIP cache_key, SKIP filter, SKIP hogql_features, SKIP http_referer, SKIP http_request_id, SKIP http_user_agent, SKIP query_settings, SKIP timings, SKIP user_email)"
+ column "trace_id" {
+ type = "String"
}
- column "ProfileEvents" {
- type = "Map(String, UInt64)"
+ column "span_id" {
+ type = "String"
}
- column "exception_name" {
- type = "String"
- alias = "errorCodeToName(exception_code)"
+ column "trace_flags" {
+ type = "Int32"
}
- column "ProfileEvents_RealTimeMicroseconds" {
- type = "Int64"
- alias = "ProfileEvents['RealTimeMicroseconds']"
+ column "timestamp" {
+ type = "DateTime64(6)"
}
- column "ProfileEvents_OSCPUVirtualTimeMicroseconds" {
- type = "Int64"
+ column "observed_timestamp" {
+ type = "DateTime64(6)"
+ }
+ column "created_at" {
+ type = "DateTime64(6)"
+ materialized = "now()"
+ }
+ column "service_name" {
+ type = "LowCardinality(String)"
+ }
+ column "metric_name" {
+ type = "LowCardinality(String)"
+ }
+ column "metric_type" {
+ type = "LowCardinality(String)"
+ }
+ column "value" {
+ type = "Float64"
+ codec = "Gorilla(8)"
+ }
+ column "count" {
+ type = "UInt64"
+ default = "1"
+ codec = "T64"
+ }
+ column "histogram_bounds" {
+ type = "Array(Float64)"
+ }
+ column "histogram_counts" {
+ type = "Array(UInt64)"
+ }
+ column "unit" {
+ type = "LowCardinality(String)"
+ }
+ column "aggregation_temporality" {
+ type = "LowCardinality(String)"
+ }
+ column "is_monotonic" {
+ type = "Bool"
+ default = "false"
+ }
+ column "resource_attributes" {
+ type = "Map(LowCardinality(String), String)"
+ }
+ column "resource_fingerprint" {
+ type = "UInt64"
+ materialized = "cityHash64(resource_attributes)"
+ }
+ column "instrumentation_scope" {
+ type = "String"
+ }
+ column "attributes_map_str" {
+ type = "Map(LowCardinality(String), String)"
+ }
+ column "attributes_map_float" {
+ type = "Map(LowCardinality(String), Float64)"
+ }
+ column "time_minute" {
+ type = "DateTime"
+ alias = "toStartOfMinute(timestamp)"
+ }
+ column "attributes" {
+ type = "Map(String, String)"
+ alias = "mapApply((k, v) -> (left(k, -5), v), attributes_map_str)"
+ }
+ index "idx_metric_name_set" {
+ expr = "metric_name"
+ type = "set(100)"
+ granularity = 1
+ }
+ index "idx_metric_type_set" {
+ expr = "metric_type"
+ type = "set(10)"
+ granularity = 1
+ }
+ index "idx_attributes_str_keys" {
+ expr = "mapKeys(attributes_map_str)"
+ type = "bloom_filter(0.01)"
+ granularity = 1
+ }
+ index "idx_attributes_str_values" {
+ expr = "mapValues(attributes_map_str)"
+ type = "bloom_filter(0.001)"
+ granularity = 1
+ }
+ index "idx_observed_minmax" {
+ expr = "observed_timestamp"
+ type = "minmax"
+ granularity = 1
+ }
+ projection "projection_aggregate_counts" {
+ query = < (concat(k, '__str'), JSONExtractString(v)), attributes)) AS attributes_map_str,
- mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), resource_attributes)) AS resource_attributes,
- toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) AS team_id,
- observed_timestamp
- + toIntervalDay(
- toInt32OrDefault(_headers.value[indexOf(_headers.name, 'retention-days')], toInt32(15))
- ) AS original_expiry_timestamp,
_partition,
_topic,
- _offset,
- toInt64OrDefault(_headers.value[indexOf(_headers.name, 'record_count')], toInt64(1)) AS _record_count,
- toInt64OrDefault(_headers.value[indexOf(_headers.name, 'bytes_uncompressed')], toInt64(0)) AS _bytes_uncompressed,
- toInt64OrDefault(_headers.value[indexOf(_headers.name, 'bytes_compressed')], toInt64(0)) AS _bytes_compressed
-FROM posthog.kafka_trace_spans_avro
-SQL
-
- column "uuid" {
- type = "String"
+ maxSimpleState(_offset) AS max_offset,
+ maxSimpleState(observed_timestamp) AS max_observed_timestamp,
+ maxSimpleState(timestamp) AS max_timestamp,
+ maxSimpleState(now()) AS max_created_at,
+ maxSimpleState(now() - observed_timestamp) AS max_lag
+FROM posthog.kafka_metrics_avro
+GROUP BY
+ _partition, _topic
+SQL
+
+ column "_partition" {
+ type = "UInt64"
+ }
+ column "_topic" {
+ type = "LowCardinality(String)"
+ }
+ column "max_offset" {
+ type = "SimpleAggregateFunction(max, UInt64)"
+ }
+ column "max_observed_timestamp" {
+ type = "SimpleAggregateFunction(max, DateTime64(6))"
+ }
+ column "max_timestamp" {
+ type = "SimpleAggregateFunction(max, DateTime64(6))"
+ }
+ column "max_created_at" {
+ type = "SimpleAggregateFunction(max, DateTime)"
+ }
+ column "max_lag" {
+ type = "SimpleAggregateFunction(max, Decimal(18, 6))"
+ }
+ }
+
+ materialized_view "kafka_metrics_avro_mv" {
+ to_table = "posthog.metrics1"
+ query = < toUInt64(x), histogram_counts) AS histogram_counts,
+ ifNull(unit, '') AS unit,
+ ifNull(aggregation_temporality, '') AS aggregation_temporality,
+ ifNull(is_monotonic, 0) AS is_monotonic,
+ mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), resource_attributes)) AS resource_attributes,
+ ifNull(instrumentation_scope, '') AS instrumentation_scope,
+ mapSort(mapApply((k, v) -> (concat(k, '__str'), JSONExtractString(v)), attributes)) AS attributes_map_str,
+ mapSort(
+ mapFilter(
+ (k, v) -> isNotNull(v),
+ mapApply(
+ (k, v) -> (concat(k, '__float'), toFloat64OrNull(JSONExtract(v, 'String'))),
+ attributes
+ )
+ )
+ ) AS attributes_map_float,
+ toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) AS team_id
+FROM posthog.kafka_metrics_avro
+SETTINGS
+ min_insert_block_size_rows = 0,
+ min_insert_block_size_bytes = 0
+SQL
+
+ column "uuid" {
+ type = "String"
+ }
+ column "trace_id" {
+ type = "String"
+ }
+ column "span_id" {
+ type = "String"
+ }
+ column "trace_flags" {
+ type = "Int32"
+ }
+ column "timestamp" {
+ type = "DateTime64(6)"
+ }
+ column "observed_timestamp" {
+ type = "DateTime64(6)"
+ }
+ column "service_name" {
+ type = "String"
+ }
+ column "metric_name" {
+ type = "String"
+ }
+ column "metric_type" {
+ type = "String"
+ }
+ column "value" {
+ type = "Float64"
+ }
+ column "count" {
+ type = "UInt64"
+ }
+ column "histogram_bounds" {
+ type = "Array(Float64)"
+ }
+ column "histogram_counts" {
+ type = "Array(UInt64)"
+ }
+ column "unit" {
+ type = "String"
+ }
+ column "aggregation_temporality" {
+ type = "String"
+ }
+ column "is_monotonic" {
+ type = "UInt8"
+ }
+ column "resource_attributes" {
+ type = "Map(String, String)"
+ }
+ column "instrumentation_scope" {
+ type = "String"
+ }
+ column "attributes_map_str" {
+ type = "Map(String, String)"
+ }
+ column "attributes_map_float" {
+ type = "Map(String, Nullable(Float64))"
+ }
+ column "team_id" {
+ type = "Int32"
+ }
+ }
+
+ materialized_view "kafka_metrics_avro_to_metric_samples" {
+ to_table = "posthog.metric_samples1"
+ query = < toUInt64(x), histogram_counts) AS histogram_counts,
+ trace_id,
+ span_id,
+ ifNull(trace_flags, 0) AS trace_flags
+FROM posthog.kafka_metrics_avro
+WHERE kafka_metrics_avro.series_fingerprint IS NOT NULL
+SETTINGS
+ min_insert_block_size_rows = 0,
+ min_insert_block_size_bytes = 0
+SQL
+
+ column "team_id" {
+ type = "Int32"
+ }
+ column "metric_name" {
+ type = "String"
+ }
+ column "series_fingerprint" {
+ type = "UInt64"
+ }
+ column "timestamp" {
+ type = "DateTime64(6)"
+ }
+ column "value" {
+ type = "Float64"
+ }
+ column "count" {
+ type = "UInt64"
+ }
+ column "histogram_bounds" {
+ type = "Array(Float64)"
+ }
+ column "histogram_counts" {
+ type = "Array(UInt64)"
+ }
+ column "trace_id" {
+ type = "String"
+ }
+ column "span_id" {
+ type = "String"
+ }
+ column "trace_flags" {
+ type = "Int32"
+ }
+ }
+
+ materialized_view "kafka_metrics_avro_to_metric_series" {
+ to_table = "posthog.metric_series1"
+ query = < (k, JSONExtractString(v)), resource_attributes)) AS resource_attributes,
+ mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), attributes)) AS attributes,
+ timestamp AS last_seen
+FROM posthog.kafka_metrics_avro
+WHERE kafka_metrics_avro.series_fingerprint IS NOT NULL
+SETTINGS
+ min_insert_block_size_rows = 0,
+ min_insert_block_size_bytes = 0
+SQL
+
+ column "team_id" {
+ type = "Int32"
+ }
+ column "metric_name" {
+ type = "String"
+ }
+ column "series_fingerprint" {
+ type = "UInt64"
+ }
+ column "metric_type" {
+ type = "String"
+ }
+ column "unit" {
+ type = "String"
+ }
+ column "aggregation_temporality" {
+ type = "String"
+ }
+ column "is_monotonic" {
+ type = "UInt8"
+ }
+ column "service_name" {
+ type = "String"
+ }
+ column "resource_attributes" {
+ type = "Map(String, String)"
+ }
+ column "attributes" {
+ type = "Map(String, String)"
+ }
+ column "last_seen" {
+ type = "DateTime64(6)"
+ }
+ }
+
+ materialized_view "kafka_trace_spans_avro_mv" {
+ to_table = "posthog.trace_spans"
+ query = < (concat(k, '__str'), JSONExtractString(v)), attributes)) AS attributes_map_str,
+ mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), resource_attributes)) AS resource_attributes,
+ toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) AS team_id,
+ observed_timestamp
+ + toIntervalDay(
+ toInt32OrDefault(_headers.value[indexOf(_headers.name, 'retention-days')], toInt32(15))
+ ) AS original_expiry_timestamp,
+ _partition,
+ _topic,
+ _offset,
+ toInt64OrDefault(_headers.value[indexOf(_headers.name, 'record_count')], toInt64(1)) AS _record_count,
+ toInt64OrDefault(_headers.value[indexOf(_headers.name, 'bytes_uncompressed')], toInt64(0)) AS _bytes_uncompressed,
+ toInt64OrDefault(_headers.value[indexOf(_headers.name, 'bytes_compressed')], toInt64(0)) AS _bytes_compressed
+FROM posthog.kafka_trace_spans_avro
+SQL
+
+ column "uuid" {
+ type = "String"
}
column "trace_id" {
type = "String"
@@ -2851,6 +3512,119 @@ SQL
}
}
+ materialized_view "metrics1_to_metric_attributes" {
+ to_table = "posthog.metric_attributes"
+ query = < ((length(k) < 256) AND (length(v) < 256)), attributes) AS attributes,
+ arrayJoin(attributes) AS attribute,
+ 'metric' AS attribute_type,
+ attribute.1 AS attribute_key,
+ attribute.2 AS attribute_value,
+ sumSimpleState(1) AS attribute_count
+ FROM posthog.metrics1
+ GROUP BY
+ team_id, time_bucket, service_name, resource_fingerprint, attributes
+ )
+SQL
+
+ column "team_id" {
+ type = "Int32"
+ }
+ column "time_bucket" {
+ type = "DateTime64(0)"
+ }
+ column "service_name" {
+ type = "LowCardinality(String)"
+ }
+ column "resource_fingerprint" {
+ type = "UInt64"
+ }
+ column "attribute_key" {
+ type = "LowCardinality(String)"
+ }
+ column "attribute_value" {
+ type = "String"
+ }
+ column "attribute_type" {
+ type = "LowCardinality(String)"
+ }
+ column "attribute_count" {
+ type = "SimpleAggregateFunction(sum, UInt64)"
+ }
+ }
+
+ materialized_view "metrics1_to_resource_attributes" {
+ to_table = "posthog.metric_attributes"
+ query = < (concat(k, '__str'), JSONExtractString(v)), attributes)) AS attributes_map_str,
- mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), resource_attributes)) AS resource_attributes,
- toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) AS team_id,
- observed_timestamp
- + toIntervalDay(
- toInt32OrDefault(_headers.value[indexOf(_headers.name, 'retention-days')], toInt32(15))
- ) AS original_expiry_timestamp,
_partition,
_topic,
- _offset,
- toInt64OrDefault(_headers.value[indexOf(_headers.name, 'record_count')], toInt64(1)) AS _record_count,
- toInt64OrDefault(_headers.value[indexOf(_headers.name, 'bytes_uncompressed')], toInt64(0)) AS _bytes_uncompressed,
- toInt64OrDefault(_headers.value[indexOf(_headers.name, 'bytes_compressed')], toInt64(0)) AS _bytes_compressed
-FROM posthog.kafka_trace_spans_avro
+ maxSimpleState(_offset) AS max_offset,
+ maxSimpleState(observed_timestamp) AS max_observed_timestamp,
+ maxSimpleState(timestamp) AS max_timestamp,
+ maxSimpleState(now()) AS max_created_at,
+ maxSimpleState(now() - observed_timestamp) AS max_lag
+FROM posthog.kafka_metrics_avro
+GROUP BY
+ _partition, _topic
SQL
- column "uuid" {
- type = "String"
- }
- column "trace_id" {
- type = "String"
+ column "_partition" {
+ type = "UInt64"
}
- column "span_id" {
- type = "String"
+ column "_topic" {
+ type = "LowCardinality(String)"
}
- column "parent_span_id" {
- type = "String"
+ column "max_offset" {
+ type = "SimpleAggregateFunction(max, UInt64)"
+ }
+ column "max_observed_timestamp" {
+ type = "SimpleAggregateFunction(max, DateTime64(6))"
+ }
+ column "max_timestamp" {
+ type = "SimpleAggregateFunction(max, DateTime64(6))"
+ }
+ column "max_created_at" {
+ type = "SimpleAggregateFunction(max, DateTime)"
+ }
+ column "max_lag" {
+ type = "SimpleAggregateFunction(max, Decimal(18, 6))"
+ }
+ }
+
+ materialized_view "kafka_metrics_avro_mv" {
+ to_table = "posthog.metrics1"
+ query = < toUInt64(x), histogram_counts) AS histogram_counts,
+ ifNull(unit, '') AS unit,
+ ifNull(aggregation_temporality, '') AS aggregation_temporality,
+ ifNull(is_monotonic, 0) AS is_monotonic,
+ mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), resource_attributes)) AS resource_attributes,
+ ifNull(instrumentation_scope, '') AS instrumentation_scope,
+ mapSort(mapApply((k, v) -> (concat(k, '__str'), JSONExtractString(v)), attributes)) AS attributes_map_str,
+ mapSort(
+ mapFilter(
+ (k, v) -> isNotNull(v),
+ mapApply(
+ (k, v) -> (concat(k, '__float'), toFloat64OrNull(JSONExtract(v, 'String'))),
+ attributes
+ )
+ )
+ ) AS attributes_map_float,
+ toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) AS team_id
+FROM posthog.kafka_metrics_avro
+SETTINGS
+ min_insert_block_size_rows = 0,
+ min_insert_block_size_bytes = 0
+SQL
+
+ column "uuid" {
+ type = "String"
+ }
+ column "trace_id" {
+ type = "String"
+ }
+ column "span_id" {
+ type = "String"
+ }
+ column "trace_flags" {
+ type = "Int32"
+ }
+ column "timestamp" {
+ type = "DateTime64(6)"
+ }
+ column "observed_timestamp" {
+ type = "DateTime64(6)"
+ }
+ column "service_name" {
+ type = "String"
+ }
+ column "metric_name" {
+ type = "String"
+ }
+ column "metric_type" {
+ type = "String"
+ }
+ column "value" {
+ type = "Float64"
+ }
+ column "count" {
+ type = "UInt64"
+ }
+ column "histogram_bounds" {
+ type = "Array(Float64)"
+ }
+ column "histogram_counts" {
+ type = "Array(UInt64)"
+ }
+ column "unit" {
+ type = "String"
+ }
+ column "aggregation_temporality" {
+ type = "String"
+ }
+ column "is_monotonic" {
+ type = "UInt8"
+ }
+ column "resource_attributes" {
+ type = "Map(String, String)"
+ }
+ column "instrumentation_scope" {
+ type = "String"
+ }
+ column "attributes_map_str" {
+ type = "Map(String, String)"
+ }
+ column "attributes_map_float" {
+ type = "Map(String, Nullable(Float64))"
+ }
+ column "team_id" {
+ type = "Int32"
+ }
+ }
+
+ materialized_view "kafka_metrics_avro_to_metric_samples" {
+ to_table = "posthog.metric_samples1"
+ query = < toUInt64(x), histogram_counts) AS histogram_counts,
+ trace_id,
+ span_id,
+ ifNull(trace_flags, 0) AS trace_flags
+FROM posthog.kafka_metrics_avro
+WHERE kafka_metrics_avro.series_fingerprint IS NOT NULL
+SETTINGS
+ min_insert_block_size_rows = 0,
+ min_insert_block_size_bytes = 0
+SQL
+
+ column "team_id" {
+ type = "Int32"
+ }
+ column "metric_name" {
+ type = "String"
+ }
+ column "series_fingerprint" {
+ type = "UInt64"
+ }
+ column "timestamp" {
+ type = "DateTime64(6)"
+ }
+ column "value" {
+ type = "Float64"
+ }
+ column "count" {
+ type = "UInt64"
+ }
+ column "histogram_bounds" {
+ type = "Array(Float64)"
+ }
+ column "histogram_counts" {
+ type = "Array(UInt64)"
+ }
+ column "trace_id" {
+ type = "String"
+ }
+ column "span_id" {
+ type = "String"
+ }
+ column "trace_flags" {
+ type = "Int32"
+ }
+ }
+
+ materialized_view "kafka_metrics_avro_to_metric_series" {
+ to_table = "posthog.metric_series1"
+ query = < (k, JSONExtractString(v)), resource_attributes)) AS resource_attributes,
+ mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), attributes)) AS attributes,
+ timestamp AS last_seen
+FROM posthog.kafka_metrics_avro
+WHERE kafka_metrics_avro.series_fingerprint IS NOT NULL
+SETTINGS
+ min_insert_block_size_rows = 0,
+ min_insert_block_size_bytes = 0
+SQL
+
+ column "team_id" {
+ type = "Int32"
+ }
+ column "metric_name" {
+ type = "String"
+ }
+ column "series_fingerprint" {
+ type = "UInt64"
+ }
+ column "metric_type" {
+ type = "String"
+ }
+ column "unit" {
+ type = "String"
+ }
+ column "aggregation_temporality" {
+ type = "String"
+ }
+ column "is_monotonic" {
+ type = "UInt8"
+ }
+ column "service_name" {
+ type = "String"
+ }
+ column "resource_attributes" {
+ type = "Map(String, String)"
+ }
+ column "attributes" {
+ type = "Map(String, String)"
+ }
+ column "last_seen" {
+ type = "DateTime64(6)"
+ }
+ }
+
+ materialized_view "kafka_trace_spans_avro_mv" {
+ to_table = "posthog.trace_spans"
+ query = < (concat(k, '__str'), JSONExtractString(v)), attributes)) AS attributes_map_str,
+ mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), resource_attributes)) AS resource_attributes,
+ toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) AS team_id,
+ observed_timestamp
+ + toIntervalDay(
+ toInt32OrDefault(_headers.value[indexOf(_headers.name, 'retention-days')], toInt32(15))
+ ) AS original_expiry_timestamp,
+ _partition,
+ _topic,
+ _offset,
+ toInt64OrDefault(_headers.value[indexOf(_headers.name, 'record_count')], toInt64(1)) AS _record_count,
+ toInt64OrDefault(_headers.value[indexOf(_headers.name, 'bytes_uncompressed')], toInt64(0)) AS _bytes_uncompressed,
+ toInt64OrDefault(_headers.value[indexOf(_headers.name, 'bytes_compressed')], toInt64(0)) AS _bytes_compressed
+FROM posthog.kafka_trace_spans_avro
+SQL
+
+ column "uuid" {
+ type = "String"
+ }
+ column "trace_id" {
+ type = "String"
+ }
+ column "span_id" {
+ type = "String"
+ }
+ column "parent_span_id" {
+ type = "String"
}
column "trace_state" {
type = "String"
@@ -19958,6 +20619,119 @@ SQL
}
}
+ materialized_view "metrics1_to_metric_attributes" {
+ to_table = "posthog.metric_attributes"
+ query = < ((length(k) < 256) AND (length(v) < 256)), attributes) AS attributes,
+ arrayJoin(attributes) AS attribute,
+ 'metric' AS attribute_type,
+ attribute.1 AS attribute_key,
+ attribute.2 AS attribute_value,
+ sumSimpleState(1) AS attribute_count
+ FROM posthog.metrics1
+ GROUP BY
+ team_id, time_bucket, service_name, resource_fingerprint, attributes
+ )
+SQL
+
+ column "team_id" {
+ type = "Int32"
+ }
+ column "time_bucket" {
+ type = "DateTime64(0)"
+ }
+ column "service_name" {
+ type = "LowCardinality(String)"
+ }
+ column "resource_fingerprint" {
+ type = "UInt64"
+ }
+ column "attribute_key" {
+ type = "LowCardinality(String)"
+ }
+ column "attribute_value" {
+ type = "String"
+ }
+ column "attribute_type" {
+ type = "LowCardinality(String)"
+ }
+ column "attribute_count" {
+ type = "SimpleAggregateFunction(sum, UInt64)"
+ }
+ }
+
+ materialized_view "metrics1_to_resource_attributes" {
+ to_table = "posthog.metric_attributes"
+ query = < (left(k, -5), v), attributes_map_str),
+ INDEX idx_metric_name_set metric_name TYPE set(100) GRANULARITY 1,
+ INDEX idx_metric_type_set metric_type TYPE set(10) GRANULARITY 1,
+ INDEX idx_attributes_str_keys mapKeys(attributes_map_str) TYPE bloom_filter(0.01) GRANULARITY 1,
+ INDEX idx_attributes_str_values mapValues(attributes_map_str) TYPE bloom_filter(0.001) GRANULARITY 1,
+ INDEX idx_observed_minmax observed_timestamp TYPE minmax GRANULARITY 1,
+ PROJECTION projection_aggregate_counts (SELECT
+ team_id,
+ time_bucket,
+ toStartOfMinute(timestamp),
+ service_name,
+ metric_name,
+ metric_type,
+ resource_fingerprint,
+ count() AS event_count,
+ sum(value) AS total_value,
+ min(value) AS min_value,
+ max(value) AS max_value
+GROUP BY
+ team_id, time_bucket, toStartOfMinute(timestamp), service_name, metric_name, metric_type, resource_fingerprint)
+) ENGINE = ReplicatedMergeTree('/clickhouse/tables/noshard/posthog.metrics1', '{replica}') ORDER BY (team_id, time_bucket, service_name, metric_name, resource_fingerprint, timestamp) PARTITION BY toDate(timestamp) SETTINGS index_granularity = 8192, index_granularity_bytes = 104857600, ttl_only_drop_parts = 1;
+CREATE TABLE posthog.metrics_kafka_metrics (
+ _partition UInt32,
+ _topic String,
+ max_offset SimpleAggregateFunction(max, UInt64),
+ max_observed_timestamp SimpleAggregateFunction(max, DateTime64(9)),
+ max_timestamp SimpleAggregateFunction(max, DateTime64(9)),
+ max_created_at SimpleAggregateFunction(max, DateTime64(9)),
+ max_lag SimpleAggregateFunction(max, UInt64)
+) ENGINE = ReplicatedAggregatingMergeTree('/clickhouse/tables/noshard/posthog.metrics_kafka_metrics', '{replica}') ORDER BY (_topic, _partition) SETTINGS index_granularity = 8192;
CREATE TABLE posthog.query_log_archive (
hostname LowCardinality(String),
user LowCardinality(String),
@@ -580,6 +672,85 @@ CREATE MATERIALIZED VIEW posthog.kafka_logs_avro_kafka_metrics_mv TO posthog.log
FROM posthog.logs34
GROUP BY
_partition, _topic;
+CREATE MATERIALIZED VIEW posthog.kafka_metrics_avro_kafka_metrics_mv TO posthog.metrics_kafka_metrics (_partition UInt64, _topic LowCardinality(String), max_offset SimpleAggregateFunction(max, UInt64), max_observed_timestamp SimpleAggregateFunction(max, DateTime64(6)), max_timestamp SimpleAggregateFunction(max, DateTime64(6)), max_created_at SimpleAggregateFunction(max, DateTime), max_lag SimpleAggregateFunction(max, Decimal(18, 6))) AS SELECT
+ _partition,
+ _topic,
+ maxSimpleState(_offset) AS max_offset,
+ maxSimpleState(observed_timestamp) AS max_observed_timestamp,
+ maxSimpleState(timestamp) AS max_timestamp,
+ maxSimpleState(now()) AS max_created_at,
+ maxSimpleState(now() - observed_timestamp) AS max_lag
+FROM posthog.kafka_metrics_avro
+GROUP BY
+ _partition, _topic;
+CREATE MATERIALIZED VIEW posthog.kafka_metrics_avro_mv TO posthog.metrics1 (uuid String, trace_id String, span_id String, trace_flags Int32, timestamp DateTime64(6), observed_timestamp DateTime64(6), service_name String, metric_name String, metric_type String, value Float64, count UInt64, histogram_bounds Array(Float64), histogram_counts Array(UInt64), unit String, aggregation_temporality String, is_monotonic UInt8, resource_attributes Map(String, String), instrumentation_scope String, attributes_map_str Map(String, String), attributes_map_float Map(String, Nullable(Float64)), team_id Int32) AS SELECT
+ uuid,
+ trace_id,
+ span_id,
+ ifNull(trace_flags, 0) AS trace_flags,
+ timestamp,
+ observed_timestamp,
+ ifNull(service_name, '') AS service_name,
+ ifNull(metric_name, '') AS metric_name,
+ ifNull(metric_type, '') AS metric_type,
+ ifNull(value, 0) AS value,
+ toUInt64(ifNull(count, 1)) AS count,
+ histogram_bounds,
+ arrayMap(x -> toUInt64(x), histogram_counts) AS histogram_counts,
+ ifNull(unit, '') AS unit,
+ ifNull(aggregation_temporality, '') AS aggregation_temporality,
+ ifNull(is_monotonic, 0) AS is_monotonic,
+ mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), resource_attributes)) AS resource_attributes,
+ ifNull(instrumentation_scope, '') AS instrumentation_scope,
+ mapSort(mapApply((k, v) -> (concat(k, '__str'), JSONExtractString(v)), attributes)) AS attributes_map_str,
+ mapSort(
+ mapFilter(
+ (k, v) -> isNotNull(v),
+ mapApply(
+ (k, v) -> (concat(k, '__float'), toFloat64OrNull(JSONExtract(v, 'String'))),
+ attributes
+ )
+ )
+ ) AS attributes_map_float,
+ toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) AS team_id
+FROM posthog.kafka_metrics_avro
+SETTINGS
+ min_insert_block_size_rows = 0,
+ min_insert_block_size_bytes = 0;
+CREATE MATERIALIZED VIEW posthog.kafka_metrics_avro_to_metric_samples TO posthog.metric_samples1 (team_id Int32, metric_name String, series_fingerprint UInt64, timestamp DateTime64(6), value Float64, count UInt64, histogram_bounds Array(Float64), histogram_counts Array(UInt64), trace_id String, span_id String, trace_flags Int32) AS SELECT
+ toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) AS team_id,
+ ifNull(metric_name, '') AS metric_name,
+ reinterpretAsUInt64(assumeNotNull(series_fingerprint)) AS series_fingerprint,
+ timestamp,
+ ifNull(value, 0) AS value,
+ toUInt64(ifNull(count, 1)) AS count,
+ histogram_bounds,
+ arrayMap(x -> toUInt64(x), histogram_counts) AS histogram_counts,
+ trace_id,
+ span_id,
+ ifNull(trace_flags, 0) AS trace_flags
+FROM posthog.kafka_metrics_avro
+WHERE kafka_metrics_avro.series_fingerprint IS NOT NULL
+SETTINGS
+ min_insert_block_size_rows = 0,
+ min_insert_block_size_bytes = 0;
+CREATE MATERIALIZED VIEW posthog.kafka_metrics_avro_to_metric_series TO posthog.metric_series1 (team_id Int32, metric_name String, series_fingerprint UInt64, metric_type String, unit String, aggregation_temporality String, is_monotonic UInt8, service_name String, resource_attributes Map(String, String), attributes Map(String, String), last_seen DateTime64(6)) AS SELECT
+ toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) AS team_id,
+ ifNull(metric_name, '') AS metric_name,
+ reinterpretAsUInt64(assumeNotNull(series_fingerprint)) AS series_fingerprint,
+ ifNull(metric_type, '') AS metric_type,
+ ifNull(unit, '') AS unit,
+ ifNull(aggregation_temporality, '') AS aggregation_temporality,
+ ifNull(is_monotonic, 0) AS is_monotonic,
+ ifNull(service_name, '') AS service_name,
+ mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), resource_attributes)) AS resource_attributes,
+ mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), attributes)) AS attributes,
+ timestamp AS last_seen
+FROM posthog.kafka_metrics_avro
+WHERE kafka_metrics_avro.series_fingerprint IS NOT NULL
+SETTINGS
+ min_insert_block_size_rows = 0,
+ min_insert_block_size_bytes = 0;
CREATE MATERIALIZED VIEW posthog.kafka_trace_spans_avro_mv TO posthog.trace_spans (uuid String, trace_id String, span_id String, parent_span_id String, trace_state String, name String, kind Int8, flags UInt32, timestamp DateTime64(6), end_time DateTime64(6), observed_timestamp DateTime64(6), service_name String, resource_attributes Map(LowCardinality(String), String), instrumentation_scope String, attributes_map_str Map(LowCardinality(String), String), dropped_attributes_count UInt32, events Array(String), dropped_events_count UInt32, links Array(String), dropped_links_count UInt32, status_code Int16, team_id Int32, original_expiry_timestamp DateTime64(6)) AS SELECT
* EXCEPT(attributes, resource_attributes, kind, flags, dropped_attributes_count, dropped_events_count, dropped_links_count, status_code),
toInt8(kind) AS kind,
@@ -804,6 +975,57 @@ FROM
)
GROUP BY
team_id, time_bucket, service_name, namespace, environment, severity_text;
+CREATE MATERIALIZED VIEW posthog.metrics1_to_metric_attributes TO posthog.metric_attributes (team_id Int32, time_bucket DateTime64(0), service_name LowCardinality(String), resource_fingerprint UInt64, attribute_key LowCardinality(String), attribute_value String, attribute_type LowCardinality(String), attribute_count SimpleAggregateFunction(sum, UInt64)) AS SELECT
+ team_id,
+ time_bucket,
+ service_name,
+ resource_fingerprint,
+ attribute_key,
+ attribute_value,
+ attribute_type,
+ attribute_count
+FROM
+ (
+ SELECT
+ team_id AS team_id,
+ toStartOfInterval(timestamp, toIntervalMinute(10)) AS time_bucket,
+ service_name AS service_name,
+ resource_fingerprint,
+ mapFilter((k, v) -> ((length(k) < 256) AND (length(v) < 256)), attributes) AS attributes,
+ arrayJoin(attributes) AS attribute,
+ 'metric' AS attribute_type,
+ attribute.1 AS attribute_key,
+ attribute.2 AS attribute_value,
+ sumSimpleState(1) AS attribute_count
+ FROM posthog.metrics1
+ GROUP BY
+ team_id, time_bucket, service_name, resource_fingerprint, attributes
+ );
+CREATE MATERIALIZED VIEW posthog.metrics1_to_resource_attributes TO posthog.metric_attributes (team_id Int32, time_bucket DateTime64(0), service_name LowCardinality(String), resource_fingerprint UInt64, attribute_key LowCardinality(String), attribute_value String, attribute_type LowCardinality(String), attribute_count SimpleAggregateFunction(sum, UInt64)) AS SELECT
+ team_id,
+ time_bucket,
+ service_name,
+ resource_fingerprint,
+ attribute_key,
+ attribute_value,
+ attribute_type,
+ attribute_count
+FROM
+ (
+ SELECT
+ team_id AS team_id,
+ toStartOfInterval(timestamp, toIntervalMinute(10)) AS time_bucket,
+ service_name AS service_name,
+ resource_fingerprint,
+ arrayJoin(resource_attributes) AS attribute,
+ 'resource' AS attribute_type,
+ attribute.1 AS attribute_key,
+ attribute.2 AS attribute_value,
+ sumSimpleState(1) AS attribute_count
+ FROM posthog.metrics1
+ GROUP BY
+ team_id, time_bucket, service_name, resource_fingerprint, resource_attributes
+ );
CREATE MATERIALIZED VIEW posthog.trace_span_to_attributes TO posthog.trace_attributes (team_id Int32, original_expiry_time_bucket DateTime64(0), time_bucket DateTime64(0), service_name LowCardinality(String), resource_fingerprint UInt64, attribute_key LowCardinality(String), attribute_value String, attribute_type LowCardinality(String), attribute_count SimpleAggregateFunction(sum, UInt64)) AS SELECT
team_id,
original_expiry_time_bucket,
@@ -1028,3 +1250,31 @@ CREATE TABLE posthog.metric_series (
attributes Map(LowCardinality(String), String),
last_seen DateTime64(6) CODEC(DoubleDelta)
) ENGINE = Distributed('posthog_single_shard', 'posthog', 'metric_series1');
+CREATE TABLE posthog.metrics (
+ time_bucket DateTime MATERIALIZED toStartOfDay(timestamp),
+ uuid String,
+ team_id Int32,
+ trace_id String,
+ span_id String,
+ trace_flags Int32,
+ timestamp DateTime64(6),
+ observed_timestamp DateTime64(6),
+ created_at DateTime64(6) MATERIALIZED now(),
+ service_name LowCardinality(String),
+ metric_name LowCardinality(String),
+ metric_type LowCardinality(String),
+ value Float64 CODEC(Gorilla(8)),
+ count UInt64 DEFAULT 1 CODEC(T64),
+ histogram_bounds Array(Float64),
+ histogram_counts Array(UInt64),
+ unit LowCardinality(String),
+ aggregation_temporality LowCardinality(String),
+ is_monotonic Bool DEFAULT false,
+ resource_attributes Map(LowCardinality(String), String),
+ resource_fingerprint UInt64 MATERIALIZED cityHash64(resource_attributes),
+ instrumentation_scope String,
+ attributes_map_str Map(LowCardinality(String), String),
+ attributes_map_float Map(LowCardinality(String), Float64),
+ time_minute DateTime ALIAS toStartOfMinute(timestamp),
+ attributes Map(String, String) ALIAS mapApply((k, v) -> (left(k, -5), v), attributes_map_str)
+) ENGINE = Distributed('posthog_single_shard', 'posthog', 'metrics1');
diff --git a/posthog/clickhouse/hcl/sql/local-single/all.sql b/posthog/clickhouse/hcl/sql/local-single/all.sql
index d006b09a1d0f..feecb809a1ff 100644
--- a/posthog/clickhouse/hcl/sql/local-single/all.sql
+++ b/posthog/clickhouse/hcl/sql/local-single/all.sql
@@ -566,6 +566,28 @@ CREATE TABLE posthog.kafka_message_assets (
is_deleted UInt8,
html String
) ENGINE = Kafka() SETTINGS kafka_broker_list = 'warpstream_cyclotron', kafka_format = 'kafka_format = \'JSONEachRow\'', kafka_group_name = 'kafka_group_name = \'clickhouse_message_assets\'', kafka_skip_broken_messages = 100, kafka_topic_list = 'kafka_topic_list = \'clickhouse_message_assets\'';
+CREATE TABLE posthog.kafka_metrics_avro (
+ uuid String,
+ trace_id String,
+ span_id String,
+ trace_flags Nullable(Int32),
+ timestamp DateTime64(6),
+ observed_timestamp DateTime64(6),
+ service_name Nullable(String),
+ metric_name Nullable(String),
+ metric_type Nullable(String),
+ value Nullable(Float64),
+ count Nullable(Int64),
+ histogram_bounds Array(Float64),
+ histogram_counts Array(Int64),
+ unit Nullable(String),
+ aggregation_temporality Nullable(String),
+ is_monotonic Nullable(UInt8),
+ resource_attributes Map(String, String),
+ instrumentation_scope Nullable(String),
+ attributes Map(String, String),
+ series_fingerprint Nullable(Int64)
+) ENGINE = Kafka() SETTINGS kafka_broker_list = 'warpstream_metrics', kafka_format = 'kafka_format = \'Avro\'', kafka_group_name = 'kafka_group_name = \'clickhouse-metrics-avro-new\'', kafka_num_consumers = 8, kafka_poll_max_batch_size = 1000, kafka_poll_timeout_ms = 3000, kafka_skip_broken_messages = 100, kafka_thread_per_consumer = 1, kafka_topic_list = 'kafka_topic_list = \'clickhouse_metrics\'';
CREATE TABLE posthog.kafka_performance_events (
uuid UUID,
session_id String,
@@ -1143,6 +1165,20 @@ CREATE TABLE posthog.message_assets_data (
INDEX person_id_idx person_id TYPE bloom_filter(0.01) GRANULARITY 1,
INDEX recipient_idx recipient TYPE bloom_filter(0.01) GRANULARITY 1
) ENGINE = ReplicatedReplacingMergeTree('/clickhouse/tables/noshard/posthog.message_assets_data', '{replica}-{shard}', version) ORDER BY (team_id, function_kind, function_id, invocation_id, action_id) PARTITION BY toYYYYMMDD(sent_at) TTL toDate(sent_at) + toIntervalDay(30) SETTINGS index_granularity = 1024, ttl_only_drop_parts = 1;
+CREATE TABLE posthog.metric_attributes (
+ team_id Int32,
+ time_bucket DateTime64(0),
+ service_name LowCardinality(String),
+ resource_fingerprint UInt64 DEFAULT 0,
+ attribute_key LowCardinality(String),
+ attribute_value String,
+ attribute_count SimpleAggregateFunction(sum, UInt64),
+ attribute_type LowCardinality(String),
+ INDEX idx_attribute_key attribute_key TYPE bloom_filter(0.01) GRANULARITY 1,
+ INDEX idx_attribute_value attribute_value TYPE bloom_filter(0.01) GRANULARITY 1,
+ INDEX idx_attribute_key_n3 attribute_key TYPE ngrambf_v1(3, 32768, 3, 0) GRANULARITY 1,
+ INDEX idx_attribute_value_n3 attribute_value TYPE ngrambf_v1(3, 32768, 3, 0) GRANULARITY 1
+) ENGINE = ReplicatedAggregatingMergeTree('/clickhouse/tables/noshard/posthog.metric_attributes', '{replica}') ORDER BY (team_id, attribute_type, time_bucket, resource_fingerprint, attribute_key, attribute_value) PARTITION BY toDate(time_bucket) SETTINGS deduplicate_merge_projection_mode = 'drop', index_granularity = 8192;
CREATE TABLE posthog.metric_samples1 (
team_id Int32,
metric_name LowCardinality(String),
@@ -1173,6 +1209,62 @@ CREATE TABLE posthog.metric_series1 (
INDEX idx_attr_keys mapKeys(attributes) TYPE bloom_filter(0.01) GRANULARITY 1,
INDEX idx_attr_values mapValues(attributes) TYPE bloom_filter(0.01) GRANULARITY 1
) ENGINE = ReplicatedReplacingMergeTree('/clickhouse/tables/noshard/posthog.metric_series1', '{replica}-{shard}', last_seen) ORDER BY (team_id, metric_name, series_fingerprint) TTL toDateTime(last_seen) + toIntervalDay(90) SETTINGS index_granularity = 8192;
+CREATE TABLE posthog.metrics1 (
+ time_bucket DateTime MATERIALIZED toStartOfDay(timestamp),
+ uuid String,
+ team_id Int32,
+ trace_id String,
+ span_id String,
+ trace_flags Int32,
+ timestamp DateTime64(6),
+ observed_timestamp DateTime64(6),
+ created_at DateTime64(6) MATERIALIZED now(),
+ service_name LowCardinality(String),
+ metric_name LowCardinality(String),
+ metric_type LowCardinality(String),
+ value Float64 CODEC(Gorilla(8)),
+ count UInt64 DEFAULT 1 CODEC(T64),
+ histogram_bounds Array(Float64),
+ histogram_counts Array(UInt64),
+ unit LowCardinality(String),
+ aggregation_temporality LowCardinality(String),
+ is_monotonic Bool DEFAULT false,
+ resource_attributes Map(LowCardinality(String), String),
+ resource_fingerprint UInt64 MATERIALIZED cityHash64(resource_attributes),
+ instrumentation_scope String,
+ attributes_map_str Map(LowCardinality(String), String),
+ attributes_map_float Map(LowCardinality(String), Float64),
+ time_minute DateTime ALIAS toStartOfMinute(timestamp),
+ attributes Map(String, String) ALIAS mapApply((k, v) -> (left(k, -5), v), attributes_map_str),
+ INDEX idx_metric_name_set metric_name TYPE set(100) GRANULARITY 1,
+ INDEX idx_metric_type_set metric_type TYPE set(10) GRANULARITY 1,
+ INDEX idx_attributes_str_keys mapKeys(attributes_map_str) TYPE bloom_filter(0.01) GRANULARITY 1,
+ INDEX idx_attributes_str_values mapValues(attributes_map_str) TYPE bloom_filter(0.001) GRANULARITY 1,
+ INDEX idx_observed_minmax observed_timestamp TYPE minmax GRANULARITY 1,
+ PROJECTION projection_aggregate_counts (SELECT
+ team_id,
+ time_bucket,
+ toStartOfMinute(timestamp),
+ service_name,
+ metric_name,
+ metric_type,
+ resource_fingerprint,
+ count() AS event_count,
+ sum(value) AS total_value,
+ min(value) AS min_value,
+ max(value) AS max_value
+GROUP BY
+ team_id, time_bucket, toStartOfMinute(timestamp), service_name, metric_name, metric_type, resource_fingerprint)
+) ENGINE = ReplicatedMergeTree('/clickhouse/tables/noshard/posthog.metrics1', '{replica}') ORDER BY (team_id, time_bucket, service_name, metric_name, resource_fingerprint, timestamp) PARTITION BY toDate(timestamp) SETTINGS index_granularity = 8192, index_granularity_bytes = 104857600, ttl_only_drop_parts = 1;
+CREATE TABLE posthog.metrics_kafka_metrics (
+ _partition UInt32,
+ _topic String,
+ max_offset SimpleAggregateFunction(max, UInt64),
+ max_observed_timestamp SimpleAggregateFunction(max, DateTime64(9)),
+ max_timestamp SimpleAggregateFunction(max, DateTime64(9)),
+ max_created_at SimpleAggregateFunction(max, DateTime64(9)),
+ max_lag SimpleAggregateFunction(max, UInt64)
+) ENGINE = ReplicatedAggregatingMergeTree('/clickhouse/tables/noshard/posthog.metrics_kafka_metrics', '{replica}') ORDER BY (_topic, _partition) SETTINGS index_granularity = 8192;
CREATE TABLE posthog.partitioned_sharded_posthog_document_embeddings (
team_id Int64,
product LowCardinality(String),
@@ -4388,6 +4480,85 @@ CREATE MATERIALIZED VIEW posthog.kafka_logs_avro_kafka_metrics_mv TO posthog.log
FROM posthog.logs34
GROUP BY
_partition, _topic;
+CREATE MATERIALIZED VIEW posthog.kafka_metrics_avro_kafka_metrics_mv TO posthog.metrics_kafka_metrics (_partition UInt64, _topic LowCardinality(String), max_offset SimpleAggregateFunction(max, UInt64), max_observed_timestamp SimpleAggregateFunction(max, DateTime64(6)), max_timestamp SimpleAggregateFunction(max, DateTime64(6)), max_created_at SimpleAggregateFunction(max, DateTime), max_lag SimpleAggregateFunction(max, Decimal(18, 6))) AS SELECT
+ _partition,
+ _topic,
+ maxSimpleState(_offset) AS max_offset,
+ maxSimpleState(observed_timestamp) AS max_observed_timestamp,
+ maxSimpleState(timestamp) AS max_timestamp,
+ maxSimpleState(now()) AS max_created_at,
+ maxSimpleState(now() - observed_timestamp) AS max_lag
+FROM posthog.kafka_metrics_avro
+GROUP BY
+ _partition, _topic;
+CREATE MATERIALIZED VIEW posthog.kafka_metrics_avro_mv TO posthog.metrics1 (uuid String, trace_id String, span_id String, trace_flags Int32, timestamp DateTime64(6), observed_timestamp DateTime64(6), service_name String, metric_name String, metric_type String, value Float64, count UInt64, histogram_bounds Array(Float64), histogram_counts Array(UInt64), unit String, aggregation_temporality String, is_monotonic UInt8, resource_attributes Map(String, String), instrumentation_scope String, attributes_map_str Map(String, String), attributes_map_float Map(String, Nullable(Float64)), team_id Int32) AS SELECT
+ uuid,
+ trace_id,
+ span_id,
+ ifNull(trace_flags, 0) AS trace_flags,
+ timestamp,
+ observed_timestamp,
+ ifNull(service_name, '') AS service_name,
+ ifNull(metric_name, '') AS metric_name,
+ ifNull(metric_type, '') AS metric_type,
+ ifNull(value, 0) AS value,
+ toUInt64(ifNull(count, 1)) AS count,
+ histogram_bounds,
+ arrayMap(x -> toUInt64(x), histogram_counts) AS histogram_counts,
+ ifNull(unit, '') AS unit,
+ ifNull(aggregation_temporality, '') AS aggregation_temporality,
+ ifNull(is_monotonic, 0) AS is_monotonic,
+ mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), resource_attributes)) AS resource_attributes,
+ ifNull(instrumentation_scope, '') AS instrumentation_scope,
+ mapSort(mapApply((k, v) -> (concat(k, '__str'), JSONExtractString(v)), attributes)) AS attributes_map_str,
+ mapSort(
+ mapFilter(
+ (k, v) -> isNotNull(v),
+ mapApply(
+ (k, v) -> (concat(k, '__float'), toFloat64OrNull(JSONExtract(v, 'String'))),
+ attributes
+ )
+ )
+ ) AS attributes_map_float,
+ toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) AS team_id
+FROM posthog.kafka_metrics_avro
+SETTINGS
+ min_insert_block_size_rows = 0,
+ min_insert_block_size_bytes = 0;
+CREATE MATERIALIZED VIEW posthog.kafka_metrics_avro_to_metric_samples TO posthog.metric_samples1 (team_id Int32, metric_name String, series_fingerprint UInt64, timestamp DateTime64(6), value Float64, count UInt64, histogram_bounds Array(Float64), histogram_counts Array(UInt64), trace_id String, span_id String, trace_flags Int32) AS SELECT
+ toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) AS team_id,
+ ifNull(metric_name, '') AS metric_name,
+ reinterpretAsUInt64(assumeNotNull(series_fingerprint)) AS series_fingerprint,
+ timestamp,
+ ifNull(value, 0) AS value,
+ toUInt64(ifNull(count, 1)) AS count,
+ histogram_bounds,
+ arrayMap(x -> toUInt64(x), histogram_counts) AS histogram_counts,
+ trace_id,
+ span_id,
+ ifNull(trace_flags, 0) AS trace_flags
+FROM posthog.kafka_metrics_avro
+WHERE kafka_metrics_avro.series_fingerprint IS NOT NULL
+SETTINGS
+ min_insert_block_size_rows = 0,
+ min_insert_block_size_bytes = 0;
+CREATE MATERIALIZED VIEW posthog.kafka_metrics_avro_to_metric_series TO posthog.metric_series1 (team_id Int32, metric_name String, series_fingerprint UInt64, metric_type String, unit String, aggregation_temporality String, is_monotonic UInt8, service_name String, resource_attributes Map(String, String), attributes Map(String, String), last_seen DateTime64(6)) AS SELECT
+ toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) AS team_id,
+ ifNull(metric_name, '') AS metric_name,
+ reinterpretAsUInt64(assumeNotNull(series_fingerprint)) AS series_fingerprint,
+ ifNull(metric_type, '') AS metric_type,
+ ifNull(unit, '') AS unit,
+ ifNull(aggregation_temporality, '') AS aggregation_temporality,
+ ifNull(is_monotonic, 0) AS is_monotonic,
+ ifNull(service_name, '') AS service_name,
+ mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), resource_attributes)) AS resource_attributes,
+ mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), attributes)) AS attributes,
+ timestamp AS last_seen
+FROM posthog.kafka_metrics_avro
+WHERE kafka_metrics_avro.series_fingerprint IS NOT NULL
+SETTINGS
+ min_insert_block_size_rows = 0,
+ min_insert_block_size_bytes = 0;
CREATE MATERIALIZED VIEW posthog.kafka_trace_spans_avro_mv TO posthog.trace_spans (uuid String, trace_id String, span_id String, parent_span_id String, trace_state String, name String, kind Int8, flags UInt32, timestamp DateTime64(6), end_time DateTime64(6), observed_timestamp DateTime64(6), service_name String, resource_attributes Map(LowCardinality(String), String), instrumentation_scope String, attributes_map_str Map(LowCardinality(String), String), dropped_attributes_count UInt32, events Array(String), dropped_events_count UInt32, links Array(String), dropped_links_count UInt32, status_code Int16, team_id Int32, original_expiry_timestamp DateTime64(6)) AS SELECT
* EXCEPT(attributes, resource_attributes, kind, flags, dropped_attributes_count, dropped_events_count, dropped_links_count, status_code),
toInt8(kind) AS kind,
@@ -4657,6 +4828,57 @@ CREATE MATERIALIZED VIEW posthog.message_assets_mv TO posthog.message_assets_dat
_offset,
_partition
FROM posthog.kafka_message_assets;
+CREATE MATERIALIZED VIEW posthog.metrics1_to_metric_attributes TO posthog.metric_attributes (team_id Int32, time_bucket DateTime64(0), service_name LowCardinality(String), resource_fingerprint UInt64, attribute_key LowCardinality(String), attribute_value String, attribute_type LowCardinality(String), attribute_count SimpleAggregateFunction(sum, UInt64)) AS SELECT
+ team_id,
+ time_bucket,
+ service_name,
+ resource_fingerprint,
+ attribute_key,
+ attribute_value,
+ attribute_type,
+ attribute_count
+FROM
+ (
+ SELECT
+ team_id AS team_id,
+ toStartOfInterval(timestamp, toIntervalMinute(10)) AS time_bucket,
+ service_name AS service_name,
+ resource_fingerprint,
+ mapFilter((k, v) -> ((length(k) < 256) AND (length(v) < 256)), attributes) AS attributes,
+ arrayJoin(attributes) AS attribute,
+ 'metric' AS attribute_type,
+ attribute.1 AS attribute_key,
+ attribute.2 AS attribute_value,
+ sumSimpleState(1) AS attribute_count
+ FROM posthog.metrics1
+ GROUP BY
+ team_id, time_bucket, service_name, resource_fingerprint, attributes
+ );
+CREATE MATERIALIZED VIEW posthog.metrics1_to_resource_attributes TO posthog.metric_attributes (team_id Int32, time_bucket DateTime64(0), service_name LowCardinality(String), resource_fingerprint UInt64, attribute_key LowCardinality(String), attribute_value String, attribute_type LowCardinality(String), attribute_count SimpleAggregateFunction(sum, UInt64)) AS SELECT
+ team_id,
+ time_bucket,
+ service_name,
+ resource_fingerprint,
+ attribute_key,
+ attribute_value,
+ attribute_type,
+ attribute_count
+FROM
+ (
+ SELECT
+ team_id AS team_id,
+ toStartOfInterval(timestamp, toIntervalMinute(10)) AS time_bucket,
+ service_name AS service_name,
+ resource_fingerprint,
+ arrayJoin(resource_attributes) AS attribute,
+ 'resource' AS attribute_type,
+ attribute.1 AS attribute_key,
+ attribute.2 AS attribute_value,
+ sumSimpleState(1) AS attribute_count
+ FROM posthog.metrics1
+ GROUP BY
+ team_id, time_bucket, service_name, resource_fingerprint, resource_attributes
+ );
CREATE MATERIALIZED VIEW posthog.performance_events_mv TO posthog.writeable_performance_events (uuid UUID, session_id String, window_id String, pageview_id String, distinct_id String, timestamp DateTime64(3), time_origin DateTime64(3, 'UTC'), entry_type LowCardinality(String), name String, team_id Int64, current_url String, start_time Float64, duration Float64, redirect_start Float64, redirect_end Float64, worker_start Float64, fetch_start Float64, domain_lookup_start Float64, domain_lookup_end Float64, connect_start Float64, secure_connection_start Float64, connect_end Float64, request_start Float64, response_start Float64, response_end Float64, decoded_body_size Int64, encoded_body_size Int64, initiator_type LowCardinality(String), next_hop_protocol LowCardinality(String), render_blocking_status LowCardinality(String), response_status Int64, transfer_size Int64, largest_contentful_paint_element String, largest_contentful_paint_render_time Float64, largest_contentful_paint_load_time Float64, largest_contentful_paint_size Float64, largest_contentful_paint_id String, largest_contentful_paint_url String, dom_complete Float64, dom_content_loaded_event Float64, dom_interactive Float64, load_event_end Float64, load_event_start Float64, redirect_count Int64, navigation_type LowCardinality(String), unload_event_end Float64, unload_event_start Float64, _timestamp Nullable(DateTime), _offset UInt64, _partition UInt64) AS SELECT
uuid,
session_id,
@@ -6048,6 +6270,34 @@ CREATE TABLE posthog.metric_series (
attributes Map(LowCardinality(String), String),
last_seen DateTime64(6) CODEC(DoubleDelta)
) ENGINE = Distributed('posthog_single_shard', 'posthog', 'metric_series1');
+CREATE TABLE posthog.metrics (
+ time_bucket DateTime MATERIALIZED toStartOfDay(timestamp),
+ uuid String,
+ team_id Int32,
+ trace_id String,
+ span_id String,
+ trace_flags Int32,
+ timestamp DateTime64(6),
+ observed_timestamp DateTime64(6),
+ created_at DateTime64(6) MATERIALIZED now(),
+ service_name LowCardinality(String),
+ metric_name LowCardinality(String),
+ metric_type LowCardinality(String),
+ value Float64 CODEC(Gorilla(8)),
+ count UInt64 DEFAULT 1 CODEC(T64),
+ histogram_bounds Array(Float64),
+ histogram_counts Array(UInt64),
+ unit LowCardinality(String),
+ aggregation_temporality LowCardinality(String),
+ is_monotonic Bool DEFAULT false,
+ resource_attributes Map(LowCardinality(String), String),
+ resource_fingerprint UInt64 MATERIALIZED cityHash64(resource_attributes),
+ instrumentation_scope String,
+ attributes_map_str Map(LowCardinality(String), String),
+ attributes_map_float Map(LowCardinality(String), Float64),
+ time_minute DateTime ALIAS toStartOfMinute(timestamp),
+ attributes Map(String, String) ALIAS mapApply((k, v) -> (left(k, -5), v), attributes_map_str)
+) ENGINE = Distributed('posthog_single_shard', 'posthog', 'metrics1');
CREATE TABLE posthog.performance_events (
uuid UUID,
session_id String,
diff --git a/posthog/clickhouse/metrics/__init__.py b/posthog/clickhouse/metrics/__init__.py
index feede7a604cf..28a9478b1a73 100644
--- a/posthog/clickhouse/metrics/__init__.py
+++ b/posthog/clickhouse/metrics/__init__.py
@@ -1,3 +1,11 @@
+from .kafka_metrics import (
+ KAFKA_METRICS_AVRO_MV,
+ KAFKA_METRICS_AVRO_TABLE_SQL,
+ KAFKA_METRICS_AVRO_TO_KAFKA_METRICS_MV,
+ KAFKA_METRICS_AVRO_TO_METRIC_SAMPLES_MV,
+ KAFKA_METRICS_AVRO_TO_METRIC_SERIES_MV,
+ METRICS_KAFKA_METRICS_TABLE_SQL,
+)
from .metric_attributes import METRIC_ATTRIBUTES_TABLE_SQL
from .metric_events import (
METRIC_SAMPLES_DISTRIBUTED_TABLE_SQL,
@@ -22,4 +30,10 @@
"METRIC_SERIES_DISTRIBUTED_TABLE_SQL",
"METRIC_SAMPLES_TABLE_SQL",
"METRIC_SAMPLES_DISTRIBUTED_TABLE_SQL",
+ "KAFKA_METRICS_AVRO_TABLE_SQL",
+ "KAFKA_METRICS_AVRO_MV",
+ "KAFKA_METRICS_AVRO_TO_METRIC_SAMPLES_MV",
+ "KAFKA_METRICS_AVRO_TO_METRIC_SERIES_MV",
+ "KAFKA_METRICS_AVRO_TO_KAFKA_METRICS_MV",
+ "METRICS_KAFKA_METRICS_TABLE_SQL",
]
diff --git a/posthog/clickhouse/metrics/kafka_metrics.py b/posthog/clickhouse/metrics/kafka_metrics.py
new file mode 100644
index 000000000000..c2f72682396a
--- /dev/null
+++ b/posthog/clickhouse/metrics/kafka_metrics.py
@@ -0,0 +1,176 @@
+"""Kafka ingest for the metrics product: `clickhouse_metrics` topic -> `kafka_metrics_avro`
+-> materialized views into `metrics1` (raw), `metric_samples1` + `metric_series1` (the split
+the samples/exemplar queries read), and `metrics_kafka_metrics` (consumer-lag bookkeeping).
+
+These objects are modeled in the all-env HCL layer
+(`posthog/clickhouse/hcl/roles/logs/metrics`); this module is the migration-applied
+equivalent so local and multinode dev get a working ingest chain too, the same way
+`posthog/clickhouse/traces/spans.py` does for trace spans. Keep the SQL in sync with the
+HCL definitions — the multinode convergence gate diffs the live schema against the goldens.
+"""
+
+from django.conf import settings
+
+from posthog.clickhouse.kafka_engine import kafka_engine
+from posthog.clickhouse.table_engines import AggregatingMergeTree, ReplicationScheme
+
+from .metric_events import SAMPLES_TABLE_NAME, SERIES_TABLE_NAME
+from .metrics1 import TABLE_NAME as METRICS1_TABLE_NAME
+
+KAFKA_TABLE_NAME = "kafka_metrics_avro"
+KAFKA_METRICS_TABLE_NAME = "metrics_kafka_metrics"
+KAFKA_NAMED_COLLECTION = "warpstream_metrics"
+KAFKA_TOPIC = "clickhouse_metrics"
+KAFKA_GROUP = "clickhouse-metrics-avro-new"
+
+
+def KAFKA_METRICS_AVRO_TABLE_SQL():
+ return f"""
+CREATE TABLE IF NOT EXISTS {settings.CLICKHOUSE_LOGS_CLUSTER_DATABASE}.{KAFKA_TABLE_NAME}
+(
+ `uuid` String,
+ `trace_id` String,
+ `span_id` String,
+ `trace_flags` Nullable(Int32),
+ `timestamp` DateTime64(6),
+ `observed_timestamp` DateTime64(6),
+ `service_name` Nullable(String),
+ `metric_name` Nullable(String),
+ `metric_type` Nullable(String),
+ `value` Nullable(Float64),
+ `count` Nullable(Int64),
+ `histogram_bounds` Array(Float64),
+ `histogram_counts` Array(Int64),
+ `unit` Nullable(String),
+ `aggregation_temporality` Nullable(String),
+ `is_monotonic` Nullable(UInt8),
+ `resource_attributes` Map(String, String),
+ `instrumentation_scope` Nullable(String),
+ `attributes` Map(String, String),
+ `series_fingerprint` Nullable(Int64)
+)
+ENGINE = {kafka_engine(topic=KAFKA_TOPIC, group=KAFKA_GROUP, serialization="Avro", named_collection=KAFKA_NAMED_COLLECTION)}
+SETTINGS
+ kafka_skip_broken_messages = 100,
+ kafka_thread_per_consumer = 1,
+ kafka_num_consumers = 8,
+ kafka_poll_timeout_ms = 3000,
+ kafka_poll_max_batch_size = 1000
+"""
+
+
+def KAFKA_METRICS_AVRO_MV():
+ db = settings.CLICKHOUSE_LOGS_CLUSTER_DATABASE
+ return f"""
+CREATE MATERIALIZED VIEW IF NOT EXISTS {db}.{KAFKA_TABLE_NAME}_mv TO {db}.{METRICS1_TABLE_NAME}
+AS SELECT
+ uuid,
+ trace_id,
+ span_id,
+ ifNull(trace_flags, 0) AS trace_flags,
+ timestamp,
+ observed_timestamp,
+ ifNull(service_name, '') AS service_name,
+ ifNull(metric_name, '') AS metric_name,
+ ifNull(metric_type, '') AS metric_type,
+ ifNull(value, 0) AS value,
+ toUInt64(ifNull(count, 1)) AS count,
+ histogram_bounds,
+ arrayMap(x -> toUInt64(x), histogram_counts) AS histogram_counts,
+ ifNull(unit, '') AS unit,
+ ifNull(aggregation_temporality, '') AS aggregation_temporality,
+ ifNull(is_monotonic, 0) AS is_monotonic,
+ mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), resource_attributes)) AS resource_attributes,
+ ifNull(instrumentation_scope, '') AS instrumentation_scope,
+ mapSort(mapApply((k, v) -> (concat(k, '__str'), JSONExtractString(v)), attributes)) AS attributes_map_str,
+ mapSort(mapFilter((k, v) -> isNotNull(v), mapApply((k, v) -> (concat(k, '__float'), toFloat64OrNull(JSONExtract(v, 'String'))), attributes))) AS attributes_map_float,
+ toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) AS team_id
+FROM {db}.{KAFKA_TABLE_NAME}
+SETTINGS
+ min_insert_block_size_rows = 0,
+ min_insert_block_size_bytes = 0
+"""
+
+
+def KAFKA_METRICS_AVRO_TO_METRIC_SAMPLES_MV():
+ db = settings.CLICKHOUSE_LOGS_CLUSTER_DATABASE
+ return f"""
+CREATE MATERIALIZED VIEW IF NOT EXISTS {db}.{KAFKA_TABLE_NAME}_to_metric_samples TO {db}.{SAMPLES_TABLE_NAME}
+AS SELECT
+ toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) AS team_id,
+ ifNull(metric_name, '') AS metric_name,
+ reinterpretAsUInt64(assumeNotNull(series_fingerprint)) AS series_fingerprint,
+ timestamp,
+ ifNull(value, 0) AS value,
+ toUInt64(ifNull(count, 1)) AS count,
+ histogram_bounds,
+ arrayMap(x -> toUInt64(x), histogram_counts) AS histogram_counts,
+ trace_id,
+ span_id,
+ ifNull(trace_flags, 0) AS trace_flags
+FROM {db}.{KAFKA_TABLE_NAME}
+WHERE {KAFKA_TABLE_NAME}.series_fingerprint IS NOT NULL
+SETTINGS
+ min_insert_block_size_rows = 0,
+ min_insert_block_size_bytes = 0
+"""
+
+
+def KAFKA_METRICS_AVRO_TO_METRIC_SERIES_MV():
+ db = settings.CLICKHOUSE_LOGS_CLUSTER_DATABASE
+ return f"""
+CREATE MATERIALIZED VIEW IF NOT EXISTS {db}.{KAFKA_TABLE_NAME}_to_metric_series TO {db}.{SERIES_TABLE_NAME}
+AS SELECT
+ toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) AS team_id,
+ ifNull(metric_name, '') AS metric_name,
+ reinterpretAsUInt64(assumeNotNull(series_fingerprint)) AS series_fingerprint,
+ ifNull(metric_type, '') AS metric_type,
+ ifNull(unit, '') AS unit,
+ ifNull(aggregation_temporality, '') AS aggregation_temporality,
+ ifNull(is_monotonic, 0) AS is_monotonic,
+ ifNull(service_name, '') AS service_name,
+ mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), resource_attributes)) AS resource_attributes,
+ mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), attributes)) AS attributes,
+ timestamp AS last_seen
+FROM {db}.{KAFKA_TABLE_NAME}
+WHERE {KAFKA_TABLE_NAME}.series_fingerprint IS NOT NULL
+SETTINGS
+ min_insert_block_size_rows = 0,
+ min_insert_block_size_bytes = 0
+"""
+
+
+def METRICS_KAFKA_METRICS_TABLE_SQL():
+ return f"""
+CREATE TABLE IF NOT EXISTS {settings.CLICKHOUSE_LOGS_CLUSTER_DATABASE}.{KAFKA_METRICS_TABLE_NAME}
+(
+ `_partition` UInt32,
+ `_topic` String,
+ `max_offset` SimpleAggregateFunction(max, UInt64),
+ `max_observed_timestamp` SimpleAggregateFunction(max, DateTime64(9)),
+ `max_timestamp` SimpleAggregateFunction(max, DateTime64(9)),
+ `max_created_at` SimpleAggregateFunction(max, DateTime64(9)),
+ `max_lag` SimpleAggregateFunction(max, UInt64)
+)
+ENGINE = {AggregatingMergeTree(KAFKA_METRICS_TABLE_NAME, replication_scheme=ReplicationScheme.REPLICATED)}
+ORDER BY (_topic, _partition)
+SETTINGS
+ index_granularity = 8192
+"""
+
+
+def KAFKA_METRICS_AVRO_TO_KAFKA_METRICS_MV():
+ db = settings.CLICKHOUSE_LOGS_CLUSTER_DATABASE
+ return f"""
+CREATE MATERIALIZED VIEW IF NOT EXISTS {db}.{KAFKA_TABLE_NAME}_kafka_metrics_mv TO {db}.{KAFKA_METRICS_TABLE_NAME}
+AS SELECT
+ _partition,
+ _topic,
+ maxSimpleState(_offset) AS max_offset,
+ maxSimpleState(observed_timestamp) AS max_observed_timestamp,
+ maxSimpleState(timestamp) AS max_timestamp,
+ maxSimpleState(now()) AS max_created_at,
+ maxSimpleState(now() - observed_timestamp) AS max_lag
+FROM {db}.{KAFKA_TABLE_NAME}
+GROUP BY _partition, _topic
+"""
diff --git a/posthog/clickhouse/metrics/metric_attributes.py b/posthog/clickhouse/metrics/metric_attributes.py
index 6efbed5745e8..67e18357aab6 100644
--- a/posthog/clickhouse/metrics/metric_attributes.py
+++ b/posthog/clickhouse/metrics/metric_attributes.py
@@ -9,12 +9,12 @@ def METRIC_ATTRIBUTES_TABLE_SQL():
return f"""
CREATE TABLE IF NOT EXISTS {settings.CLICKHOUSE_LOGS_CLUSTER_DATABASE}.{TABLE_NAME}
(
- `team_id` Int32 CODEC(DoubleDelta, ZSTD(1)),
- `time_bucket` DateTime64(0) CODEC(DoubleDelta, ZSTD(1)),
- `service_name` LowCardinality(String) CODEC(ZSTD(1)),
- `resource_fingerprint` UInt64 DEFAULT 0 CODEC(DoubleDelta, ZSTD(1)),
- `attribute_key` LowCardinality(String) CODEC(ZSTD(1)),
- `attribute_value` String CODEC(ZSTD(1)),
+ `team_id` Int32,
+ `time_bucket` DateTime64(0),
+ `service_name` LowCardinality(String),
+ `resource_fingerprint` UInt64 DEFAULT 0,
+ `attribute_key` LowCardinality(String),
+ `attribute_value` String,
`attribute_count` SimpleAggregateFunction(sum, UInt64),
`attribute_type` LowCardinality(String),
INDEX idx_attribute_key attribute_key TYPE bloom_filter(0.01) GRANULARITY 1,
diff --git a/posthog/clickhouse/metrics/metrics1.py b/posthog/clickhouse/metrics/metrics1.py
index 0cca91b7f210..70449766bc4a 100644
--- a/posthog/clickhouse/metrics/metrics1.py
+++ b/posthog/clickhouse/metrics/metrics1.py
@@ -11,30 +11,30 @@ def METRICS1_TABLE_SQL():
return f"""
CREATE TABLE IF NOT EXISTS {settings.CLICKHOUSE_LOGS_CLUSTER_DATABASE}.{TABLE_NAME}
(
- `time_bucket` DateTime MATERIALIZED toStartOfDay(timestamp) CODEC(DoubleDelta, ZSTD(1)),
- `uuid` String CODEC(ZSTD(1)),
- `team_id` Int32 CODEC(ZSTD(1)),
- `trace_id` String CODEC(ZSTD(1)),
- `span_id` String CODEC(ZSTD(1)),
- `trace_flags` Int32 CODEC(ZSTD(1)),
- `timestamp` DateTime64(6) CODEC(DoubleDelta, ZSTD(1)),
- `observed_timestamp` DateTime64(6) CODEC(DoubleDelta, ZSTD(1)),
- `created_at` DateTime64(6) MATERIALIZED now() CODEC(DoubleDelta, ZSTD(1)),
- `service_name` LowCardinality(String) CODEC(ZSTD(1)),
- `metric_name` LowCardinality(String) CODEC(ZSTD(1)),
- `metric_type` LowCardinality(String) CODEC(ZSTD(1)),
- `value` Float64 CODEC(Gorilla, ZSTD(1)),
- `count` UInt64 DEFAULT 1 CODEC(T64, ZSTD(1)),
- `histogram_bounds` Array(Float64) CODEC(ZSTD(1)),
- `histogram_counts` Array(UInt64) CODEC(ZSTD(1)),
- `unit` LowCardinality(String) CODEC(ZSTD(1)),
- `aggregation_temporality` LowCardinality(String) CODEC(ZSTD(1)),
- `is_monotonic` Bool DEFAULT false CODEC(ZSTD(1)),
- `resource_attributes` Map(LowCardinality(String), String) CODEC(ZSTD(1)),
- `resource_fingerprint` UInt64 MATERIALIZED cityHash64(resource_attributes) CODEC(DoubleDelta, ZSTD(1)),
- `instrumentation_scope` String CODEC(ZSTD(1)),
- `attributes_map_str` Map(LowCardinality(String), String) CODEC(ZSTD(1)),
- `attributes_map_float` Map(LowCardinality(String), Float64) CODEC(ZSTD(1)),
+ `time_bucket` DateTime MATERIALIZED toStartOfDay(timestamp),
+ `uuid` String,
+ `team_id` Int32,
+ `trace_id` String,
+ `span_id` String,
+ `trace_flags` Int32,
+ `timestamp` DateTime64(6),
+ `observed_timestamp` DateTime64(6),
+ `created_at` DateTime64(6) MATERIALIZED now(),
+ `service_name` LowCardinality(String),
+ `metric_name` LowCardinality(String),
+ `metric_type` LowCardinality(String),
+ `value` Float64 CODEC(Gorilla),
+ `count` UInt64 DEFAULT 1 CODEC(T64),
+ `histogram_bounds` Array(Float64),
+ `histogram_counts` Array(UInt64),
+ `unit` LowCardinality(String),
+ `aggregation_temporality` LowCardinality(String),
+ `is_monotonic` Bool DEFAULT false,
+ `resource_attributes` Map(LowCardinality(String), String),
+ `resource_fingerprint` UInt64 MATERIALIZED cityHash64(resource_attributes),
+ `instrumentation_scope` String,
+ `attributes_map_str` Map(LowCardinality(String), String),
+ `attributes_map_float` Map(LowCardinality(String), Float64),
`time_minute` DateTime ALIAS toStartOfMinute(timestamp),
`attributes` Map(String, String) ALIAS mapApply((k, v) -> (left(k, -5), v), attributes_map_str),
INDEX idx_metric_name_set metric_name TYPE set(100) GRANULARITY 1,
diff --git a/posthog/clickhouse/migrations/0285_metric_series_temporality_samples_histograms.py b/posthog/clickhouse/migrations/0285_metric_series_temporality_samples_histograms.py
index cf3374dc1601..d115e0ad286b 100644
--- a/posthog/clickhouse/migrations/0285_metric_series_temporality_samples_histograms.py
+++ b/posthog/clickhouse/migrations/0285_metric_series_temporality_samples_histograms.py
@@ -11,12 +11,12 @@
# after their last sample (samples themselves expire at 30).
#
# This migration covers the storage tables only. The ingest MVs that populate the
-# new columns (and carry the table-qualified NULL-fingerprint guard) are
-# hand-managed, like all Avro Kafka-engine objects on the logs cluster — the
-# canonical definitions live in bin/clickhouse-metrics.sql and must be dropped and
-# recreated on each region as part of the fingerprint-cutover DDL. Until that
-# manual step runs, rows written by the old MVs simply leave the new columns at
-# their defaults.
+# new columns (and carry the table-qualified NULL-fingerprint guard) were
+# hand-managed at the time, like all Avro Kafka-engine objects on the logs
+# cluster, and had to be dropped and recreated on each region as part of the
+# fingerprint-cutover DDL. Until that manual step ran, rows written by the old
+# MVs simply left the new columns at their defaults. 0305 has since brought the
+# ingest objects under migrations (posthog/clickhouse/metrics/kafka_metrics.py).
_DB = settings.CLICKHOUSE_LOGS_CLUSTER_DATABASE
diff --git a/posthog/clickhouse/migrations/0308_metrics_kafka_ingest.py b/posthog/clickhouse/migrations/0308_metrics_kafka_ingest.py
new file mode 100644
index 000000000000..f1cf60892649
--- /dev/null
+++ b/posthog/clickhouse/migrations/0308_metrics_kafka_ingest.py
@@ -0,0 +1,40 @@
+from posthog.clickhouse.client.connection import NodeRole
+from posthog.clickhouse.client.migration_tools import run_sql_with_exceptions
+from posthog.clickhouse.metrics import (
+ KAFKA_METRICS_AVRO_MV,
+ KAFKA_METRICS_AVRO_TABLE_SQL,
+ KAFKA_METRICS_AVRO_TO_KAFKA_METRICS_MV,
+ KAFKA_METRICS_AVRO_TO_METRIC_SAMPLES_MV,
+ KAFKA_METRICS_AVRO_TO_METRIC_SERIES_MV,
+ METRIC_ATTRIBUTES_MV,
+ METRIC_ATTRIBUTES_TABLE_SQL,
+ METRIC_RESOURCE_ATTRIBUTES_MV,
+ METRICS1_TABLE_SQL,
+ METRICS_DISTRIBUTED_TABLE_SQL,
+ METRICS_KAFKA_METRICS_TABLE_SQL,
+)
+
+# This migration replaces the hand-run bin/clickhouse-metrics.sql init script (deleted in
+# the same change) — metrics was the last product provisioning its ClickHouse objects
+# that way; logs and traces already moved to migrations.
+#
+# None of these objects had a creating migration: on cloud they are provisioned out of
+# band (the HCL logs-cluster layer), and on a dev laptop through the init script or the
+# schema-sync path — none of which runs in a migrations-only environment. 0283 created
+# metric_samples1/metric_series1 but left out metrics1 (the kafka_metrics_avro_mv
+# destination), metric_attributes and its two feeder views, and the whole Kafka ingest
+# chain. Create them here, storage tables before the views that write to them. All
+# statements are IF NOT EXISTS, so this is a no-op where the objects already exist.
+operations = [
+ run_sql_with_exceptions(METRICS1_TABLE_SQL(), node_roles=[NodeRole.LOGS]),
+ run_sql_with_exceptions(METRICS_DISTRIBUTED_TABLE_SQL(), node_roles=[NodeRole.LOGS]),
+ run_sql_with_exceptions(METRIC_ATTRIBUTES_TABLE_SQL(), node_roles=[NodeRole.LOGS]),
+ run_sql_with_exceptions(METRIC_ATTRIBUTES_MV(), node_roles=[NodeRole.LOGS]),
+ run_sql_with_exceptions(METRIC_RESOURCE_ATTRIBUTES_MV(), node_roles=[NodeRole.LOGS]),
+ run_sql_with_exceptions(METRICS_KAFKA_METRICS_TABLE_SQL(), node_roles=[NodeRole.LOGS]),
+ run_sql_with_exceptions(KAFKA_METRICS_AVRO_TABLE_SQL(), node_roles=[NodeRole.LOGS]),
+ run_sql_with_exceptions(KAFKA_METRICS_AVRO_MV(), node_roles=[NodeRole.LOGS]),
+ run_sql_with_exceptions(KAFKA_METRICS_AVRO_TO_METRIC_SAMPLES_MV(), node_roles=[NodeRole.LOGS]),
+ run_sql_with_exceptions(KAFKA_METRICS_AVRO_TO_METRIC_SERIES_MV(), node_roles=[NodeRole.LOGS]),
+ run_sql_with_exceptions(KAFKA_METRICS_AVRO_TO_KAFKA_METRICS_MV(), node_roles=[NodeRole.LOGS]),
+]
diff --git a/posthog/clickhouse/migrations/max_migration.txt b/posthog/clickhouse/migrations/max_migration.txt
index 8ce74c420526..237df0f08dd1 100644
--- a/posthog/clickhouse/migrations/max_migration.txt
+++ b/posthog/clickhouse/migrations/max_migration.txt
@@ -1 +1 @@
-0307_logs_pattern_kafka
+0308_metrics_kafka_ingest
diff --git a/posthog/clickhouse/test/__snapshots__/test_schema.ambr b/posthog/clickhouse/test/__snapshots__/test_schema.ambr
index c671d0535e80..b6bffbdc990a 100644
--- a/posthog/clickhouse/test/__snapshots__/test_schema.ambr
+++ b/posthog/clickhouse/test/__snapshots__/test_schema.ambr
@@ -4137,12 +4137,12 @@
CREATE TABLE IF NOT EXISTS posthog_test.metric_attributes
(
- `team_id` Int32 CODEC(DoubleDelta, ZSTD(1)),
- `time_bucket` DateTime64(0) CODEC(DoubleDelta, ZSTD(1)),
- `service_name` LowCardinality(String) CODEC(ZSTD(1)),
- `resource_fingerprint` UInt64 DEFAULT 0 CODEC(DoubleDelta, ZSTD(1)),
- `attribute_key` LowCardinality(String) CODEC(ZSTD(1)),
- `attribute_value` String CODEC(ZSTD(1)),
+ `team_id` Int32,
+ `time_bucket` DateTime64(0),
+ `service_name` LowCardinality(String),
+ `resource_fingerprint` UInt64 DEFAULT 0,
+ `attribute_key` LowCardinality(String),
+ `attribute_value` String,
`attribute_count` SimpleAggregateFunction(sum, UInt64),
`attribute_type` LowCardinality(String),
INDEX idx_attribute_key attribute_key TYPE bloom_filter(0.01) GRANULARITY 1,
@@ -4223,30 +4223,30 @@
CREATE TABLE IF NOT EXISTS posthog_test.metrics1
(
- `time_bucket` DateTime MATERIALIZED toStartOfDay(timestamp) CODEC(DoubleDelta, ZSTD(1)),
- `uuid` String CODEC(ZSTD(1)),
- `team_id` Int32 CODEC(ZSTD(1)),
- `trace_id` String CODEC(ZSTD(1)),
- `span_id` String CODEC(ZSTD(1)),
- `trace_flags` Int32 CODEC(ZSTD(1)),
- `timestamp` DateTime64(6) CODEC(DoubleDelta, ZSTD(1)),
- `observed_timestamp` DateTime64(6) CODEC(DoubleDelta, ZSTD(1)),
- `created_at` DateTime64(6) MATERIALIZED now() CODEC(DoubleDelta, ZSTD(1)),
- `service_name` LowCardinality(String) CODEC(ZSTD(1)),
- `metric_name` LowCardinality(String) CODEC(ZSTD(1)),
- `metric_type` LowCardinality(String) CODEC(ZSTD(1)),
- `value` Float64 CODEC(Gorilla, ZSTD(1)),
- `count` UInt64 DEFAULT 1 CODEC(T64, ZSTD(1)),
- `histogram_bounds` Array(Float64) CODEC(ZSTD(1)),
- `histogram_counts` Array(UInt64) CODEC(ZSTD(1)),
- `unit` LowCardinality(String) CODEC(ZSTD(1)),
- `aggregation_temporality` LowCardinality(String) CODEC(ZSTD(1)),
- `is_monotonic` Bool DEFAULT false CODEC(ZSTD(1)),
- `resource_attributes` Map(LowCardinality(String), String) CODEC(ZSTD(1)),
- `resource_fingerprint` UInt64 MATERIALIZED cityHash64(resource_attributes) CODEC(DoubleDelta, ZSTD(1)),
- `instrumentation_scope` String CODEC(ZSTD(1)),
- `attributes_map_str` Map(LowCardinality(String), String) CODEC(ZSTD(1)),
- `attributes_map_float` Map(LowCardinality(String), Float64) CODEC(ZSTD(1)),
+ `time_bucket` DateTime MATERIALIZED toStartOfDay(timestamp),
+ `uuid` String,
+ `team_id` Int32,
+ `trace_id` String,
+ `span_id` String,
+ `trace_flags` Int32,
+ `timestamp` DateTime64(6),
+ `observed_timestamp` DateTime64(6),
+ `created_at` DateTime64(6) MATERIALIZED now(),
+ `service_name` LowCardinality(String),
+ `metric_name` LowCardinality(String),
+ `metric_type` LowCardinality(String),
+ `value` Float64 CODEC(Gorilla),
+ `count` UInt64 DEFAULT 1 CODEC(T64),
+ `histogram_bounds` Array(Float64),
+ `histogram_counts` Array(UInt64),
+ `unit` LowCardinality(String),
+ `aggregation_temporality` LowCardinality(String),
+ `is_monotonic` Bool DEFAULT false,
+ `resource_attributes` Map(LowCardinality(String), String),
+ `resource_fingerprint` UInt64 MATERIALIZED cityHash64(resource_attributes),
+ `instrumentation_scope` String,
+ `attributes_map_str` Map(LowCardinality(String), String),
+ `attributes_map_float` Map(LowCardinality(String), Float64),
`time_minute` DateTime ALIAS toStartOfMinute(timestamp),
`attributes` Map(String, String) ALIAS mapApply((k, v) -> (left(k, -5), v), attributes_map_str),
INDEX idx_metric_name_set metric_name TYPE set(100) GRANULARITY 1,
@@ -10899,12 +10899,12 @@
CREATE TABLE IF NOT EXISTS posthog_test.metric_attributes
(
- `team_id` Int32 CODEC(DoubleDelta, ZSTD(1)),
- `time_bucket` DateTime64(0) CODEC(DoubleDelta, ZSTD(1)),
- `service_name` LowCardinality(String) CODEC(ZSTD(1)),
- `resource_fingerprint` UInt64 DEFAULT 0 CODEC(DoubleDelta, ZSTD(1)),
- `attribute_key` LowCardinality(String) CODEC(ZSTD(1)),
- `attribute_value` String CODEC(ZSTD(1)),
+ `team_id` Int32,
+ `time_bucket` DateTime64(0),
+ `service_name` LowCardinality(String),
+ `resource_fingerprint` UInt64 DEFAULT 0,
+ `attribute_key` LowCardinality(String),
+ `attribute_value` String,
`attribute_count` SimpleAggregateFunction(sum, UInt64),
`attribute_type` LowCardinality(String),
INDEX idx_attribute_key attribute_key TYPE bloom_filter(0.01) GRANULARITY 1,
@@ -10979,30 +10979,30 @@
CREATE TABLE IF NOT EXISTS posthog_test.metrics1
(
- `time_bucket` DateTime MATERIALIZED toStartOfDay(timestamp) CODEC(DoubleDelta, ZSTD(1)),
- `uuid` String CODEC(ZSTD(1)),
- `team_id` Int32 CODEC(ZSTD(1)),
- `trace_id` String CODEC(ZSTD(1)),
- `span_id` String CODEC(ZSTD(1)),
- `trace_flags` Int32 CODEC(ZSTD(1)),
- `timestamp` DateTime64(6) CODEC(DoubleDelta, ZSTD(1)),
- `observed_timestamp` DateTime64(6) CODEC(DoubleDelta, ZSTD(1)),
- `created_at` DateTime64(6) MATERIALIZED now() CODEC(DoubleDelta, ZSTD(1)),
- `service_name` LowCardinality(String) CODEC(ZSTD(1)),
- `metric_name` LowCardinality(String) CODEC(ZSTD(1)),
- `metric_type` LowCardinality(String) CODEC(ZSTD(1)),
- `value` Float64 CODEC(Gorilla, ZSTD(1)),
- `count` UInt64 DEFAULT 1 CODEC(T64, ZSTD(1)),
- `histogram_bounds` Array(Float64) CODEC(ZSTD(1)),
- `histogram_counts` Array(UInt64) CODEC(ZSTD(1)),
- `unit` LowCardinality(String) CODEC(ZSTD(1)),
- `aggregation_temporality` LowCardinality(String) CODEC(ZSTD(1)),
- `is_monotonic` Bool DEFAULT false CODEC(ZSTD(1)),
- `resource_attributes` Map(LowCardinality(String), String) CODEC(ZSTD(1)),
- `resource_fingerprint` UInt64 MATERIALIZED cityHash64(resource_attributes) CODEC(DoubleDelta, ZSTD(1)),
- `instrumentation_scope` String CODEC(ZSTD(1)),
- `attributes_map_str` Map(LowCardinality(String), String) CODEC(ZSTD(1)),
- `attributes_map_float` Map(LowCardinality(String), Float64) CODEC(ZSTD(1)),
+ `time_bucket` DateTime MATERIALIZED toStartOfDay(timestamp),
+ `uuid` String,
+ `team_id` Int32,
+ `trace_id` String,
+ `span_id` String,
+ `trace_flags` Int32,
+ `timestamp` DateTime64(6),
+ `observed_timestamp` DateTime64(6),
+ `created_at` DateTime64(6) MATERIALIZED now(),
+ `service_name` LowCardinality(String),
+ `metric_name` LowCardinality(String),
+ `metric_type` LowCardinality(String),
+ `value` Float64 CODEC(Gorilla),
+ `count` UInt64 DEFAULT 1 CODEC(T64),
+ `histogram_bounds` Array(Float64),
+ `histogram_counts` Array(UInt64),
+ `unit` LowCardinality(String),
+ `aggregation_temporality` LowCardinality(String),
+ `is_monotonic` Bool DEFAULT false,
+ `resource_attributes` Map(LowCardinality(String), String),
+ `resource_fingerprint` UInt64 MATERIALIZED cityHash64(resource_attributes),
+ `instrumentation_scope` String,
+ `attributes_map_str` Map(LowCardinality(String), String),
+ `attributes_map_float` Map(LowCardinality(String), Float64),
`time_minute` DateTime ALIAS toStartOfMinute(timestamp),
`attributes` Map(String, String) ALIAS mapApply((k, v) -> (left(k, -5), v), attributes_map_str),
INDEX idx_metric_name_set metric_name TYPE set(100) GRANULARITY 1,