From 46f663c8bdb7f8ead6d4d6e73c7202b8726a984a Mon Sep 17 00:00:00 2001 From: Oliver Newland Date: Wed, 2 Sep 2026 17:05:41 -0700 Subject: [PATCH 1/4] feat(eap-items): add version column to eap_items_1_downsample_8_local Mirrors the version column added to eap_items_1 in 0064, aligning the downsample_8 tier's schema with its target ReplacingMergeTree ahead of attaching existing parts. Only the local table is touched. We never query `version`, and the downsample tiers are fed by materialized views that write straight to *_local, so the distributed table is neither a read nor an insert path for this column. Defaults to 0, the sentinel for rows written before versioning existed, which deterministically loses to every real write. --- .../0065_add_version_column_downsample_8.py | 66 +++++++++++++++++++ 1 file changed, 66 insertions(+) create mode 100644 snuba/snuba_migrations/events_analytics_platform/0065_add_version_column_downsample_8.py diff --git a/snuba/snuba_migrations/events_analytics_platform/0065_add_version_column_downsample_8.py b/snuba/snuba_migrations/events_analytics_platform/0065_add_version_column_downsample_8.py new file mode 100644 index 0000000000..92438fe3a4 --- /dev/null +++ b/snuba/snuba_migrations/events_analytics_platform/0065_add_version_column_downsample_8.py @@ -0,0 +1,66 @@ +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 +local_table_name = "eap_items_1_downsample_8_local" + +new_column_name = "version" + +# Mirrors the `version` column added to eap_items_1 in migration 0064, so this +# downsample tier's schema lines up with its target ReplacingMergeTree before +# existing parts are attached onto it. +# +# ADD COLUMN does not rewrite existing parts, so rows written before this +# migration have no `version` on disk and this DEFAULT is what gets evaluated +# when they are read or merged. It therefore has to be deterministic: under +# ReplacingMergeTree(version) a now64() default would hand those legacy rows a +# version of "whenever the merge happened", which is always newer than a real +# write, letting a stale row silently beat a genuine update. `0` is the sentinel +# for "written before versioning existed" and deterministically loses to +# everything else. +new_column: Column[Modifiers] = Column( + new_column_name, + UInt( + 64, + modifiers=Modifiers( + default="0", + codecs=["ZSTD(1)"], + ), + ), +) + +# Only the *_local table gets the column, unlike 0064 which also touched +# eap_items_1_dist. `version` is internal bookkeeping for the +# ReplacingMergeTree cutover: it is absent from the storage YAML so we never +# query it, and the downsample tiers are fed by materialized views that write +# straight to *_local, so the distributed table is never an insert target +# either. A Distributed table that declares a subset of its local table's +# columns reads normally; only a query naming `version` through the dist table +# would fail, and nothing does that. The same reasoning covers *_dist_ro. + + +class Migration(migration.ClickhouseNodeMigration): + blocking = False + + def forwards_ops(self) -> list[operations.SqlOperation]: + return [ + operations.AddColumn( + storage_set=storage_set, + table_name=local_table_name, + column=new_column, + target=OperationTarget.LOCAL, + ), + ] + + def backwards_ops(self) -> list[operations.SqlOperation]: + return [ + operations.DropColumn( + storage_set=storage_set, + table_name=local_table_name, + column_name=new_column_name, + target=OperationTarget.LOCAL, + ), + ] From 369eaff475ef8779c5723465623b576192fc8809 Mon Sep 17 00:00:00 2001 From: Oliver Newland Date: Wed, 2 Sep 2026 17:05:42 -0700 Subject: [PATCH 2/4] feat(eap-items): add version column to eap_items_1_downsample_64_local Mirrors the version column added to eap_items_1 in 0064, aligning the downsample_64 tier's schema with its target ReplacingMergeTree ahead of attaching existing parts. Only the local table is touched. We never query `version`, and the downsample tiers are fed by materialized views that write straight to *_local, so the distributed table is neither a read nor an insert path for this column. Defaults to 0, the sentinel for rows written before versioning existed, which deterministically loses to every real write. --- .../0066_add_version_column_downsample_64.py | 66 +++++++++++++++++++ 1 file changed, 66 insertions(+) create mode 100644 snuba/snuba_migrations/events_analytics_platform/0066_add_version_column_downsample_64.py diff --git a/snuba/snuba_migrations/events_analytics_platform/0066_add_version_column_downsample_64.py b/snuba/snuba_migrations/events_analytics_platform/0066_add_version_column_downsample_64.py new file mode 100644 index 0000000000..6d5cc02b97 --- /dev/null +++ b/snuba/snuba_migrations/events_analytics_platform/0066_add_version_column_downsample_64.py @@ -0,0 +1,66 @@ +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 +local_table_name = "eap_items_1_downsample_64_local" + +new_column_name = "version" + +# Mirrors the `version` column added to eap_items_1 in migration 0064, so this +# downsample tier's schema lines up with its target ReplacingMergeTree before +# existing parts are attached onto it. +# +# ADD COLUMN does not rewrite existing parts, so rows written before this +# migration have no `version` on disk and this DEFAULT is what gets evaluated +# when they are read or merged. It therefore has to be deterministic: under +# ReplacingMergeTree(version) a now64() default would hand those legacy rows a +# version of "whenever the merge happened", which is always newer than a real +# write, letting a stale row silently beat a genuine update. `0` is the sentinel +# for "written before versioning existed" and deterministically loses to +# everything else. +new_column: Column[Modifiers] = Column( + new_column_name, + UInt( + 64, + modifiers=Modifiers( + default="0", + codecs=["ZSTD(1)"], + ), + ), +) + +# Only the *_local table gets the column, unlike 0064 which also touched +# eap_items_1_dist. `version` is internal bookkeeping for the +# ReplacingMergeTree cutover: it is absent from the storage YAML so we never +# query it, and the downsample tiers are fed by materialized views that write +# straight to *_local, so the distributed table is never an insert target +# either. A Distributed table that declares a subset of its local table's +# columns reads normally; only a query naming `version` through the dist table +# would fail, and nothing does that. The same reasoning covers *_dist_ro. + + +class Migration(migration.ClickhouseNodeMigration): + blocking = False + + def forwards_ops(self) -> list[operations.SqlOperation]: + return [ + operations.AddColumn( + storage_set=storage_set, + table_name=local_table_name, + column=new_column, + target=OperationTarget.LOCAL, + ), + ] + + def backwards_ops(self) -> list[operations.SqlOperation]: + return [ + operations.DropColumn( + storage_set=storage_set, + table_name=local_table_name, + column_name=new_column_name, + target=OperationTarget.LOCAL, + ), + ] From 6e61e23fae01360841c7d457716d4222d0b2de37 Mon Sep 17 00:00:00 2001 From: Oliver Newland Date: Wed, 2 Sep 2026 17:05:44 -0700 Subject: [PATCH 3/4] feat(eap-items): add version column to eap_items_1_downsample_512_local Mirrors the version column added to eap_items_1 in 0064, aligning the downsample_512 tier's schema with its target ReplacingMergeTree ahead of attaching existing parts. Only the local table is touched. We never query `version`, and the downsample tiers are fed by materialized views that write straight to *_local, so the distributed table is neither a read nor an insert path for this column. Defaults to 0, the sentinel for rows written before versioning existed, which deterministically loses to every real write. --- .../0067_add_version_column_downsample_512.py | 66 +++++++++++++++++++ 1 file changed, 66 insertions(+) create mode 100644 snuba/snuba_migrations/events_analytics_platform/0067_add_version_column_downsample_512.py diff --git a/snuba/snuba_migrations/events_analytics_platform/0067_add_version_column_downsample_512.py b/snuba/snuba_migrations/events_analytics_platform/0067_add_version_column_downsample_512.py new file mode 100644 index 0000000000..ced3e36044 --- /dev/null +++ b/snuba/snuba_migrations/events_analytics_platform/0067_add_version_column_downsample_512.py @@ -0,0 +1,66 @@ +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 +local_table_name = "eap_items_1_downsample_512_local" + +new_column_name = "version" + +# Mirrors the `version` column added to eap_items_1 in migration 0064, so this +# downsample tier's schema lines up with its target ReplacingMergeTree before +# existing parts are attached onto it. +# +# ADD COLUMN does not rewrite existing parts, so rows written before this +# migration have no `version` on disk and this DEFAULT is what gets evaluated +# when they are read or merged. It therefore has to be deterministic: under +# ReplacingMergeTree(version) a now64() default would hand those legacy rows a +# version of "whenever the merge happened", which is always newer than a real +# write, letting a stale row silently beat a genuine update. `0` is the sentinel +# for "written before versioning existed" and deterministically loses to +# everything else. +new_column: Column[Modifiers] = Column( + new_column_name, + UInt( + 64, + modifiers=Modifiers( + default="0", + codecs=["ZSTD(1)"], + ), + ), +) + +# Only the *_local table gets the column, unlike 0064 which also touched +# eap_items_1_dist. `version` is internal bookkeeping for the +# ReplacingMergeTree cutover: it is absent from the storage YAML so we never +# query it, and the downsample tiers are fed by materialized views that write +# straight to *_local, so the distributed table is never an insert target +# either. A Distributed table that declares a subset of its local table's +# columns reads normally; only a query naming `version` through the dist table +# would fail, and nothing does that. The same reasoning covers *_dist_ro. + + +class Migration(migration.ClickhouseNodeMigration): + blocking = False + + def forwards_ops(self) -> list[operations.SqlOperation]: + return [ + operations.AddColumn( + storage_set=storage_set, + table_name=local_table_name, + column=new_column, + target=OperationTarget.LOCAL, + ), + ] + + def backwards_ops(self) -> list[operations.SqlOperation]: + return [ + operations.DropColumn( + storage_set=storage_set, + table_name=local_table_name, + column_name=new_column_name, + target=OperationTarget.LOCAL, + ), + ] From 3e5b4a15f7d3eac8f378f4571699e8a845645426 Mon Sep 17 00:00:00 2001 From: Oliver Newland Date: Thu, 3 Sep 2026 07:33:11 -0700 Subject: [PATCH 4/4] feat(eap-items): carry version through the downsample MVs Rebuilds the downsample materialized views mv_9 -> mv_10 so they project the version column added in 0064-0067. An MV only inserts the columns its SELECT produces, so until now every row the views wrote landed with version resolved from the destination table's DEFAULT of 0 rather than the value on the source row. Once the tiers move to ReplacingMergeTree(version) that would have them deduping on a column that is 0 for everything, disagreeing with the base table about which copy of an item wins. version is not in TRANSFORMED_COLUMNS, so listing it is enough for downsample_mv_select to copy it straight through. --- .../0068_downsample_mv_carry_version.py | 106 ++++++++++++++++++ 1 file changed, 106 insertions(+) create mode 100644 snuba/snuba_migrations/events_analytics_platform/0068_downsample_mv_carry_version.py diff --git a/snuba/snuba_migrations/events_analytics_platform/0068_downsample_mv_carry_version.py b/snuba/snuba_migrations/events_analytics_platform/0068_downsample_mv_carry_version.py new file mode 100644 index 0000000000..9f863a3518 --- /dev/null +++ b/snuba/snuba_migrations/events_analytics_platform/0068_downsample_mv_carry_version.py @@ -0,0 +1,106 @@ +from collections.abc import Callable + +from snuba.clickhouse.columns import Column, UInt +from snuba.clusters.storage_sets import StorageSetKey +from snuba.migrations import migration +from snuba.migrations.columns import MigrationModifiers as Modifiers +from snuba.migrations.operations import SqlOperation +from snuba.snuba_migrations.events_analytics_platform.templates import ( + downsample_mv_select, + get_eap_items_columns, + swap_downsample_materialized_views, +) +from snuba.utils.schemas import UUID, Array, Bool, Float, Int, Map, String + +storage_set = StorageSetKey.EVENTS_ANALYTICS_PLATFORM + +# mv_9 (migration 0061) -> mv_10: additionally project `version`. +# +# Migrations 0064-0067 added the `version` column to eap_items_1_local and to +# each downsample tier's local table, but the downsample materialized views +# were left projecting the mv_9 column set. Because an MV inserts only the +# columns its SELECT produces, every row the views write currently lands with +# `version` resolved from the destination table's DEFAULT (0) rather than the +# value on the source row. +# +# That is wrong once the tiers move to ReplacingMergeTree(version): a +# downsampled row is a copy of a source row and has to carry the source's +# version, otherwise the tiers dedupe on a column that is 0 for everything and +# the tier disagrees with the base table about which copy of an item wins. +# +# `version` is not in TRANSFORMED_COLUMNS, so adding it to the column list is +# enough for downsample_mv_select() to copy it straight through. +mv_old_version = 9 +mv_new_version = mv_old_version + 1 + +_codec = Modifiers(codecs=["ZSTD(1)"]) + +# The mv_9 column set, reconstructed exactly as 0061 assembled it: the base +# columns, plus the array attribute maps from 0059, plus session_id (0060) and +# ai_conversation_id (0061) inserted after trace_id. +array_attribute_columns: list[Column[Modifiers]] = [ + Column("attributes_array_string", Map(String(), Array(String()), modifiers=_codec)), + Column("attributes_array_int", Map(String(), Array(Int(64)), modifiers=_codec)), + Column("attributes_array_float", Map(String(), Array(Float(64)), modifiers=_codec)), + Column("attributes_array_bool", Map(String(), Array(Bool()), modifiers=_codec)), +] +session_id: Column[Modifiers] = Column("session_id", UUID()) +ai_conversation_id: Column[Modifiers] = Column("ai_conversation_id", String(modifiers=_codec)) + +# Declared without the destination table's DEFAULT, matching how the existing +# view columns are declared (e.g. retention_days omits its DEFAULT 30). The +# view always projects an explicit value, so the default is never consulted. +version_column: Column[Modifiers] = Column("version", UInt(64, modifiers=_codec)) + + +def _mv9_columns() -> list[Column[Modifiers]]: + columns = get_eap_items_columns() + columns.extend(array_attribute_columns) + at = next(i for i, c in enumerate(columns) if c.name == "trace_id") + 1 + return columns[:at] + [session_id, ai_conversation_id] + columns[at:] + + +def _mv10_columns() -> list[Column[Modifiers]]: + # Appended last to mirror the destination tables, where 0065-0067 added + # `version` at the end. Ordering is cosmetic: ClickHouse matches a + # materialized view's output to its target table by column name, which is + # already relied on here since downsample_mv_select() emits the + # transformed columns after the passthrough ones. + return _mv9_columns() + [version_column] + + +def _query_for_weight(columns: list[Column[Modifiers]]) -> Callable[[int], str]: + def inner(sampling_weight: int) -> str: + return downsample_mv_select( + columns, + sampling_weight, + where_predicate=f"cityHash64(item_id) % {sampling_weight}", + ) + + return inner + + +class Migration(migration.ClickhouseNodeMigration): + blocking = False + + def forwards_ops(self) -> list[SqlOperation]: + columns = _mv10_columns() + return list( + swap_downsample_materialized_views( + columns=columns, + create_version=mv_new_version, + drop_version=mv_old_version, + query_for_weight=_query_for_weight(columns), + ) + ) + + def backwards_ops(self) -> list[SqlOperation]: + columns = _mv9_columns() + return list( + swap_downsample_materialized_views( + columns=columns, + create_version=mv_old_version, + drop_version=mv_new_version, + query_for_weight=_query_for_weight(columns), + ) + )