feat: add daily and epoch level aggregation tables for attestation_liveness_by_entity

buf.lock

@@ -0,0 +1,6 @@
+# Generated by buf. DO NOT EDIT.
+version: v2
+deps:
+  - name: buf.build/googleapis/googleapis
+    commit: 72c8614f3bd0466ea67931ef2c43d608
+    digest: b5:13efeea24e633fd45327390bdee941207a8727e96cf01affb84c1e4100fd8f48a42bbd508df11930cd2884629bafad685df1ac3111bc78cdaefcd38c9371c6b1

migrations/025_attestation_liveness_by_entity_epoch_head.down.sql

@@ -0,0 +1,2 @@
+DROP TABLE IF EXISTS `${NETWORK_NAME}`.fct_attestation_liveness_by_entity_epoch_head ON CLUSTER '{cluster}';
+DROP TABLE IF EXISTS `${NETWORK_NAME}`.fct_attestation_liveness_by_entity_epoch_head_local ON CLUSTER '{cluster}';

migrations/025_attestation_liveness_by_entity_epoch_head.up.sql

@@ -0,0 +1,36 @@
+-- Create local table with ReplicatedReplacingMergeTree
+CREATE TABLE `${NETWORK_NAME}`.fct_attestation_liveness_by_entity_epoch_head_local ON CLUSTER '{cluster}' (
+    `updated_date_time` DateTime COMMENT 'Timestamp when the record was last updated' CODEC(DoubleDelta, ZSTD(1)),
+    `epoch` UInt32 COMMENT 'The epoch number' CODEC(DoubleDelta, ZSTD(1)),
+    `epoch_start_date_time` DateTime COMMENT 'The wall clock time when the epoch started' CODEC(DoubleDelta, ZSTD(1)),
+    `entity` String COMMENT 'The entity (staking provider) associated with the validators, unknown if not mapped' CODEC(ZSTD(1)),
+    `status` String COMMENT 'Attestation status: attested or missed' CODEC(ZSTD(1)),
+    `attestation_count` UInt64 COMMENT 'Sum of attestations for this epoch/entity/status combination' CODEC(ZSTD(1))
+) ENGINE = ReplicatedReplacingMergeTree(
+    '/clickhouse/{installation}/{cluster}/tables/{shard}/{database}/{table}',
+    '{replica}',
+    `updated_date_time`
+) PARTITION BY toStartOfMonth(epoch_start_date_time)
+ORDER BY
+    (`epoch_start_date_time`, `entity`, `status`)
+SETTINGS
+    deduplicate_merge_projection_mode = 'rebuild',
+    min_age_to_force_merge_seconds = 384,
+    min_age_to_force_merge_on_partition_only=false
+COMMENT 'Attestation liveness aggregated by entity and epoch for the head chain. Reduces slot-level data by ~32x.';
+
+-- Create distributed table wrapper
+CREATE TABLE `${NETWORK_NAME}`.fct_attestation_liveness_by_entity_epoch_head ON CLUSTER '{cluster}' AS `${NETWORK_NAME}`.fct_attestation_liveness_by_entity_epoch_head_local ENGINE = Distributed(
+    '{cluster}',
+    '${NETWORK_NAME}',
+    fct_attestation_liveness_by_entity_epoch_head_local,
+    cityHash64(`epoch`)
+);
+
+-- Add secondary projection for efficient epoch number queries
+ALTER TABLE `${NETWORK_NAME}`.fct_attestation_liveness_by_entity_epoch_head_local ON CLUSTER '{cluster}'
+ADD PROJECTION p_by_epoch
+(
+    SELECT *
+    ORDER BY (`epoch`)
+);

migrations/026_attestation_liveness_by_entity_day_head.down.sql

@@ -0,0 +1,2 @@
+DROP TABLE IF EXISTS `${NETWORK_NAME}`.fct_attestation_liveness_by_entity_day_head ON CLUSTER '{cluster}';
+DROP TABLE IF EXISTS `${NETWORK_NAME}`.fct_attestation_liveness_by_entity_day_head_local ON CLUSTER '{cluster}';

