-- 查看默认压缩算法
SELECT name, value
FROM system.settings
WHERE name LIKE '%compression%';

-- 创建表时指定默认压缩算法
CREATE TABLE lz4_table
(
    event_date Date,
    user_id UInt64,
    data String
)
ENGINE = MergeTree()
ORDER BY event_date
SETTINGS min_compress_block_size = 65536,
         max_compress_block_size = 1048576;

-- 对整个表设置压缩算法
CREATE TABLE zstd_table
(
    event_date Date,
    user_id UInt64,
    data String
)
ENGINE = MergeTree()
ORDER BY event_date;

-- 修改列的压缩算法
ALTER TABLE zstd_table MODIFY COLUMN data CODEC(ZSTD(3));

CREATE TABLE timeseries_data
(
    timestamp DateTime64(3),
    host LowCardinality(String),
    cpu_usage Float64,
    memory_usage Float64,
    disk_read UInt64,
    disk_write UInt64,
    network_in UInt64,
    network_out UInt64
)
ENGINE = MergeTree()
ORDER BY (host, timestamp)
-- 列级编码
CODEC(
    timestamp    Delta(8) ZSTD(1),      -- 时间戳差值 + ZSTD
    cpu_usage    Gorilla ZSTD(1),       -- 浮点 XOR + ZSTD
    memory_usage Gorilla ZSTD(1),       -- 浮点 XOR + ZSTD
    disk_read    Delta(8) ZSTD(1),      -- 整数差值 + ZSTD
    disk_write   Delta(8) ZSTD(1),
    network_in   Delta(8) ZSTD(1),
    network_out  Delta(8) ZSTD(1)
);

-- 插入模拟数据
INSERT INTO timeseries_data
SELECT
    toDateTime('2026-05-01 00:00:00') + toIntervalSecond(number * 10) AS timestamp,
    'server-01' AS host,
    50 + sin(number / 100) * 30 AS cpu_usage,
    60 + cos(number / 50) * 20 AS memory_usage,
    toUInt64(rand() % 1000000) AS disk_read,
    toUInt64(rand() % 500000) AS disk_write,
    toUInt64(rand() % 10000000) AS network_in,
    toUInt64(rand() % 5000000) AS network_out
FROM numbers(8640);  -- 一天的数据，每 10 秒一条

-- 查看压缩效果
SELECT
    name,
    formatReadableSize(data_uncompressed_bytes) AS uncompressed,
    formatReadableSize(data_compressed_bytes) AS compressed,
    round(data_uncompressed_bytes / data_compressed_bytes, 2) AS ratio,
    formatReadableSize(marks_bytes) AS marks_size
FROM system.columns
WHERE table = 'timeseries_data' AND database = currentDatabase();

-- T64 适合值范围较小的整数列
CREATE TABLE temperature_sensors
(
    sensor_id UInt32,
    reading_time DateTime,
    temperature Int16,          -- 温度变化范围小（-50 ~ 50 度）
    humidity UInt8              -- 湿度 0~100
)
ENGINE = MergeTree()
ORDER BY (sensor_id, reading_time)
CODEC(
    temperature T64 ZSTD(1),
    humidity T64 LZ4
);

-- 查看编码效果
SELECT
    column,
    formatReadableSize(data_uncompressed_bytes) AS uncompressed,
    formatReadableSize(data_compressed_bytes) AS compressed,
    round(data_uncompressed_bytes / data_compressed_bytes, 2) AS ratio
FROM system.columns
WHERE table = 'temperature_sensors' AND database = currentDatabase();

-- Gorilla（Facebook 论文）: 相邻浮点值 XOR 后用变长编码
-- 适合变化平缓的浮点数据

CREATE TABLE stock_prices
(
    trade_date Date,
    stock_code LowCardinality(String),
    open_price Float64,
    close_price Float64,
    high_price Float64,
    low_price Float64,
    volume UInt64
)
ENGINE = MergeTree()
ORDER BY (stock_code, trade_date)
CODEC(
    open_price   Gorilla ZSTD(1),
    close_price  Gorilla ZSTD(1),
    high_price   Gorilla ZSTD(1),
    low_price    Gorilla ZSTD(1),
    volume       Delta(8) ZSTD(1)
);

-- 创建测试表对比不同压缩算法
CREATE TABLE compress_test_lz4
(
    event_date Date,
    value UInt64,
    description String
) ENGINE = MergeTree() ORDER BY event_date
CODEC(LZ4);

CREATE TABLE compress_test_zstd
(
    event_date Date,
    value UInt64,
    description String
) ENGINE = MergeTree() ORDER BY event_date
CODEC(ZSTD(1));

