自研选择性复制表 Selective Replication

CREATE TABLE events
(
    event_date  Date,
    user_id     UInt64,
    action      String
)
ENGINE = ReplicatedMergeTree
PARTITION BY toYYYYMM(event_date)
SHARD BY cityHash64(user_id) INTO 16
ORDER BY (user_id, event_date)
SETTINGS replication_factor = 2;

INSERT INTO events VALUES
    ('2024-01-15', 1001, 'login'),
    ('2024-01-15', 1002, 'logout'),
    ('2024-02-01', 2001, 'purchase');

SELECT count() FROM events;
SELECT * FROM events WHERE user_id = 1001;

SELECT partition_id, assigned_replicas, is_under_replicated
FROM system.selective_assignments
WHERE database = currentDatabase() AND table = 'events'
ORDER BY partition_id;
-- 预期结果：partition_id 形如 202401-0 至 202401-15
-- 每行 assigned_replicas 包含 2 个副本，is_under_replicated 为 0

CREATE TABLE [db.]table_name
(
    col1 Type1,
    ...
)
ENGINE = ReplicatedMergeTree
[PARTITION BY <user_partition_expr>]
[ SHARD BY <shard_expr> INTO <N> | SHARD RANDOM INTO <N> ]
ORDER BY <order_expr>
SETTINGS replication_factor = <M>;

SHARD BY <expr> INTO <N>      -- 确定性分桶，桶号为 positiveModulo(<expr>, N)
SHARD RANDOM INTO <N>         -- 随机分桶，桶号为 modulo(randConstant(), N)

SHARD BY INTO N requires N >= 2, got 1
SHARD BY INTO N requires N <= 256, got 512

SHARD BY expects a single expression, got a tuple of 2
SHARD BY expression must have an integer type, but `name` has type String.
Wrap it into a hash function such as `cityHash64`.

Column `_shard` is reserved for the SHARD BY clause and cannot be defined by the user
Cannot ALTER column `_shard` because it is maintained by the SHARD BY clause

PARTITION BY column #1 has non-integral type 'String' — SHARD BY tables with role_aware
require every PARTITION BY column to be an integral, Date/DateTime, or IPv4 type.
Wrap the expression with cityHash64() or similar to obtain an integer.

replication_factor mismatch: table in ZooKeeper has replication_factor=2, but this replica
has replication_factor=3. All replicas must use the same value

First replica was created without selective replication (replication_factor=0), but this
replica has replication_factor=2. All replicas must use the same value

-- 正确：SHARD BY 的 user_id 包含在 ORDER BY 中
CREATE TABLE ok (event_date Date, user_id UInt64, v UInt64)
ENGINE = ReplicatedReplacingMergeTree
PARTITION BY toYYYYMM(event_date)
SHARD BY user_id INTO 8
ORDER BY (user_id, event_date)
SETTINGS replication_factor = 2;

-- 错误：v 未包含在 ORDER BY 中，同 key 数据被拆至不同 partition，去重失效
-- 当前版本不会报错，请勿使用该配置
CREATE TABLE bad (event_date Date, user_id UInt64, v UInt64)
ENGINE = ReplicatedReplacingMergeTree
PARTITION BY toYYYYMM(event_date)
SHARD BY v INTO 8
ORDER BY (user_id, event_date)
SETTINGS replication_factor = 2;

SHARD RANDOM cannot be used with ReplicatedUniqueMergeTree because random sharding breaks
deduplication semantics

UniqueMergeTree with SHARD BY requires all shard key columns to be part of the unique key.
Column 'uid' is used in SHARD BY but is not in the unique key [id]

CREATE TABLE t (key UInt64, v String)
ENGINE = ReplicatedMergeTree
ORDER BY key;

<clickhouse>
    <role>read</role>
</clickhouse>

INSERT INTO events VALUES
    ('2024-01-15', 1001, 'login'),
    ('2024-01-15', 1002, 'logout');

SET max_partitions_per_insert_block = 1000;

Selective replication: no replicas assigned for partition 202401-3 (original: 202401-3).
Check cluster health and rebalance state.

Selective replication INSERT: all assigned replicas [r1, r2] failed for 3 partition(s).
Last error: ...

insert_quorum (3) cannot be greater than replication_factor (2) in selective replication mode.
Only the 2 assigned replicas for each partition will confirm the write.
Lower insert_quorum to at most 2.

Selective replication: partition 202401-3 has only 1 effective assigned replica(s), but
insert_quorum requires 2. Either reduce insert_quorum or wait for more replicas to become
available.

SELECT * FROM events WHERE user_id = 1001;
SELECT count() FROM events WHERE event_date = '2024-01-15';

CREATE TABLE orders (dt Date, uid UInt64, amount Decimal64(2))
ENGINE = ReplicatedMergeTree('/clickhouse/tables/orders', 'r1')
PARTITION BY toYYYYMM(dt)
SHARD BY uid INTO 4
ORDER BY uid
SETTINGS replication_factor = 2;

