feat(eap): Shard meta tables by trace ID

[colin-sentry]

We'd like to be able to write queries like this one:

SELECT
    quantileTDigest(0.75)(duration_ms)
FROM eap_spans_dist
    PREWHERE name = '/npv' and is_segment = true
WHERE
    organization_id = 22381
  AND project_id = 1506863
  AND attr_str_0['transaction.op'] = 'navigation'
  AND attr_str_14['platform'] != 'roku'
  AND attr_str_14['environment'] = 'prod'
  AND _sort_timestamp > toDateTime(1725576602)
  AND (_sort_timestamp, trace_id) IN (
    SELECT any(timestamp) AS ts, trace_id FROM spans_str_attrs_local WHERE (organization_id, attr_key, attr_value) IN ((22381, 'transaction.op', 'navigation'), (22381, 'environment', 'prod'))
    AND timestamp > toDateTime(1725576602)
    GROUP BY trace_id
    HAVING groupBitOr(attr_key = 'transaction.op' AND attr_value = 'navigation')
       AND groupBitOr(attr_key = 'environment' AND attr_value = 'prod')
    ORDER BY ts
  )
LIMIT 0, 51
SETTINGS max_threads=10, max_memory_usage=0, min_bytes_to_use_direct_io=1,
distributed_product_mode = 'allow'

The eap_spans_dist table is sharded by this same shard key cityHash64(reinterpretAsUInt128(trace_id))

If you don't shard both with the same shard key, you either have to do a two step process (GLOBAL):

• select all trace IDs from all shards
• send that list of all trace IDs to all shards
  this is slow.

Or, you can do

• select trace IDs on each shard

This is incorrect, since you might skip spans which match the conditions, but just happen not to have the meta table rows stored on the same shard.

If you shard both tables by trace_id, you can join on trace_id in a distributed manner, which appears to be ~5x faster.

[github-actions]

This PR has a migration; here is the generated SQL

-- start migrations

-- forward migration events_analytics_platform : 0013_span_attribute_table_shard_keys
Distributed op: DROP TABLE IF EXISTS spans_num_attrs_dist;
Distributed op: CREATE TABLE IF NOT EXISTS spans_num_attrs_dist (organization_id UInt64, trace_id UUID, project_id UInt64, attr_key String, attr_value Float64, timestamp DateTime CODEC (ZSTD(1)), retention_days UInt16, duration_ms SimpleAggregateFunction(max, UInt32), count SimpleAggregateFunction(sum, UInt64)) ENGINE Distributed(`cluster_one_sh`, default, spans_num_attrs_local, cityHash64(reinterpretAsUInt128(trace_id)));
Distributed op: DROP TABLE IF EXISTS spans_str_attrs_dist;
Distributed op: CREATE TABLE IF NOT EXISTS spans_str_attrs_dist (organization_id UInt64, project_id UInt64, trace_id UUID, attr_key String, attr_value String, timestamp DateTime CODEC (ZSTD(1)), retention_days UInt16, count SimpleAggregateFunction(sum, UInt64)) ENGINE Distributed(`cluster_one_sh`, default, spans_str_attrs_local, cityHash64(reinterpretAsUInt128(trace_id)));
-- end forward migration events_analytics_platform : 0013_span_attribute_table_shard_keys




-- backward migration events_analytics_platform : 0013_span_attribute_table_shard_keys
Distributed op: DROP TABLE IF EXISTS spans_num_attrs_dist;
Distributed op: CREATE TABLE IF NOT EXISTS spans_num_attrs_dist (organization_id UInt64, trace_id UUID, project_id UInt64, attr_key String, attr_value Float64, timestamp DateTime CODEC (ZSTD(1)), retention_days UInt16, duration_ms SimpleAggregateFunction(max, UInt32), count SimpleAggregateFunction(sum, UInt64)) ENGINE Distributed(`cluster_one_sh`, default, spans_num_attrs_local);
Distributed op: DROP TABLE IF EXISTS spans_str_attrs_dist;
Distributed op: CREATE TABLE IF NOT EXISTS spans_str_attrs_dist (organization_id UInt64, project_id UInt64, trace_id UUID, attr_key String, attr_value String, timestamp DateTime CODEC (ZSTD(1)), retention_days UInt16, count SimpleAggregateFunction(sum, UInt64)) ENGINE Distributed(`cluster_one_sh`, default, spans_str_attrs_local);
-- end backward migration events_analytics_platform : 0013_span_attribute_table_shard_keys