CREATE TABLE compress_test_zstd_high
(
    event_date Date,
    value UInt64,
    description String
) ENGINE = MergeTree() ORDER BY event_date
CODEC(ZSTD(9));

-- 插入相同数据
INSERT INTO compress_test_lz4 SELECT
    toDate('2026-01-01') + toIntervalDay(number % 365),
    rand64(),
    concat('Event description for record number ', toString(number), ' with some random text content')
FROM numbers(1000000);

INSERT INTO compress_test_zstd SELECT * FROM compress_test_lz4;
INSERT INTO compress_test_zstd_high SELECT * FROM compress_test_lz4;

-- 对比压缩效果
SELECT
    table,
    formatReadableSize(sum(data_uncompressed_bytes)) AS uncompressed,
    formatReadableSize(sum(data_compressed_bytes)) AS compressed,
    round(sum(data_uncompressed_bytes) / sum(data_compressed_bytes), 2) AS ratio,
    round(sum(data_compressed_bytes) / sum(data_uncompressed_bytes) * 100, 2) AS compression_pct
FROM system.parts
WHERE database = currentDatabase()
  AND table LIKE 'compress_test_%'
  AND active = 1
GROUP BY table
ORDER BY ratio DESC;

-- 查询压缩算法对读取性能的影响
-- 使用系统表查看实际读取统计
SELECT
    query,
    read_rows,
    formatReadableSize(read_bytes) AS read_size,
    query_duration_ms,
    formatReadableSize(memory_usage) AS memory
FROM system.query_log
WHERE type = 'QueryFinish'
  AND query LIKE '%compress_test_%'
  AND event_time > now() - INTERVAL 1 HOUR
ORDER BY query_duration_ms
LIMIT 10;

-- 查看压缩统计
SELECT
    event_time,
    type,
    database,
    table,
    compressed_size,
    uncompressed_size,
    compression_codec
FROM system.part_log
WHERE table = 'your_table'
  AND event_time > now() - INTERVAL 1 DAY
ORDER BY event_time DESC
LIMIT 20;

-- 查看压缩块大小设置
SELECT name, value
FROM system.settings
WHERE name IN (
    'min_compress_block_size',
    'max_compress_block_size',
    'max_block_size'
);

-- 建表时配置压缩参数
CREATE TABLE optimized_compress
(
    event_date Date,
    user_id UInt64,
    data String
)
ENGINE = MergeTree()
ORDER BY event_date
SETTINGS
    min_compress_block_size = 65536,       -- 最小压缩块大小（字节）
    max_compress_block_size = 1048576;     -- 最大压缩块大小（字节）

-- 全局压缩配置（config.xml）
-- <compression>
--     <case>
--         <min_part_size>10000000000</min_part_size>
--         <method>zstd</method>
--     </case>
-- </compression>

-- 测试不同压缩算法的写入速度
-- 创建表
CREATE TABLE write_test_lz4 (id UInt64, data String) ENGINE = MergeTree() ORDER BY id CODEC(LZ4);
CREATE TABLE write_test_zstd (id UInt64, data String) ENGINE = MergeTree() ORDER BY id CODEC(ZSTD(1));

-- 批量写入 LZ4 压缩表
INSERT INTO write_test_lz4 SELECT
    number,
    concat('row_', toString(number), '_data_padding_content')
FROM numbers(1000000);

-- 批量写入 ZSTD 压缩表
INSERT INTO write_test_zstd SELECT
    number,
    concat('row_', toString(number), '_data_padding_content')
FROM numbers(1000000);

-- 查看写入统计
SELECT
    table,
    formatReadableSize(sum(data_compressed_bytes)) AS compressed_size,
    formatReadableSize(sum(data_uncompressed_bytes)) AS uncompressed_size,
    round(sum(data_uncompressed_bytes) / sum(data_compressed_bytes), 2) AS ratio
FROM system.parts
WHERE database = currentDatabase()
  AND table LIKE 'write_test_%'
  AND active = 1
GROUP BY table;

-- 监控系统：每天 100 亿条指标数据
CREATE TABLE metrics.compressed_metrics
(
    timestamp DateTime64(3) CODEC(Delta(8), ZSTD(1)),
    host LowCardinality(String) CODEC(LZ4),
    metric_name LowCardinality(String) CODEC(LZ4),
    value Float64 CODEC(Gorilla, ZSTD(1)),
    tags Map(String, String) CODEC(ZSTD(3))
)
ENGINE = MergeTree()
PARTITION BY toYYYYMMDD(timestamp)
ORDER BY (host, metric_name, timestamp);

