feat(metrics): Add a meta table to counters

snuba/migrations/group_loader.py

@@ -318,6 +318,9 @@ def get_migrations(self) -> Sequence[str]:
             "0027_sets_add_raw_tags_column",
             "0028_distributions_add_indexed_tags_column",
             "0029_add_use_case_id_index",
+            "0030_add_record_meta_column",
+            "0031_counters_meta_table",
+            "0032_counters_meta_table_mv",
         ]
 
 

snuba/snuba_migrations/generic_metrics/0030_add_record_meta_column.py

@@ -0,0 +1,33 @@
+from typing import Sequence
+
+from snuba.clickhouse.columns import Column, UInt
+from snuba.clusters.storage_sets import StorageSetKey
+from snuba.migrations import migration, operations
+from snuba.migrations.columns import MigrationModifiers as Modifiers
+
+
+class Migration(migration.ClickhouseNodeMigration):
+    blocking = False
+    local_table_name = "generic_metric_counters_raw_local"
+    storage_set_key = StorageSetKey.GENERIC_METRICS_COUNTERS
+
+    def forwards_ops(self) -> Sequence[operations.SqlOperation]:
+        return [
+            operations.AddColumn(
+                storage_set=self.storage_set_key,
+                table_name=self.local_table_name,
+                column=Column("record_meta", UInt(8, Modifiers(default=str("0")))),
+                target=operations.OperationTarget.LOCAL,
+                after="materialization_version",
+            ),
+        ]
+
+    def backwards_ops(self) -> Sequence[operations.SqlOperation]:
+        return [
+            operations.DropColumn(
+                column_name="record_meta",
+                storage_set=self.storage_set_key,
+                table_name=self.local_table_name,
+                target=operations.OperationTarget.LOCAL,
+            ),
+        ]

snuba/snuba_migrations/generic_metrics/0031_counters_meta_table.py

@@ -0,0 +1,75 @@
+from typing import Sequence
+
+from snuba.clickhouse.columns import (
+    AggregateFunction,
+    Array,
+    Column,
+    DateTime,
+    String,
+    UInt,
+)
+from snuba.clusters.storage_sets import StorageSetKey
+from snuba.migrations import migration, operations, table_engines
+from snuba.migrations.columns import MigrationModifiers as Modifiers
+from snuba.migrations.operations import OperationTarget
+from snuba.utils.schemas import Float
+
+
+class Migration(migration.ClickhouseNodeMigration):
+    blocking = False
+    granularity = "2048"
+    local_table_name = "generic_metric_counters_meta_aggregated_local"
+    dist_table_name = "generic_metric_counters_meta_aggregated_dist"
+    storage_set_key = StorageSetKey.GENERIC_METRICS_COUNTERS
+    columns: Sequence[Column[Modifiers]] = [
+        Column("org_id", UInt(64)),
+        Column("project_id", UInt(64)),
+        Column("use_case_id", String(Modifiers(low_cardinality=True))),
+        Column("metric_id", UInt(64)),
+        Column("tag_key", String()),
+        Column("timestamp", DateTime(modifiers=Modifiers(codecs=["DoubleDelta"]))),
+        Column("retention_days", UInt(16)),
+        Column("tag_values", AggregateFunction("groupUniqArray", [Array(String())])),
+        Column("count", AggregateFunction("sum", [Float(64)])),
+    ]
+
+    def forwards_ops(self) -> Sequence[operations.SqlOperation]:
+        return [
+            operations.CreateTable(
+                storage_set=self.storage_set_key,
+                table_name=self.local_table_name,
+                engine=table_engines.AggregatingMergeTree(
+                    storage_set=self.storage_set_key,
+                    order_by="(org_id, project_id, use_case_id, metric_id, tag_key, timestamp)",
+                    primary_key="(org_id, project_id, use_case_id, metric_id, tag_key, timestamp)",
+                    partition_by="(retention_days, toMonday(timestamp))",
+                    settings={"index_granularity": self.granularity},
+                    ttl="timestamp + toIntervalDay(retention_days)",
+                ),
+                columns=self.columns,
+                target=OperationTarget.LOCAL,
+            ),
+            operations.CreateTable(
+                storage_set=self.storage_set_key,
+                table_name=self.dist_table_name,
+                engine=table_engines.Distributed(
+                    local_table_name=self.local_table_name, sharding_key=None
+                ),
+                columns=self.columns,
+                target=OperationTarget.DISTRIBUTED,
+            ),
+        ]
+
+    def backwards_ops(self) -> Sequence[operations.SqlOperation]:
+        return [
+            operations.DropTable(
+                storage_set=self.storage_set_key,
+                table_name=self.dist_table_name,
+                target=OperationTarget.DISTRIBUTED,
+            ),
+            operations.DropTable(
+                storage_set=self.storage_set_key,
+                table_name=self.local_table_name,
+                target=OperationTarget.LOCAL,
+            ),
+        ]

