07. ClickHouse 高性能分析引擎

-- 查看是否使用向量化执行
EXPLAIN PIPELINE SELECT sum(amount) FROM orders WHERE amount > 100;
-- 输出中的 ExpressionTransform 即向量化算子

CREATE TABLE orders (
    order_id UInt64,
    user_id UInt32,
    amount Decimal(18,2),
    status UInt8,
    create_time DateTime,
    city String
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(create_time)        -- 按月分区
ORDER BY (city, create_time, order_id)    -- 主键顺序（稀疏索引）
PRIMARY KEY (city, create_time)           -- 主键（用于索引）
SETTINGS index_granularity = 8192;         -- 索引粒度

-- 按 order_id 去重，保留最新 version
CREATE TABLE user_events (
    user_id UInt64,
    event_type String,
    event_time DateTime,
    properties String,
    version UInt32           -- 版本号，用于冲突解决
) ENGINE = ReplacingMergeTree(version)
PARTITION BY toYYYYMMDD(event_time)
ORDER BY (user_id, event_type, event_time);

-- 查询时通常需要 FINAL 强制合并
SELECT * FROM user_events FINAL WHERE user_id = 123;
-- 或依赖后台 merge（非实时去重）

-- 自动按主键汇总数值列
CREATE TABLE orders_daily (
    dt Date,
    city String,
    category String,
    order_count UInt64,       -- 会被聚合
    total_amount Decimal(18,2), -- 会被聚合
    -- 非数值列保留首次插入值
    first_order_id UInt64
) ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(dt)
ORDER BY (dt, city, category);

-- 批量写入明细（小文件暂存）
INSERT INTO orders_daily VALUES ('2024-01-01', 'Beijing', 'Electronics', 1, 999.00, 1001);
INSERT INTO orders_daily VALUES ('2024-01-01', 'Beijing', 'Electronics', 1, 599.00, 1002);

-- 后台合并后自动聚合为:
-- ('2024-01-01', 'Beijing', 'Electronics', 2, 1598.00, 1001)

-- 查询时务必加 FINAL
SELECT * FROM orders_daily FINAL;

-- 存储聚合函数中间状态，实时聚合查询
CREATE TABLE user_metrics (
    user_id UInt64,
    dt Date,
    -- 使用 AggregateFunction 类型存储中间状态
    total_amount AggregateFunction(sum, Decimal(18,2)),
    order_count AggregateFunction(count, UInt64),
    unique_cities AggregateFunction(uniqExact, String),
    max_amount AggregateFunction(max, Decimal(18,2))
) ENGINE = AggregatingMergeTree()
PARTITION BY toYYYYMM(dt)
ORDER BY (user_id, dt);

-- 写入需使用 -State 后缀
INSERT INTO user_metrics
SELECT
    user_id,
    today() as dt,
    sumState(amount) as total_amount,
    countState() as order_count,
    uniqExactState(city) as unique_cities,
    maxState(amount) as max_amount
FROM orders
GROUP BY user_id;

-- 查询需使用 -Merge 后缀合并状态
SELECT
    user_id,
    sumMerge(total_amount) as total,
    countMerge(order_count) as cnt,
    uniqExactMerge(unique_cities) as cities,
    maxMerge(max_amount) as max_amt
FROM user_metrics
GROUP BY user_id;

CREATE TABLE events (
    user_id UInt64,
    event_time DateTime,
    url String,
    response_time UInt32,
    
    -- 跳数索引：加速特定过滤条件
    INDEX idx_url url TYPE bloom_filter GRANULARITY 4,
    INDEX idx_resp response_time TYPE minmax GRANULARITY 4,
    INDEX idx_set url TYPE set(100) GRANULARITY 4
) ENGINE = MergeTree()
ORDER BY (user_id, event_time);

-- 1. 利用主键过滤（最左前缀）
SELECT * FROM orders WHERE city = 'Beijing' AND create_time > '2024-01-01';
-- 有效 ✓

SELECT * FROM orders WHERE order_id = 12345;
-- 低效 ✗（order_id 不在主键最左列）

-- 2. 分区裁剪
SELECT * FROM orders WHERE create_time >= '2024-01-01' AND create_time < '2024-02-01';
-- 只扫描 202401 分区

-- 3. 预过滤（Projection）
CREATE TABLE orders (
    ...
) ENGINE = MergeTree()
ORDER BY (city, create_time)
PROJECTION projection_by_user
(SELECT user_id, sum(amount) GROUP BY user_id);

-- 4. LIMIT 优化
SELECT * FROM orders ORDER BY create_time DESC LIMIT 10;
-- 若主键包含 create_time，无需全表排序

-- 5. 避免 SELECT *
SELECT user_id, amount FROM orders;  -- 只读取两列
SELECT * FROM orders;                 -- 读取所有列

-- 创建带物化视图的表
CREATE TABLE orders (
    order_id UInt64,
    user_id UInt64,
    amount Decimal(18,2),
    create_time DateTime,
    city String,
    
    PROJECTION projection_user_daily
    (
        SELECT 
            toDate(create_time) as dt,
            user_id,
            sum(amount) as total_amount,
            count() as order_count
        GROUP BY dt, user_id
    ),
    
    PROJECTION projection_city_monthly
    (
        SELECT 
            toYYYYMM(create_time) as month,
            city,
            sum(amount) as total_amount
        GROUP BY month, city
    )
) ENGINE = MergeTree()
ORDER BY (city, create_time, order_id);

-- 目标表
CREATE TABLE orders_summary (
    dt Date,
    city String,
    total_amount AggregateFunction(sum, Decimal(18,2)),
    order_count AggregateFunction(count, UInt64)
) ENGINE = AggregatingMergeTree()
ORDER BY (dt, city);

-- 物化视图（写入 orders 时自动触发）
CREATE MATERIALIZED VIEW orders_summary_mv
TO orders_summary
AS SELECT
    toDate(create_time) as dt,
    city,
    sumState(amount) as total_amount,
    countState() as order_count
FROM orders
GROUP BY dt, city;

-- 查询物化视图
SELECT 
    dt, city,
    sumMerge(total_amount) as gmv,
    countMerge(order_count) as cnt
FROM orders_summary
GROUP BY dt, city;

-- 1. 本地表（每个节点存储实际数据）
CREATE TABLE orders_local ON CLUSTER '{cluster}' (
    order_id UInt64,
    user_id UInt64,
    amount Decimal(18,2),
    create_time DateTime
) ENGINE = ReplicatedMergeTree('/clickhouse/{cluster}/tables/{shard}/orders', '{replica}')
PARTITION BY toYYYYMM(create_time)
ORDER BY (user_id, create_time);

-- 2. 分布式表（路由层）
CREATE TABLE orders_dist ON CLUSTER '{cluster}' AS orders_local
ENGINE = Distributed('{cluster}', 'default', 'orders_local', rand());

-- 3. 写入分布式表
INSERT INTO orders_dist VALUES (...);
-- 自动按 sharding_key (rand()) 分发到各 shard

-- 4. 查询分布式表
SELECT user_id, sum(amount) FROM orders_dist GROUP BY user_id;
-- 自动下发到各 shard → 本地聚合 → 汇总结果

-- 方式一：直接写本地表（需客户端自己分片）
-- 适合离线批量导入
INSERT INTO orders_local VALUES (...);

-- 方式二：写分布式表 + 本地表双写
-- BI 查分布式表，实时应用写本地表

-- 方式三：异步 distributed_directory_monitor
-- clickhouse 自动将分布式目录数据分发到 shard

-- 查询当前正在执行的查询
SELECT * FROM system.processes WHERE is_cancelled = 0;

-- 查询慢查询日志
SELECT * FROM system.query_log WHERE event_time > now() - 3600 ORDER BY query_duration_ms DESC LIMIT 20;

-- 查看 Part 合并情况
SELECT database, table, partition, name, bytes_on_disk, modification_time 
FROM system.parts WHERE active = 1;

<!-- config.xml 关键配置 -->
<max_concurrent_queries>100</max_concurrent_queries>
<max_memory_usage>50000000000</max_memory_usage>  <!-- 50GB -->
<max_execution_time>300</max_execution_time>       <!-- 5分钟超时 -->
<max_partitions_per_insert_block>100</max_partitions_per_insert_block>

<!-- 压缩配置 -->
<merge_tree>
    <min_compress_block_size>65536</min_compress_block_size>
    <max_compress_block_size>1048576</max_compress_block_size>
</merge_tree>

-- 查看 Part 合并状态
SELECT 
    database, table, partition, name,
    level, bytes_on_disk, rows,
    modification_time,
    active                                  -- 1=活跃, 0=待清理
FROM system.parts 
WHERE table = 'orders' AND active = 1
ORDER BY partition, name;

-- 手动触发合并（慎用，生产环境避免高峰期执行）
OPTIMIZE TABLE orders PARTITION '202401' FINAL;

-- 查看合并任务队列
SELECT * FROM system.merges;

<merge_tree>
    <!-- 触发合并的最小 Part 数 -->
    <parts_to_delay_insert>300</parts_to_delay_insert>
    <parts_to_throw_insert>600</parts_to_throw_insert>
    <!-- 最大 Part 尺寸 -->
    <max_bytes_to_merge_at_max_space_in_pool>107374182400</max_bytes_to_merge_at_max_space_in_pool>
    <!-- 旧数据合并阈值（降低冷数据合并频率） -->
    <min_age_to_force_merge_seconds>86400</min_age_to_force_merge_seconds>
    <min_age_to_force_merge_on_partition_only>false</min_age_to_force_merge_on_partition_only>
</merge_tree>

-- 按时间自动删除旧数据
CREATE TABLE logs (
    event_time DateTime,
    message String,
    level String
) ENGINE = MergeTree()
ORDER BY event_time
TTL event_time + INTERVAL 90 DAY;           -- 90 天后自动删除

-- 按时间自动转移到冷存储（S3 / 另一磁盘卷）
CREATE TABLE logs_tiered (
    event_time DateTime,
    message String
) ENGINE = MergeTree()
ORDER BY event_time
TTL event_time + INTERVAL 7 DAY TO VOLUME 's3_cold',
    event_time + INTERVAL 30 DAY DELETE;    -- 7 天后转 S3，30 天后删除

-- 查看 TTL 任务状态
SELECT 
    table,
    name as partition,
    delete_ttl_info_min,
    delete_ttl_info_max,
    move_ttl_info.expression
FROM system.parts 
WHERE table = 'logs' AND active = 1;

-- 建表时指定索引粒度
CREATE TABLE high_cardinality_events (
    event_id UUID,
    user_id UInt64,
    event_time DateTime
) ENGINE = MergeTree()
ORDER BY event_id
SETTINGS index_granularity = 512;           -- 更细粒度索引

-- 运行时查看表设置
SELECT * FROM system.tables WHERE name = 'orders' \G

-- 1. 创建目标表（存储聚合状态）
CREATE TABLE events_agg (
    dt Date,
    domain String,
    pv AggregateFunction(count, UInt64),
    uv AggregateFunction(uniqExact, UInt64),
    total_latency AggregateFunction(sum, UInt64)
) ENGINE = AggregatingMergeTree()
ORDER BY (dt, domain);

-- 2. 创建物化视图（自动触发）
CREATE MATERIALIZED VIEW events_agg_mv
TO events_agg
AS SELECT
    toDate(event_time) as dt,
    domain,
    countState() as pv,
    uniqExactState(user_id) as uv,
    sumState(latency_ms) as total_latency
FROM events
GROUP BY dt, domain;

-- 3. 查询物化视图（极速）
SELECT 
    dt, domain,
    countMerge(pv) as pv,
    uniqExactMerge(uv) as uv,
    sumMerge(total_latency) / countMerge(pv) as avg_latency
FROM events_agg
WHERE dt = today()
GROUP BY dt, domain;

CREATE TABLE events_with_proj (
    event_time DateTime,
    user_id UInt64,
    domain String,
    page String,
    latency_ms UInt32,
    
    -- Projection 1: 按 domain + date 聚合 PV/UV
    PROJECTION proj_domain_daily
    (
        SELECT 
            toDate(event_time) as dt,
            domain,
            count() as pv,
            uniqExact(user_id) as uv,
            sum(latency_ms) as total_latency
        GROUP BY dt, domain
    ),
    
    -- Projection 2: 按 page 聚合访问情况
    PROJECTION proj_page_stats
    (
        SELECT 
            domain,
            page,
            count() as pv,
            avg(latency_ms) as avg_latency
        GROUP BY domain, page
    )
    
) ENGINE = MergeTree()
ORDER BY (domain, event_time);

-- 自动路由：以下查询会自动使用 proj_domain_daily
SELECT 
    toDate(event_time) as dt,
    domain,
    count() as pv,
    uniqExact(user_id) as uv
FROM events_with_proj
WHERE dt = today()
GROUP BY dt, domain;

-- 验证是否命中 Projection（查看执行计划）
EXPLAIN ACTIONS 
SELECT domain, count() FROM events_with_proj GROUP BY domain;
-- 输出中含 "ReadFromProjection" 即命中预聚合数据

-- 分布式表定义：基于 sharding_key 分发
CREATE TABLE orders_dist ON CLUSTER '{cluster}' AS orders_local
ENGINE = Distributed('{cluster}', 'default', 'orders_local', 
    cityHash64(user_id)        -- 分片键：保证同一 user_id 发到同一分片
);

-- 常见分片策略对比
-- 1. rand()           → 均匀分布，但不利于按维度聚合
-- 2. cityHash64(id)   → 按业务键哈希，利于本地 JOIN/聚合
-- 3. toYYYYMM(dt)     → 按时间分片，适合时序场景
-- 4. jumpConsistentHash(user_id, 8) → 一致性哈希，扩容友好

-- 查看分布式查询执行状态
SELECT 
    initial_query_id,
    host_name,
    type,
    event_time,
    query_duration_ms,
    read_rows,
    read_bytes
FROM clusterAllReplicas('{cluster}', system.query_log)
WHERE initial_query_id = 'xxx'
ORDER BY event_time;

-- 本地副本表（每个节点独立运行）
CREATE TABLE orders_local ON CLUSTER '{cluster}' (
    order_id UInt64,
    user_id UInt64,
    amount Decimal(18,2),
    create_time DateTime
) ENGINE = ReplicatedMergeTree(
    '/clickhouse/{cluster}/tables/{shard}/orders',   -- ZooKeeper 路径
    '{replica}'                                        -- 副本标识
)
PARTITION BY toYYYYMM(create_time)
ORDER BY (user_id, create_time);

-- 查看副本同步延迟
SELECT 
    database, table,
    is_leader,
    can_become_leader,
    is_readonly,
    future_parts,
    parts_to_check,
    zookeeper_path,
    replica_name,
    queue_size,                         -- 待处理队列大小
    inserts_in_queue,                   -- 待插入 Part 数
    merges_in_queue,                    -- 待合并任务数
    absolute_delay,                     -- 绝对延迟（秒）
    total_replicas,
    active_replicas
FROM system.replicas 
WHERE table = 'orders_local';

-- 查看 ZooKeeper 操作统计
SELECT * FROM system.zookeeper WHERE path = '/clickhouse';

-- 原始查询
SELECT user_id, amount, city 
FROM orders 
WHERE city = 'Beijing' AND amount > 1000;

-- 优化器自动重写为：
SELECT user_id, amount, city
FROM orders
PREWHERE city = 'Beijing'       -- 先只用 city 列过滤，减少后续数据量
WHERE amount > 1000;

-- 手动控制 PREWHERE（当优化器判断不准时）
SELECT user_id, amount
FROM orders
PREWHERE create_time > '2024-01-01'
WHERE status = 1;

-- 确认查询是否使用向量化执行（查看 PIPELINE）
EXPLAIN PIPELINE 
SELECT sum(amount), avg(latency_ms) 
FROM events 
WHERE domain = 'api.example.com';

-- 期望输出包含以下算子（表示向量化路径）：
-- ExpressionTransform
-- FilterTransform
-- AggregatingTransform

-- 使用 Int32 而非 String 编码状态，触发 SIMD
SELECT count() FROM events WHERE status_code > 400;
-- 若 status_code 为 UInt16，ClickHouse 会用 SSE/AVX2 批量比较

-- 开启 Profile 日志（users.xml）
<trace_log>
    <database>system</database>
    <table>trace_log</table>
</trace_log>

-- 查看最近慢查询 Top 20
SELECT 
    event_time,
    query_id,
    query,
    query_duration_ms,
    read_rows,
    read_bytes,
    result_rows,
    memory_usage,
    Settings['max_threads'] as threads
FROM system.query_log
WHERE event_time > now() - INTERVAL 1 HOUR
  AND type = 'QueryFinish'
ORDER BY query_duration_ms DESC
LIMIT 20;

-- 分析单次查询详细 Stage
SELECT 
    event_name,
    duration_ms,
    read_rows,
    read_bytes
FROM system.query_execution_log        -- ClickHouse 24.3+
WHERE query_id = 'xxx'
ORDER BY event_time;

-- 使用 EXPLAIN 分析索引命中情况
EXPLAIN indexes = 1 
SELECT * FROM orders WHERE order_id = 12345;

-- 若 order_id 不在 ORDER BY 最左前缀，输出将显示
-- "Condition(order_id = 12345) is not analyzed" 或无索引信息

-- 创建 Kafka 消费表（只做消费端接入）
CREATE TABLE events_kafka (
    event_time DateTime,
    user_id UInt64,
    domain String,
    page String,
    latency_ms UInt32
) ENGINE = Kafka()
SETTINGS 
    kafka_broker_list = 'kafka-1:9092,kafka-2:9092',
    kafka_topic_list = 'clickhouse-events',
    kafka_group_name = 'ch-events-consumer',
    kafka_format = 'JSONEachRow',
    kafka_num_consumers = 4,            -- 并发消费者数
    kafka_max_block_size = 1048576,     -- 每批次最大行数
    kafka_skip_broken_messages = 10;    -- 跳过错误消息阈值

-- 创建目标 MergeTree 表
CREATE TABLE events (
    event_time DateTime,
    user_id UInt64,
    domain String,
    page String,
    latency_ms UInt32
) ENGINE = MergeTree()
ORDER BY (domain, event_time);

-- 创建物化视图桥接 Kafka 表 → 目标表
CREATE MATERIALIZED VIEW events_consumer TO events
AS SELECT * FROM events_kafka;

-- MySQL 引擎：直接查询远端 MySQL 表
CREATE TABLE mysql_users (
    id UInt64,
    name String,
    email String
) ENGINE = MySQL('mysql-host:3306', 'mydb', 'users', 'reader', 'password');

-- 直接查询（数据不落地 ClickHouse）
SELECT * FROM mysql_users WHERE id > 1000;

-- PostgreSQL 引擎
CREATE TABLE pg_orders (
    order_id UInt64,
    amount Decimal(18,2)
) ENGINE = PostgreSQL('pg-host:5432', 'shop', 'orders', 'reader', 'password');

-- 联邦 JOIN：ClickHouse 拉取远端小表做本地 JOIN
SELECT 
    o.order_id,
    u.name,
    o.amount
FROM pg_orders o
JOIN mysql_users u ON o.user_id = u.id
WHERE o.amount > 100;

-- 直接查询 S3 上的 Parquet 文件（无需预先加载）
SELECT 
    user_id,
    count() as cnt,
    sum(amount) as total
FROM s3(
    'https://bucket.s3.amazonaws.com/data/orders/*.parquet',
    'AKIA...',                          -- Access Key
    'secret...',                        -- Secret Key
    'Parquet'                           -- 文件格式
)
WHERE create_time > '2024-01-01'
GROUP BY user_id;

-- 插入数据到 S3（导出冷存）
INSERT INTO FUNCTION s3(
    'https://bucket.s3.amazonaws.com/archive/orders_202401.parquet',
    'AKIA...', 
    'secret...',
    'Parquet'
)
SELECT * FROM orders WHERE create_time < '2024-02-01';

-- S3 作为外部卷挂载到表 TTL
CREATE TABLE logs (
    event_time DateTime,
    message String
) ENGINE = MergeTree()
ORDER BY event_time
TTL event_time + INTERVAL 30 DAY TO DISK 's3_disk';

<!-- remote_servers.xml 配置示例 -->
<remote_servers>
    <prod_cluster>
        <shard>
            <internal_replication>true</internal_replication>
            <replica>
                <host>ch-shard1-replica1</host>
                <port>9000</port>
            </replica>
            <replica>
                <host>ch-shard1-replica2</host>
                <port>9000</port>
            </replica>
        </shard>
        <shard>
            <internal_replication>true</internal_replication>
            <replica>
                <host>ch-shard2-replica1</host>
                <port>9000</port>
            </replica>
            <replica>
                <host>ch-shard2-replica2</host>
                <port>9000</port>
            </replica>
        </shard>
    </prod_cluster>
</remote_servers>

# ch-proxy.yml
server:
  http:
    listen_addr: ":9090"
  
clusters:
  - name: "prod_cluster"
    scheme: "http"
    nodes:
      - "192.168.1.1:8123"
      - "192.168.1.2:8123"
      - "192.168.1.3:8123"
      - "192.168.1.4:8123"
    users:
      - name: "reader"
        password: "xxx"
        max_concurrent_queries: 50
        max_execution_time: 120s
    # 负载均衡策略
    kill_query_user:
      name: "admin"
      password: "xxx"
    heartbeat:
      interval: 5s
      timeout: 3s
      request: "/ping"

apiVersion: clickhouse.altinity.com/v1
kind: ClickHouseInstallation
metadata:
  name: prod-cluster
spec:
  configuration:
    clusters:
      - name: prod
        layout:
          shardsCount: 2
          replicasCount: 2
    zookeeper:
      nodes:
        - host: zookeeper-0.zk
          port: 2181
  templates:
    podTemplates:
      - name: clickhouse-pod
        spec:
          containers:
            - name: clickhouse
              image: clickhouse/clickhouse-server:24.3
              resources:
                requests:
                  memory: "8Gi"
                  cpu: "4"
                limits:
                  memory: "32Gi"
                  cpu: "16"
    volumeClaimTemplates:
      - name: data
        spec:
          accessModes:
            - ReadWriteOnce
          resources:
            requests:
              storage: 500Gi

-- 查看各数据库/表磁盘占用
SELECT 
    database,
    table,
    formatReadableSize(sum(bytes_on_disk)) as disk_size,
    sum(rows) as total_rows,
    count() as parts_count,
    max(modification_time) as last_modified
FROM system.parts
WHERE active = 1
GROUP BY database, table
ORDER BY sum(bytes_on_disk) DESC
LIMIT 20;

-- 查看各磁盘卷使用情况
SELECT 
    name,
    path,
    formatReadableSize(free_space) as free,
    formatReadableSize(total_space) as total,
    round((1 - free_space / total_space) * 100, 2) as usage_pct
FROM system.disks;

-- 监控 Merge 导致的临时磁盘膨胀
SELECT 
    database, table,
    round(100 * sum(bytes_on_disk * (1 - active)) / sum(bytes_on_disk), 2) as inactive_pct
FROM system.parts
GROUP BY database, table
HAVING inactive_pct > 20;               -- 活跃数据占比过低，说明旧 Part 堆积

-- ClickHouse 24.3+ 原生备份（推荐）
BACKUP TABLE orders TO File('/backups/orders_20240101.zip');

-- 备份整个数据库
BACKUP DATABASE default TO S3('https://bucket.s3.amazonaws.com/ch-backup/default_20240101', 'AK', 'SK');

-- 恢复表
RESTORE TABLE orders FROM File('/backups/orders_20240101.zip');

-- 使用 clickhouse-backup 工具（生产推荐）
# 1. 创建备份
clickhouse-backup create orders_backup_20240101

# 2. 上传到远程存储
clickhouse-backup upload orders_backup_20240101 --storage=s3

# 3. 恢复（先停写入）
clickhouse-backup restore orders_backup_20240101 --table=orders

# 滚动升级流程（副本保证可用性）
# 1. 标记副本为只读（避免写入到待升级节点）
echo "SYSTEM STOP REPLICATED SENDS" | clickhouse-client

# 2. 检查副本同步延迟（确保 <= 10s）
SELECT absolute_delay FROM system.replicas WHERE replica_name = 'replica_01';

# 3. 停止 ClickHouse 服务
systemctl stop clickhouse-server

# 4. 替换二进制文件（保留 config）
apt install clickhouse-server=24.8.1 clickhouse-client=24.8.1

# 5. 启动并验证
systemctl start clickhouse-server
clickhouse-client --query "SELECT version()"

# 6. 恢复副本同步
echo "SYSTEM START REPLICATED SENDS" | clickhouse-client

# 7. 逐节点重复（确保每轮至少一个副本可用）

-- 故障 1：副本同步中断（queue_size 持续增长）
-- 解决：检查 ZooKeeper 连接、磁盘空间、手动恢复
SYSTEM RESTART REPLICA orders_local;

-- 故障 2：Too many parts（写入过于频繁，小文件堆积）
-- 解决：增大 batch_size、启用 Buffer 表、调大 parts_to_delay_insert
SELECT 
    table,
    count() as part_count
FROM system.parts
WHERE active = 1
GROUP BY table
HAVING part_count > 300;

-- 故障 3：查询卡死（max_execution_time 不生效）
-- 根因：某些算子（如复杂 JOIN）不响应取消信号
-- 解决：Kill 查询、限制 Join 表大小、改用 GLOBAL JOIN
KILL QUERY WHERE query_id = 'xxx';

-- 故障 4：内存溢出（OOM Killer）
-- 解决：调低 max_memory_usage、开启内存溢出转磁盘（allow_experimental_memory_bound_aggregator）
SET max_memory_usage = 10_000_000_000;      -- 10GB 上限
SET max_bytes_before_external_group_by = 5_000_000_000;
SET max_bytes_before_external_sort = 5_000_000_000;