feat(eap-items): add version column to eap_items_1

[onewland]

Adds a version UInt64 column to eap_items_1_local and eap_items_1_dist.

ALTER TABLE eap_items_1_local ADD COLUMN IF NOT EXISTS version UInt64 DEFAULT 0 CODEC (ZSTD(1));
ALTER TABLE eap_items_1_dist  ADD COLUMN IF NOT EXISTS version UInt64 DEFAULT toUnixTimestamp64Milli(now64(3)) CODEC (ZSTD(1));

Why

We intend to move eap_items to a ReplacingMergeTree that uses version as its version column. Before we can attach the existing parts onto the new table, the current table's schema has to line up with the target schema — so the column needs to land here first, and subsequently be copied to the downsampled tables.

This PR is schema-only. It does not change the engine, so no dedup behavior changes yet.

Why the local and dist defaults differ

This asymmetry is deliberate. Both halves were verified empirically against ClickHouse 25.3.

Local is DEFAULT 0. ADD COLUMN does not rewrite existing parts — the column is simply absent from them (confirmed via system.parts_columns), so reads synthesize it from the default expression at read time. With a now64() default that is non-deterministic; the same row returns a different value on consecutive reads:

read #1:  id=1 -> 1788300553940
read #2:  id=1 -> 1788300555980     # 2s later, same row

Under ReplacingMergeTree(version) this is actively harmful. The legacy row's version gets computed at merge time, which is always later than any real write, so the stale row wins. Inserting a genuine update against a DEFAULT now64() destination and running OPTIMIZE ... FINAL silently discarded the update in favour of the pre-existing row. Against a DEFAULT 0 destination the real update won and the untouched legacy row stayed a stable 0.

So 0 is the sentinel for "written before versioning existed", and it deterministically loses to every real write. This matches the received_at precedent in snuba/manual_jobs/create_eap_received_at_version_test.py, which uses a bare UInt64 and filters WHERE received_at != 0.

Dist is DEFAULT toUnixTimestamp64Milli(now64(3)). A Distributed table evaluates its own DEFAULT expressions on the initiator node and ships the materialized value to the shard, so for inserts routed through _dist the local default never fires. Measured, with deliberately distinguishable values:

local	dist	value stored in local part
DEFAULT 111	DEFAULT 999	999
DEFAULT 111	no default	0
no default	DEFAULT 999	999

Note the middle row: declaring the column on _dist without a default does not fall through to the local default, it ships an explicit 0. So the dist default is what actually assigns a version to live writes, and it is required if we want inserts to carry a real timestamp before the consumer is taught to populate the column.

End-to-end, the pair gives the semantics we want:

id=1 (legacy part) -> 0              stable across re-reads
id=2 (live insert) -> 1788300664281  = 2026-09-01 22:11:04.281

Scope

Limited to _local + _dist. The downsample tiers (eap_items_1_downsample_{8,64,512}) are a follow-up, along with the corresponding materialized view rebuild.

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 one runs, so fresh and existing deployments both end up without the column there — there is no environment drift. If we ever want to read version through the read-only routing path it can be added in its own migration.

The column is likewise not added to eap_items.yaml, so it stays invisible to the query layer for now (same as client_sample_rate).

Testing

• tests/migrations/test_runner.py::test_no_schema_differences
• tests/migrations/test_runner.py::test_run_and_reverse_all

[github-actions]

This PR has a migration; here is the generated SQL for ./snuba/migrations/groups.py ()

-- start migrations

-- forward migration events_analytics_platform : 0064_add_version_column
Local op: ALTER TABLE eap_items_1_local ON CLUSTER 'cluster_one_sh' ADD COLUMN IF NOT EXISTS version UInt64 DEFAULT 0 CODEC (ZSTD(1));
Distributed op: ALTER TABLE eap_items_1_dist ON CLUSTER 'cluster_one_sh' ADD COLUMN IF NOT EXISTS version UInt64 DEFAULT toUnixTimestamp64Milli(now64(3)) CODEC (ZSTD(1));
-- end forward migration events_analytics_platform : 0064_add_version_column




-- backward migration events_analytics_platform : 0064_add_version_column
Distributed op: ALTER TABLE eap_items_1_dist ON CLUSTER 'cluster_one_sh' DROP COLUMN IF EXISTS version;
Local op: ALTER TABLE eap_items_1_local ON CLUSTER 'cluster_one_sh' DROP COLUMN IF EXISTS version;
-- end backward migration events_analytics_platform : 0064_add_version_column