feat(eap): Add v2 co-occurring attributes storage with count, last_seen, and per-type attribute keys

[phacops]

Add a new SummingMergeTree-based storage (eap_item_co_occurring_attrs_v2) for
co-occurring attributes. Compared to the existing ReplacingMergeTree approach
(eap_item_co_occurring_attrs), the v2 table:

• includes a count column that is summed on merge, giving an occurrence count per
  set of co-occurring attributes;

• uses a materialized key_hash (a hash of the sorted, distinct attribute keys) in
  the sort key so rows with the same attribute set are deduplicated/collapsed during
  merges;

• adds a last_seen column (SimpleAggregateFunction(max, DateTime)) tracking the most
  recent timestamp at which a set of attributes was seen (the SummingMergeTree applies
  max on merge);

• represents every attribute type, mirroring the typed maps on eap_items, with one
  key array per type so each attribute can be surfaced with its AttributeKey type:

  column	AttributeKey type	source map on eap_items
  attributes_string	TYPE_STRING	attributes_string_*
  attributes_float	TYPE_FLOAT / TYPE_DOUBLE	attributes_float_*
  attributes_int	TYPE_INT	attributes_int
  attributes_bool	TYPE_BOOLEAN	attributes_bool
  attributes_array_string	TYPE_ARRAY_STRING	attributes_array_string
  attributes_array_int	TYPE_ARRAY_INT	attributes_array_int
  attributes_array_float	TYPE_ARRAY_DOUBLE	attributes_array_float
  attributes_array_bool	TYPE_ARRAY_BOOL	attributes_array_bool

  Both key_hash and the bloom-filter attribute_keys_hash are derived from a single
  arrayConcat(...) of all the key arrays, so dedup and key lookups cover every
  attribute key regardless of type.

Migration

0062_add_count_to_co_occurring_attrs.py creates the local/dist tables, the
bf_attribute_keys_hash bloom-filter index, and a refreshable materialized view
from eap_items_1_local:

CREATE MATERIALIZED VIEW ... REFRESH EVERY 1 MINUTE APPEND TO eap_item_co_occurring_attrs_2_local AS ...

Rather than running incrementally on every insert, the view re-runs on a 1-minute
schedule and appends its result. Because it re-executes the whole query each cycle, the
SELECT aggregates a recent window of items with count() / max(timestamp) and
GROUP BY the key columns (instead of the per-row 1 AS count an incremental view would
emit). The scan window (90s) is a bit larger than the interval so late-arriving items are
not missed; the resulting overlap can slightly inflate count, which is acceptable for
approximate autocomplete counts.

Important

Operational caveats for the refreshable view (verify before rollout):

• It may require allow_experimental_refreshable_materialized_view=1 depending on the
  ClickHouse build.
• On a multi-replica cluster each replica refreshes independently and would APPEND
  duplicate rows — coordinating the refresh (e.g. a Replicated database engine) is
  needed to avoid double counting.
• The scan window vs refresh interval is a tunable trade-off: too small risks missing
  keys (gaps), too large inflates counts (overlap).

Dependencies

Bumps sentry-protos to >=0.35.0, which adds the TYPE_ARRAY_STRING /
TYPE_ARRAY_INT / TYPE_ARRAY_DOUBLE / TYPE_ARRAY_BOOL enum values the split array
columns map to.

Validation (Python 3.13)

• EventsAnalyticsPlatformLoader loads all EAP migrations with no duplicate/gap errors
  (latest is 0062).
• The migration renders valid ClickHouse DDL — the refreshable MV and the target table
  include all eight per-type key arrays, count, and last_seen.
• snuba/validate_configs.py reports all configs valid; ruff check/format pass.

Note

The attribute-names RPC (endpoint_trace_item_attribute_names) still reads the v1
storage, so it does not yet surface count, last_seen, or the int/array key arrays.
Switching it to v2 (behind a rollout flag) and tagging the new array types is a
follow-up.

Agent transcript: https://claudescope.sentry.dev/share/jjGnsb7JWH13GyrGe-wbHapP5rwLIJPOJyGwWJKv-70

[github-actions]

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

-- start migrations

