From 6dbddc6ec3e247e34ff189c525ee15d7fc48d424 Mon Sep 17 00:00:00 2001 From: Oliver Newland Date: Tue, 1 Sep 2026 15:06:37 -0700 Subject: [PATCH 1/2] feat(eap-items): add version column to eap_items_1 Adds a UInt64 `version` column defaulting to the current time in milliseconds to eap_items_1_local and eap_items_1_dist. This is a schema-only step toward moving eap_items to a ReplacingMergeTree keyed on `version`. The column needs to exist on the current table before old parts can be attached to the new table. --- .../0064_add_version_column.py | 60 +++++++++++++++++++ 1 file changed, 60 insertions(+) create mode 100644 snuba/snuba_migrations/events_analytics_platform/0064_add_version_column.py diff --git a/snuba/snuba_migrations/events_analytics_platform/0064_add_version_column.py b/snuba/snuba_migrations/events_analytics_platform/0064_add_version_column.py new file mode 100644 index 0000000000..5eb8b9c102 --- /dev/null +++ b/snuba/snuba_migrations/events_analytics_platform/0064_add_version_column.py @@ -0,0 +1,60 @@ +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 +from snuba.migrations.operations import OperationTarget + +storage_set = StorageSetKey.EVENTS_ANALYTICS_PLATFORM +table_name_prefix = "eap_items_1" + +new_column_name = "version" + +# Milliseconds since the unix epoch at the time the row is written. +default_expression = "toUnixTimestamp64Milli(now64(3))" + +new_column: Column[Modifiers] = Column( + new_column_name, + UInt( + 64, + modifiers=Modifiers( + default=default_expression, + codecs=["ZSTD(1)"], + ), + ), +) + + +class Migration(migration.ClickhouseNodeMigration): + blocking = False + + def forwards_ops(self) -> list[operations.SqlOperation]: + return [ + operations.AddColumn( + storage_set=storage_set, + table_name=f"{table_name_prefix}_local", + column=new_column, + target=OperationTarget.LOCAL, + ), + operations.AddColumn( + storage_set=storage_set, + table_name=f"{table_name_prefix}_dist", + column=new_column, + target=OperationTarget.DISTRIBUTED, + ), + ] + + def backwards_ops(self) -> list[operations.SqlOperation]: + return [ + operations.DropColumn( + storage_set=storage_set, + table_name=f"{table_name_prefix}_dist", + column_name=new_column_name, + target=OperationTarget.DISTRIBUTED, + ), + operations.DropColumn( + storage_set=storage_set, + table_name=f"{table_name_prefix}_local", + column_name=new_column_name, + target=OperationTarget.LOCAL, + ), + ] From d158ec508e4aec92fe73b12cf7563f071152b775 Mon Sep 17 00:00:00 2001 From: Oliver Newland Date: Tue, 1 Sep 2026 15:18:36 -0700 Subject: [PATCH 2/2] ref(eap-items): use asymmetric defaults for version column ADD COLUMN does not rewrite existing parts, so rows predating this migration have no `version` on disk and the local DEFAULT is evaluated at read/merge time. Under ReplacingMergeTree(version) a now64() default on the local table would give those legacy rows a version of whenever the merge ran -- always newer than a real write -- letting a stale row silently beat a genuine update. The local table now defaults to 0, the sentinel for rows written before versioning existed, which deterministically loses to every real write. Inserts go through the distributed table, which evaluates its own DEFAULT on the initiator and ships the materialized value to the shard, so the dist table keeps the now64() default and live writes still get a real millisecond timestamp. --- .../0064_add_version_column.py | 47 ++++++++++++++++--- 1 file changed, 41 insertions(+), 6 deletions(-) diff --git a/snuba/snuba_migrations/events_analytics_platform/0064_add_version_column.py b/snuba/snuba_migrations/events_analytics_platform/0064_add_version_column.py index 5eb8b9c102..af4cac1599 100644 --- a/snuba/snuba_migrations/events_analytics_platform/0064_add_version_column.py +++ b/snuba/snuba_migrations/events_analytics_platform/0064_add_version_column.py @@ -9,20 +9,55 @@ new_column_name = "version" -# Milliseconds since the unix epoch at the time the row is written. -default_expression = "toUnixTimestamp64Milli(now64(3))" +# The local and distributed tables deliberately carry *different* DEFAULT +# expressions. This is not an oversight; see below. -new_column: Column[Modifiers] = Column( +# ADD COLUMN does not rewrite existing parts, so every row written before this +# migration has no `version` on disk and the local DEFAULT is what gets +# evaluated when those rows are read or merged. Once eap_items moves to +# ReplacingMergeTree(version), a non-deterministic local default such as +# now64() would hand those legacy rows a version of "whenever the merge +# happened", which is always newer than a genuine write -- so a stale legacy row +# would silently win over a real update. `0` is therefore the sentinel for +# "written before versioning existed", and it deterministically loses to +# everything else. +local_column: Column[Modifiers] = Column( new_column_name, UInt( 64, modifiers=Modifiers( - default=default_expression, + default="0", codecs=["ZSTD(1)"], ), ), ) +# Inserts arrive through the distributed table, and a Distributed table +# evaluates its own DEFAULT expressions on the initiator node and then ships the +# materialized value to the shard -- the local DEFAULT never fires for that +# path. (Declaring the column here *without* a default would not fall through to +# the local default either; it would ship an explicit 0.) So this is the default +# that actually assigns a version to live writes, until the consumer starts +# populating the column explicitly. +dist_column: Column[Modifiers] = Column( + new_column_name, + UInt( + 64, + modifiers=Modifiers( + default="toUnixTimestamp64Milli(now64(3))", + codecs=["ZSTD(1)"], + ), + ), +) + +# `version` is intentionally not added to eap_items_1_dist_ro. It is internal +# bookkeeping for the ReplacingMergeTree cutover and is absent from the storage +# YAML, so nothing the query layer generates can reference it. Migration 0056 +# creates the _dist_ro tables before this migration runs, so fresh and existing +# deployments both end up without the column there -- no environment drift. +# If we ever want to read `version` through the read-only routing path, it +# should be added in its own migration. + class Migration(migration.ClickhouseNodeMigration): blocking = False @@ -32,13 +67,13 @@ def forwards_ops(self) -> list[operations.SqlOperation]: operations.AddColumn( storage_set=storage_set, table_name=f"{table_name_prefix}_local", - column=new_column, + column=local_column, target=OperationTarget.LOCAL, ), operations.AddColumn( storage_set=storage_set, table_name=f"{table_name_prefix}_dist", - column=new_column, + column=dist_column, target=OperationTarget.DISTRIBUTED, ), ]