-- 预期压缩率: 20:1 ~ 50:1
-- 100 亿条 × ~100 字节 = ~1TB 原始数据
-- 压缩后: ~20GB ~ 50GB

CREATE TABLE logs.compressed_logs
(
    log_time DateTime CODEC(Delta(4), LZ4),
    service LowCardinality(String) CODEC(LZ4),
    level LowCardinality(String) CODEC(LZ4),
    host LowCardinality(String) CODEC(LZ4),
    message String CODEC(ZSTD(3)),
    stack_trace String CODEC(ZSTD(5)),
    extra Map(String, String) CODEC(ZSTD(1))
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(log_time)
ORDER BY (log_time, service, host)
TTL log_time + INTERVAL 90 DAY DELETE;

-- 查看实际压缩效果
SELECT
    column,
    formatReadableSize(sum(data_compressed_bytes)) AS compressed,
    formatReadableSize(sum(data_uncompressed_bytes)) AS uncompressed,
    round(sum(data_uncompressed_bytes) / sum(data_compressed_bytes), 1) AS ratio
FROM system.parts
ARRAY JOIN [name] AS column
WHERE table = 'compressed_logs' AND active = 1
GROUP BY column;

CREATE TABLE analytics.compressed_events
(
    event_date Date CODEC(Delta(2), LZ4),
    event_time DateTime64(3) CODEC(Delta(8), ZSTD(1)),
    user_id UInt64 CODEC(T64, ZSTD(1)),
    session_id UUID CODEC(LZ4),
    event_type LowCardinality(String) CODEC(LZ4),
    page_url String CODEC(ZSTD(1)),
    referrer String CODEC(ZSTD(1)),
    user_agent String CODEC(ZSTD(3)),
    ip IPv4 CODEC(T64, LZ4),
    device_info Map(String, String) CODEC(ZSTD(1))
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_date)
ORDER BY (event_date, user_id, event_time);

-- ❌ 低效: 使用 String 存储低基数数据
status String     -- 唯一值只有 5 个，但每个值都存储完整字符串

-- ✅ 高效: 使用 LowCardinality(String) 或 Enum8
status LowCardinality(String)  -- 字典编码，存储整数索引
status Enum8('pending'=1, 'active'=2, 'deleted'=3)  -- 1 字节整数

-- ❌ 低效: 使用 DateTime 存储日期
event_date DateTime  -- 8 字节

-- ✅ 高效: 使用 Date 存储日期
event_date Date      -- 2 字节

-- 查找压缩效果最差的列
SELECT
    table,
    column,
    formatReadableSize(sum(data_compressed_bytes)) AS compressed,
    formatReadableSize(sum(data_uncompressed_bytes)) AS uncompressed,
    round(sum(data_uncompressed_bytes) / greatest(sum(data_compressed_bytes), 1), 2) AS ratio
FROM system.parts AS p
LEFT JOIN system.columns AS c ON p.table = c.table AND p.database = c.database
WHERE p.active = 1 AND p.database = currentDatabase()
GROUP BY table, column
HAVING ratio < 2
ORDER BY ratio ASC;

-- 查找占用空间最大的列
SELECT
    table,
    column,
    formatReadableSize(sum(data_compressed_bytes)) AS size
FROM system.parts
ARRAY JOIN [name] AS column
WHERE active = 1 AND database = currentDatabase()
GROUP BY table, column
ORDER BY sum(data_compressed_bytes) DESC
LIMIT 20;

-- 压缩统计信息
SELECT
    database,
    table,
    column,
    formatReadableSize(sum(data_compressed_bytes)) AS compressed_size,
    formatReadableSize(sum(data_uncompressed_bytes)) AS uncompressed_size,
    round(sum(data_uncompressed_bytes) / sum(data_compressed_bytes), 2) AS compression_ratio,
    count() AS part_count
FROM system.parts
WHERE active = 1 AND database = currentDatabase()
GROUP BY database, table, column
ORDER BY sum(data_compressed_bytes) DESC
LIMIT 20;

-- 整体压缩率
SELECT
    database,
    table,
    formatReadableSize(sum(data_compressed_bytes)) AS total_compressed,
    formatReadableSize(sum(data_uncompressed_bytes)) AS total_uncompressed,
    round(sum(data_uncompressed_bytes) / sum(data_compressed_bytes), 2) AS ratio,
    sum(rows) AS total_rows,
    count() AS parts_count
FROM system.parts
WHERE active = 1 AND database = currentDatabase()
GROUP BY database, table
ORDER BY sum(data_compressed_bytes) DESC;