-- forward migration events_analytics_platform : 0062_add_count_to_co_occurring_attrs
Local op: CREATE TABLE IF NOT EXISTS eap_item_co_occurring_attrs_2_local ON CLUSTER 'cluster_one_sh' (organization_id UInt64, project_id UInt64, item_type UInt8, date Date CODEC (DoubleDelta, ZSTD(1)), retention_days UInt16, attribute_keys_hash Array(UInt64) MATERIALIZED arrayMap(k -> cityHash64(k), arrayDistinct(arrayConcat(attributes_string, attributes_float, attributes_int, attributes_bool, attributes_array_string, attributes_array_int, attributes_array_float, attributes_array_bool))), attributes_string Array(String), attributes_float Array(String), attributes_int Array(String), attributes_bool Array(String), attributes_array_string Array(String), attributes_array_int Array(String), attributes_array_float Array(String), attributes_array_bool Array(String), key_hash UInt64 MATERIALIZED cityHash64(arraySort(arrayDistinct(arrayConcat(attributes_string, attributes_float, attributes_int, attributes_bool, attributes_array_string, attributes_array_int, attributes_array_float, attributes_array_bool)))), count UInt64, last_seen SimpleAggregateFunction(max, DateTime)) ENGINE ReplicatedSummingMergeTree('/clickhouse/tables/events_analytics_platform/{shard}/default/eap_item_co_occurring_attrs_2_local', '{replica}') PRIMARY KEY (organization_id, project_id, date, item_type, key_hash) ORDER BY (organization_id, project_id, date, item_type, key_hash, retention_days) PARTITION BY (retention_days, toMonday(date)) TTL date + toIntervalDay(retention_days);
Distributed op: CREATE TABLE IF NOT EXISTS eap_item_co_occurring_attrs_2_dist ON CLUSTER 'cluster_one_sh' (organization_id UInt64, project_id UInt64, item_type UInt8, date Date CODEC (DoubleDelta, ZSTD(1)), retention_days UInt16, attribute_keys_hash Array(UInt64) MATERIALIZED arrayMap(k -> cityHash64(k), arrayDistinct(arrayConcat(attributes_string, attributes_float, attributes_int, attributes_bool, attributes_array_string, attributes_array_int, attributes_array_float, attributes_array_bool))), attributes_string Array(String), attributes_float Array(String), attributes_int Array(String), attributes_bool Array(String), attributes_array_string Array(String), attributes_array_int Array(String), attributes_array_float Array(String), attributes_array_bool Array(String), key_hash UInt64 MATERIALIZED cityHash64(arraySort(arrayDistinct(arrayConcat(attributes_string, attributes_float, attributes_int, attributes_bool, attributes_array_string, attributes_array_int, attributes_array_float, attributes_array_bool)))), count UInt64, last_seen SimpleAggregateFunction(max, DateTime)) ENGINE Distributed(`cluster_one_sh`, default, eap_item_co_occurring_attrs_2_local);
Local op: ALTER TABLE eap_item_co_occurring_attrs_2_local ON CLUSTER 'cluster_one_sh' ADD INDEX IF NOT EXISTS bf_attribute_keys_hash attribute_keys_hash TYPE bloom_filter GRANULARITY 1;
Local op: CREATE MATERIALIZED VIEW IF NOT EXISTS eap_item_co_occurring_attrs_3_mv ON CLUSTER 'cluster_one_sh' TO eap_item_co_occurring_attrs_2_local (organization_id UInt64, project_id UInt64, item_type UInt8, date Date CODEC (DoubleDelta, ZSTD(1)), retention_days UInt16, attribute_keys_hash Array(UInt64) MATERIALIZED arrayMap(k -> cityHash64(k), arrayDistinct(arrayConcat(attributes_string, attributes_float, attributes_int, attributes_bool, attributes_array_string, attributes_array_int, attributes_array_float, attributes_array_bool))), attributes_string Array(String), attributes_float Array(String), attributes_int Array(String), attributes_bool Array(String), attributes_array_string Array(String), attributes_array_int Array(String), attributes_array_float Array(String), attributes_array_bool Array(String), key_hash UInt64 MATERIALIZED cityHash64(arraySort(arrayDistinct(arrayConcat(attributes_string, attributes_float, attributes_int, attributes_bool, attributes_array_string, attributes_array_int, attributes_array_float, attributes_array_bool)))), count UInt64, last_seen SimpleAggregateFunction(max, DateTime)) AS 
SELECT
    organization_id AS organization_id,
    project_id AS project_id,
    item_type as item_type,
    toMonday(timestamp) AS date,
    retention_days as retention_days,
    arrayConcat(mapKeys(attributes_string_0), mapKeys(attributes_string_1), mapKeys(attributes_string_2), mapKeys(attributes_string_3), mapKeys(attributes_string_4), mapKeys(attributes_string_5), mapKeys(attributes_string_6), mapKeys(attributes_string_7), mapKeys(attributes_string_8), mapKeys(attributes_string_9), mapKeys(attributes_string_10), mapKeys(attributes_string_11), mapKeys(attributes_string_12), mapKeys(attributes_string_13), mapKeys(attributes_string_14), mapKeys(attributes_string_15), mapKeys(attributes_string_16), mapKeys(attributes_string_17), mapKeys(attributes_string_18), mapKeys(attributes_string_19), mapKeys(attributes_string_20), mapKeys(attributes_string_21), mapKeys(attributes_string_22), mapKeys(attributes_string_23), mapKeys(attributes_string_24), mapKeys(attributes_string_25), mapKeys(attributes_string_26), mapKeys(attributes_string_27), mapKeys(attributes_string_28), mapKeys(attributes_string_29), mapKeys(attributes_string_30), mapKeys(attributes_string_31), mapKeys(attributes_string_32), mapKeys(attributes_string_33), mapKeys(attributes_string_34), mapKeys(attributes_string_35), mapKeys(attributes_string_36), mapKeys(attributes_string_37), mapKeys(attributes_string_38), mapKeys(attributes_string_39)) AS attributes_string,
    arrayConcat(mapKeys(attributes_float_0), mapKeys(attributes_float_1), mapKeys(attributes_float_2), mapKeys(attributes_float_3), mapKeys(attributes_float_4), mapKeys(attributes_float_5), mapKeys(attributes_float_6), mapKeys(attributes_float_7), mapKeys(attributes_float_8), mapKeys(attributes_float_9), mapKeys(attributes_float_10), mapKeys(attributes_float_11), mapKeys(attributes_float_12), mapKeys(attributes_float_13), mapKeys(attributes_float_14), mapKeys(attributes_float_15), mapKeys(attributes_float_16), mapKeys(attributes_float_17), mapKeys(attributes_float_18), mapKeys(attributes_float_19), mapKeys(attributes_float_20), mapKeys(attributes_float_21), mapKeys(attributes_float_22), mapKeys(attributes_float_23), mapKeys(attributes_float_24), mapKeys(attributes_float_25), mapKeys(attributes_float_26), mapKeys(attributes_float_27), mapKeys(attributes_float_28), mapKeys(attributes_float_29), mapKeys(attributes_float_30), mapKeys(attributes_float_31), mapKeys(attributes_float_32), mapKeys(attributes_float_33), mapKeys(attributes_float_34), mapKeys(attributes_float_35), mapKeys(attributes_float_36), mapKeys(attributes_float_37), mapKeys(attributes_float_38), mapKeys(attributes_float_39)) AS attributes_float,
    mapKeys(attributes_int) AS attributes_int,
    mapKeys(attributes_bool) AS attributes_bool,
    mapKeys(attributes_array_string) AS attributes_array_string,
    mapKeys(attributes_array_int) AS attributes_array_int,
    mapKeys(attributes_array_float) AS attributes_array_float,
    mapKeys(attributes_array_bool) AS attributes_array_bool,
    1 AS count,
    timestamp AS last_seen
