1
# 压缩算法性能基准测试
2
import time
3
import snappy
4
import lz4.frame
5
import zstandard as zstd
6

7
data = b'a' * 100_000_000  # 100MB 测试数据
8

9
# Snappy
10
start = time.time()
11
compressed = snappy.compress(data)
12
snappy_time = time.time() - start
13
snappy_ratio = len(data) / len(compressed)
14

15
# LZ4
16
start = time.time()
17
compressed = lz4.frame.compress(data)
18
lz4_time = time.time() - start
19
lz4_ratio = len(data) / len(compressed)
20

21
# ZSTD
22
cctx = zstd.ZstdCompressor(level=3)
23
start = time.time()
24
compressed = cctx.compress(data)
25
zstd_time = time.time() - start
26
zstd_ratio = len(data) / len(compressed)
27

28
print(f"Snappy: {snappy_time:.3f}s, ratio={snappy_ratio:.1f}x")
29
print(f"LZ4:    {lz4_time:.3f}s, ratio={lz4_ratio:.1f}x")
30
print(f"ZSTD-3: {zstd_time:.3f}s, ratio={zstd_ratio:.1f}x")

1
# RLE 编码
2
def rle_encode(values):
3
    """游程编码"""
4
    if not values:
5
        return []
6

7
    encoded = []
8
    current = values[0]
9
    count = 1
10

11
    for v in values[1:]:
12
        if v == current:
13
            count += 1
14
        else:
15
            encoded.append((current, count))
16
            current = v
17
            count = 1
18
    encoded.append((current, count))
19

20
    return encoded
21

22
# 示例：性别列
23
values = ['M', 'M', 'M', 'F', 'F', 'M', 'M', 'M', 'M', 'F']
24
encoded = rle_encode(values)
25
# [('M', 3), ('F', 2), ('M', 4), ('F', 1)]
26

27
# 压缩比：10 个值 → 4 个对，压缩比 2.5x
28
# 对于低基数字段（性别、状态、国家），RLE 压缩比可达 10–100x

1
# 字典编码
2
class DictionaryEncoding:
3
    def __init__(self):
4
        self.dictionary = {}  # 值 → 索引
5
        self.values = []      # 索引 → 值
6
        self.encoded = []     # 编码后的索引序列
7

8
    def encode(self, value):
9
        """编码一个值"""
10
        if value not in self.dictionary:
11
            idx = len(self.values)
12
            self.dictionary[value] = idx
13
            self.values.append(value)
14
        return self.dictionary[value]
15

16
    def decode(self, idx):
17
        """解码一个索引"""
18
        return self.values[idx]
19

20
# 示例：城市列
21
encoder = DictionaryEncoding()
22
cities = ['北京', '上海', '北京', '广州', '上海', '北京', '深圳', '北京']
23
encoded = [encoder.encode(c) for c in cities]
24
# 字典: {'北京': 0, '上海': 1, '广州': 2, '深圳': 3}
25
# 编码: [0, 1, 0, 2, 1, 0, 3, 0]
26

27
# 存储：4 个字典项 + 8 个 2-bit 索引 = ~4 bytes
28
# 原始：8 × 6 bytes = 48 bytes（假设每个城市 6 字节 UTF-8）
29
# 压缩比：12x

1
# 位打包编码
2
def bit_pack_encode(values):
3
    """位打包编码"""
4
    max_val = max(values)
5
    # 计算需要的位数
6
    bits_per_value = max_val.bit_length()
7

8
    # 将值打包到位数组
9
    packed = 0
10
    offset = 0
11
    for v in values:
12
        packed |= (v << offset)
13
        offset += bits_per_value
14

15
    return packed, bits_per_value, len(values)
16

17
# 示例：年龄列（0-127，7 bits 足够）
18
ages = [25, 30, 45, 28, 35, 42, 50, 22]
19
packed, bits, count = bit_pack_encode(ages)
20
# 每个值只需 7 bits（而非 int32 的 32 bits）
21
# 压缩比：32/7 ≈ 4.6x

1
# Delta 编码
2
def delta_encode(values):
3
    """Delta 编码"""
4
    if not values:
5
        return []
6

7
    encoded = [values[0]]  # 第一个值原样存储
8
    for i in range(1, len(values)):
9
        encoded.append(values[i] - values[i-1])
10

11
    return encoded
12

13
# 示例：时间戳列（毫秒）
14
timestamps = [1718000000000, 1718000000150, 1718000000320,
15
              1718000000500, 1718000000680]
16
encoded = delta_encode(timestamps)
17
# [1718000000000, 150, 170, 180, 180]
18

19
# 原始：5 × 8 bytes = 40 bytes（int64）
20
# 编码：8 + 4 × 2 bytes = 16 bytes（int64 + 4 × int16）
21
# 压缩比：2.5x
22
# 如果再配合位打包，压缩比更高

1
# Parquet 列编码流程
2
def parquet_encode_column(values, data_type):
3
    """Parquet 列编码"""