migrations/026_attestation_liveness_by_entity_day_head.up.sql

@@ -0,0 +1,27 @@
+-- Create local table with ReplicatedReplacingMergeTree
+CREATE TABLE `${NETWORK_NAME}`.fct_attestation_liveness_by_entity_day_head_local ON CLUSTER '{cluster}' (
+    `updated_date_time` DateTime COMMENT 'Timestamp when the record was last updated' CODEC(DoubleDelta, ZSTD(1)),
+    `date` Date COMMENT 'The calendar date' CODEC(DoubleDelta, ZSTD(1)),
+    `entity` String COMMENT 'The entity (staking provider) associated with the validators, unknown if not mapped' CODEC(ZSTD(1)),
+    `status` String COMMENT 'Attestation status: attested or missed' CODEC(ZSTD(1)),
+    `attestation_count` UInt64 COMMENT 'Sum of attestations for this date/entity/status combination' CODEC(ZSTD(1))
+) ENGINE = ReplicatedReplacingMergeTree(
+    '/clickhouse/{installation}/{cluster}/tables/{shard}/{database}/{table}',
+    '{replica}',
+    `updated_date_time`
+) PARTITION BY toStartOfMonth(date)
+ORDER BY
+    (`date`, `entity`, `status`)
+SETTINGS
+    deduplicate_merge_projection_mode = 'rebuild',
+    min_age_to_force_merge_seconds = 384,
+    min_age_to_force_merge_on_partition_only=false
+COMMENT 'Attestation liveness aggregated by entity and day for the head chain. Reduces epoch-level data by ~225x.';
+
+-- Create distributed table wrapper
+CREATE TABLE `${NETWORK_NAME}`.fct_attestation_liveness_by_entity_day_head ON CLUSTER '{cluster}' AS `${NETWORK_NAME}`.fct_attestation_liveness_by_entity_day_head_local ENGINE = Distributed(
+    '{cluster}',
+    '${NETWORK_NAME}',
+    fct_attestation_liveness_by_entity_day_head_local,
+    cityHash64(`date`)
+);

models/transformations/fct_attestation_liveness_by_entity_day_head.sql

@@ -0,0 +1,17 @@
+---
+table: fct_attestation_liveness_by_entity_day_head
+type: incremental
+interval:
+  type: slot
+  max: 50000
+---
+INSERT INTO `{{ .self.database }}`.`{{ .self.table }}`
+SELECT
+    fromUnixTimestamp({{ .task.start }}) as updated_date_time,
+    toDate(epoch_start_date_time) as date,
+    entity,
+    status,
+    SUM(attestation_count) as attestation_count
+FROM {{ index .dep "{{transformation}}" "fct_attestation_liveness_by_entity_epoch_head" "helpers" "from" }} FINAL
+WHERE epoch_start_date_time BETWEEN fromUnixTimestamp({{ .bounds.start }}) AND fromUnixTimestamp({{ .bounds.end }})
+GROUP BY toDate(epoch_start_date_time), entity, status

models/transformations/fct_attestation_liveness_by_entity_epoch_head.sql

@@ -0,0 +1,18 @@
+---
+table: fct_attestation_liveness_by_entity_epoch_head
+type: incremental
+interval:
+  type: slot
+  max: 50000
+---
+INSERT INTO `{{ .self.database }}`.`{{ .self.table }}`
+SELECT
+    fromUnixTimestamp({{ .task.start }}) as updated_date_time,
+    epoch,
+    epoch_start_date_time,
+    entity,
+    status,
+    SUM(attestation_count) as attestation_count
+FROM {{ index .dep "{{transformation}}" "fct_attestation_liveness_by_entity_head" "helpers" "from" }} FINAL
+WHERE slot_start_date_time BETWEEN fromUnixTimestamp({{ .bounds.start }}) AND fromUnixTimestamp({{ .bounds.end }})
+GROUP BY epoch, epoch_start_date_time, entity, status