FROM eap_items_1_local
;
-- end forward migration events_analytics_platform : 0062_add_count_to_co_occurring_attrs




-- backward migration events_analytics_platform : 0062_add_count_to_co_occurring_attrs
Local op: DROP TABLE IF EXISTS eap_item_co_occurring_attrs_3_mv ON CLUSTER 'cluster_one_sh' SYNC;
Distributed op: DROP TABLE IF EXISTS eap_item_co_occurring_attrs_2_dist ON CLUSTER 'cluster_one_sh' SYNC;
Local op: DROP TABLE IF EXISTS eap_item_co_occurring_attrs_2_local ON CLUSTER 'cluster_one_sh' SYNC;
-- end backward migration events_analytics_platform : 0062_add_count_to_co_occurring_attrs

[linear-code]

EAP-573

[onewland]

Seems fine. You might want to verify the last_seen behavior that you mention here: https://github.com/getsentry/snuba/pull/7801/changes#diff-5f6187dc49efbb83ada5486a8becebd8b791c3adbbe38a7e78f027deac398a17R65-R68

[phacops]

Seems fine. You might want to verify the last_seen behavior that you mention here: https://github.com/getsentry/snuba/pull/7801/changes#diff-5f6187dc49efbb83ada5486a8becebd8b791c3adbbe38a7e78f027deac398a17R65-R68

Yes, this is correct. It will select the timestamp and then the aggregation column will just max it: https://github.com/getsentry/snuba/pull/7801/changes#diff-f1cfe315bbb96650580649e6fa4a1f8b7e672b6f860cdf4a7417a12c574a9b5dR73

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit bf2ae33. Configure here.}

[cursor]

Refresh MV duplicates per replica

High Severity

The refreshable eap_item_co_occurring_attrs_3_mv is created via RunSql without ON CLUSTER, so Snuba runs the DDL on every local node. On multi-replica shards each replica refreshes independently, scans the same eap_items_1_local data, and APPENDs duplicate aggregates into ReplicatedSummingMergeTree, inflating count without DatabaseReplicated coordination.

 

^{Reviewed by Cursor Bugbot for commit bf2ae33. Configure here.}