CREATE TABLE users (dt Date, uid UInt64, name String)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/users', 'r1')
PARTITION BY toYYYYMM(dt)
SHARD BY uid INTO 4
ORDER BY uid
SETTINGS replication_factor = 2;

-- 本地 JOIN
SELECT count() FROM orders o JOIN users u ON o.uid = u.uid;

-- 本地 IN
SELECT sum(amount) FROM orders WHERE uid IN (SELECT uid FROM users);

-- 执行目标查询，客户端指定 --query_id 'check_1'
SELECT count() FROM orders o JOIN users u ON o.uid = u.uid;

SYSTEM FLUSH LOGS query_log;

SELECT
    ProfileEvents['SelectiveReplicationColocatedJoin']                 AS colocated_join,
    ProfileEvents['SelectiveReplicationColocatedJoinDegradedToGlobal'] AS degraded_to_global,
    ProfileEvents['SelectiveReplicationAutoGlobalRewrite']             AS rewritten_to_global,
    ProfileEvents['SelectiveReplicationColocatedIn']                   AS colocated_in,
    ProfileEvents['SelectiveReplicationFullJoinDegradedToInitiator']   AS full_join_on_initiator
FROM system.query_log
WHERE query_id = 'check_1' AND type = 'QueryFinish'
ORDER BY event_time_microseconds DESC
LIMIT 1;

CREATE TABLE orders (dt Date, uid UInt64, amount Decimal64(2))
ENGINE = ReplicatedMergeTree
PARTITION BY toYYYYMM(dt)
SHARD BY uid INTO 16
ORDER BY (uid, dt)
SETTINGS replication_factor = 2;

CREATE TABLE users (dt Date, uid UInt64, name String, city String)
ENGINE = ReplicatedMergeTree
PARTITION BY toYYYYMM(dt)
SHARD BY uid INTO 16
ORDER BY (uid, dt)
SETTINGS replication_factor = 2;

-- 本地 JOIN，无需广播 users 表
SELECT u.city, count() AS order_cnt, sum(o.amount) AS gmv
FROM orders o JOIN users u ON o.uid = u.uid
WHERE o.dt >= '2024-01-01'
GROUP BY u.city;

-- 确认是否真的本地执行（应为 1）
SELECT ProfileEvents['SelectiveReplicationColocatedJoin']
FROM system.query_log WHERE query_id = '<your_query_id>' AND type = 'QueryFinish';

CREATE TABLE detail_log
(
    event_time  DateTime,
    event_date  Date,
    device_id   UInt64,
    payload     String
)
ENGINE = ReplicatedMergeTree
PARTITION BY toYYYYMMDD(event_date)
SHARD BY cityHash64(device_id) INTO 12
ORDER BY (device_id, event_time)
SETTINGS replication_factor = 2;

SELECT partition_id, formatReadableSize(sum(bytes_on_disk)) AS size
FROM system.parts
WHERE database = currentDatabase() AND table = 'detail_log' AND active
GROUP BY partition_id
ORDER BY sum(bytes_on_disk) DESC
LIMIT 20;

-- 核心交易表：3 副本，容忍 2 节点故障
CREATE TABLE trade_core (dt Date, order_id UInt64, amount Decimal64(2))
ENGINE = ReplicatedMergeTree
PARTITION BY toYYYYMM(dt)
SHARD BY order_id INTO 6
ORDER BY (order_id, dt)
SETTINGS replication_factor = 3;

-- 明细日志表：2 副本，控制存储成本
CREATE TABLE trade_log (dt Date, order_id UInt64, detail String)
ENGINE = ReplicatedMergeTree
PARTITION BY toYYYYMMDD(dt)
SHARD BY order_id INTO 24
ORDER BY (order_id, dt)
SETTINGS replication_factor = 2;

ALTER TABLE trade_log MODIFY REPLICATION FACTOR 3;
ALTER TABLE trade_log START SELECTIVE REBALANCE SYNC;

-- 在控制台扩容后自动触发 Rebalance、同步等待完成，无须人工操作，以下仅列出操作命令
ALTER TABLE detail_log START SELECTIVE REBALANCE SYNC;

-- 观察迁移进度
SELECT state, count() FROM system.selective_migrations
WHERE database = currentDatabase() AND table = 'detail_log'
GROUP BY state;

-- 确认各节点负载已均衡
SELECT role, replica, partition_count
FROM system.selective_replica_load
WHERE database = currentDatabase() AND table = 'detail_log'
ORDER BY partition_count DESC;

CREATE TABLE app_log
(
    log_time  DateTime,
    log_date  Date,
    level     LowCardinality(String),
    message   String
)
ENGINE = ReplicatedMergeTree
PARTITION BY toYYYYMMDD(log_date)
SHARD RANDOM INTO 12
ORDER BY (log_time, level)
SETTINGS replication_factor = 2;

SET min_insert_block_size_rows  = 1000000,
    min_insert_block_size_bytes = 134217728;

-- 显式关闭选择性复制表
CREATE TABLE dim_config (id UInt64, name String, config String)
ENGINE = ReplicatedMergeTree
ORDER BY id
SETTINGS replication_factor = 0;