4
    # 1. 统计分析
5
    cardinality = len(set(values))
6
    is_sorted = all(values[i] <= values[i+1] for i in range(len(values)-1))
7
    null_count = sum(1 for v in values if v is None)
8

9
    # 2. 选择编码策略
10
    if cardinality < 1000:
11
        # 低基数：字典编码
12
        dict_encoded = dictionary_encode(values)
13
        # 字典值再用 RLE 或位打包
14
        if is_sorted:
15
            return rle_encode(dict_encoded)
16
        else:
17
            return bit_pack_encode(dict_encoded)
18

19
    elif is_sorted and data_type in ('INT', 'TIMESTAMP'):
20
        # 排序整数：Delta + 位打包
21
        delta = delta_encode(values)
22
        return bit_pack_encode(delta)
23

24
    else:
25
        # 其他：Plain + 块压缩
26
        return plain_encode(values)  # + LZ4/ZSTD
27

28
# 实际压缩比示例
29
# 性别列（M/F）:     字典 + RLE → 100x
30
# 国家列（~200 值）:   字典 + 位打包 → 30x
31
# 日期列（排序）:     Delta + 位打包 → 20x
32
# 金额列（随机）:     Plain + ZSTD → 3x
33
# 描述列（长文本）:   Plain + ZSTD → 2x

1
-- 创建压缩表
2
CREATE TABLE compressed_table (
3
    id BIGINT PRIMARY KEY,
4
    data VARCHAR(255),
5
    payload BLOB
6
) ENGINE=InnoDB
7
  ROW_FORMAT=COMPRESSED
8
  KEY_BLOCK_SIZE=8;  -- 压缩后页大小 8KB
9

10
-- 查看压缩效果
11
SELECT
12
    table_name,
13
    row_format,
14
    table_rows,
15
    avg_row_length,
16
    data_length,
17
    index_length
18
FROM information_schema.tables
19
WHERE table_name = 'compressed_table';

1
# RocksDB 压缩配置
2
# 典型配置：快速压缩 + 底层高压缩比
3
rocksdb_compression_config = {
4
    "per_level_compression": {
5
        0: "kNoCompression",     # L0: 不压缩（频繁读写）
6
        1: "kLZ4Compression",    # L1: LZ4（快速）
7
        2: "kLZ4Compression",    # L2: LZ4
8
        3: "kZSTD",              # L3+: ZSTD（高压缩比）
9
        4: "kZSTD",
10
        5: "kZSTD",
11
        6: "kZSTD",
12
    },
13
    # 底层压缩：最底层的 SSTable 用最高压缩比
14
    "bottommost_compression": "kZSTD",
15
    "bottommost_compression_level": 9,  # ZSTD level 9
16

17
    # 压缩缓存：避免重复压缩/解压
18
    "compression_dict": True,  # 训练字典
19
}

1
# 压缩对查询性能的影响（TPC-H Q6 示例）
2
# 场景：扫描 lineitem 表 7.2GB
3

4
results = {
5
    "无压缩": {
6
        "io_read": "7.2 GB",
7
        "cpu_decompress": "0s",
8
        "total_time": "12.5s",  # I/O 瓶颈
9
    },
10
    "Snappy": {
11
        "io_read": "2.8 GB",
12
        "cpu_decompress": "1.9s",
13
        "total_time": "6.6s",   # 快 1.9x
14
    },
15
    "ZSTD-3": {
16
        "io_read": "2.1 GB",
17
        "cpu_decompress": "2.1s",
18
        "total_time": "5.7s",   # 快 2.2x
19
    },
20
    "Gzip-6": {
21
        "io_read": "1.7 GB",
22
        "cpu_decompress": "8.5s",
23
        "total_time": "11.3s",  # 反而更慢！CPU 瓶颈
24
    },
25
}

1
-- 创建压缩和非压缩表对比
2
CREATE TABLE normal_table (id INT PRIMARY KEY, data VARCHAR(100))
3
    ENGINE=InnoDB ROW_FORMAT=DYNAMIC;
4

5
CREATE TABLE compressed_table (id INT PRIMARY KEY, data VARCHAR(100))
6
    ENGINE=InnoDB ROW_FORMAT=COMPRESSED KEY_BLOCK_SIZE=4;
7

8
-- 插入相同数据后比较
9
SELECT table_name,
10
       pg_size_pretty(data_length + index_length) AS total_size
11
FROM information_schema.tables
12
WHERE table_schema = 'testdb' AND table_name LIKE '%_table';

1
# RocksDB 压缩统计
2
# 通过 db.GetProperty() 获取
3
rocksdb.compression-ratio-at-level0  # L0 压缩比
4
rocksdb.compression-ratio-at-level1  # L1 压缩比
5
rocksdb.compression-ratio-at-level2  # L2 压缩比
6

7
# 压缩输入/输出字节数
8
rocksdb.compression-input-bytes
9
rocksdb.compression-output-bytes