snuba/snuba_migrations/generic_metrics/0032_counters_meta_table_mv.py

@@ -0,0 +1,77 @@
+from typing import Sequence
+
+from snuba.clickhouse.columns import (
+    AggregateFunction,
+    Array,
+    Column,
+    DateTime,
+    String,
+    UInt,
+)
+from snuba.clusters.storage_sets import StorageSetKey
+from snuba.migrations import migration, operations
+from snuba.migrations.columns import MigrationModifiers as Modifiers
+from snuba.migrations.operations import OperationTarget
+from snuba.utils.schemas import Float
+
+
+class Migration(migration.ClickhouseNodeMigration):
+    blocking = False
+    view_name = "generic_metric_counters_meta_aggregation_mv"
+    dest_table_name = "generic_metric_counters_meta_aggregated_local"
+    dest_table_columns: Sequence[Column[Modifiers]] = [
+        Column("org_id", UInt(64)),
+        Column("project_id", UInt(64)),
+        Column("use_case_id", String(Modifiers(low_cardinality=True))),
+        Column("metric_id", UInt(64)),
+        Column("tag_key", String()),
+        Column("timestamp", DateTime(modifiers=Modifiers(codecs=["DoubleDelta"]))),
+        Column("retention_days", UInt(16)),
+        Column("tag_values", AggregateFunction("groupUniqArray", [Array(String())])),
+        Column("value", AggregateFunction("sum", [Float(64)])),
+    ]
+    storage_set_key = StorageSetKey.GENERIC_METRICS_COUNTERS
+
+    def forwards_ops(self) -> Sequence[operations.SqlOperation]:
+        return [
+            operations.CreateMaterializedView(
+                storage_set=self.storage_set_key,
+                view_name=self.view_name,
+                columns=self.dest_table_columns,
+                destination_table_name=self.dest_table_name,
+                target=OperationTarget.LOCAL,
+                query="""
+                SELECT
+                    org_id,
+                    project_id,
+                    use_case_id,
+                    metric_id,
+                    tag_key,
+                    toStartOfWeek(timestamp) as timestamp,
+                    retention_days,
+                    groupUniqArrayState(tag_value) as `tag_values`,
+                    sumState(count_value) as count
+                FROM generic_metric_counters_raw_local
+                ARRAY JOIN
+                    tags.key AS tag_key, tags.raw_value AS tag_value
+                WHERE record_meta = 1
+                GROUP BY
+                    org_id,
+                    project_id,
+                    use_case_id,
+                    metric_id,
+                    tag_key,
+                    timestamp,
+                    retention_days
+                """,
+            ),
+        ]
+
+    def backwards_ops(self) -> Sequence[operations.SqlOperation]:
+        return [
+            operations.DropTable(
+                storage_set=self.storage_set_key,
+                table_name=self.view_name,
+                target=OperationTarget.LOCAL,
+            )
+        ]