自研存算分离表引擎 KeeperMergeTree

CREATE TABLE analytics.events
(
    event_id   UInt64,
    user_id    UInt64,
    event_time DateTime,
    payload    String
)
ENGINE = KeeperMergeTree
SHARD BY user_id INTO 8 
ORDER BY (event_time, user_id);

-- 以下两种写法等价，最终都会创建为 KeeperMergeTree
CREATE TABLE analytics.events
(
    event_id   UInt64,
    user_id    UInt64,
    event_time DateTime,
    payload    String
)
ENGINE = MergeTree
SHARD BY user_id INTO 8
ORDER BY (event_time, user_id);

CREATE TABLE analytics.events
(
    event_id   UInt64,
    user_id    UInt64,
    event_time DateTime,
    payload    String
)
ENGINE = KeeperMergeTree
SHARD BY user_id INTO 8
ORDER BY (event_time, user_id);

-- DDL 直接执行，自动全集群生效
CREATE DATABASE analytics;
ALTER TABLE analytics.events ADD COLUMN source String DEFAULT 'unknown';
DROP TABLE analytics.events;

-- RBAC 操作同理
CREATE USER analyst IDENTIFIED WITH sha256_password BY 'password';
CREATE ROLE analyst_role;
GRANT SELECT ON analytics.* TO analyst_role;
GRANT analyst_role TO analyst;

INSERT INTO analytics.events VALUES
    (1, 100, '2025-01-01 00:00:00', 'click'),
    (2, 200, '2025-01-01 00:00:01', 'view'),
    (3, 100, '2025-01-01 00:00:02', 'purchase');

-- 完整结果：跨所有 Shard 并行
SELECT count(), sum(user_id)
FROM distributed(currentDatabase(), events);

-- 按 Shard Key 过滤：自动裁剪，只命中目标 Shard
SELECT event_id, payload
FROM distributed(currentDatabase(), events)
WHERE user_id = 10086;

-- 按 Shard Key GROUP BY：每个分组完整落在单个 Shard，无需跨节点聚合
SELECT user_id, count()
FROM distributed(currentDatabase(), events)
GROUP BY user_id;

-- IN 查询：只命中相关 Shard
SELECT count()
FROM distributed(currentDatabase(), events)
WHERE user_id IN (1, 2, 3, 100);

-- 右表显式使用 local()：按左表 shard 分发，避免全表扫描
SELECT e.event_id, u.name
FROM distributed(currentDatabase(), events) AS e
JOIN local(currentDatabase(), users) AS u ON e.user_id = u.user_id;

-- 右表未使用 local()：每个 shard 节点都会全表扫描右表
SELECT e.event_id, u.name
FROM distributed(currentDatabase(), events) AS e
JOIN users AS u ON e.user_id = u.user_id;

-- 当前节点 Shard 的数据量
SELECT count()
FROM local(currentDatabase(), events);

-- 本地 Shard 内按 Shard Key 查询（命中则返回，未命中返回空）
SELECT event_id, payload
FROM local(currentDatabase(), events)
WHERE user_id = 10086;

-- 指定 shard_num，查询第 4 号逻辑分片的数据
SELECT count()
FROM local(currentDatabase(), events, 4);

-- 指定数组，查询第 1、3、5 号逻辑分片的数据
SELECT count()
FROM local(currentDatabase(), events, [1, 3, 5]);

-- 直接查本地表，任意节点返回全量数据
SELECT count() FROM events;

-- 配合 parallel replicas 在多节点并行加速全表扫描
SET enable_parallel_replicas = 1;
SELECT count(), sum(user_id) FROM events;

distributed(database, table)
local(database, table [, shard_num])

逻辑 Shard 编号 = hash(shard_key_value) % shard_num

<merge_tree>
    <enable_shard_affinity>1</enable_shard_affinity>
</merge_tree>