压缩含短文本列的数据库并保留快速前缀索引查询方案咨询

场景背景 #

你需要处理数十亿条包含短文本（长度<20字符）和整数ID的数据集，目标是最小化磁盘占用，同时保留基于短文本的快速前缀查询能力。当前测试显示：

• 原始单条字符串平均约11字节
• 带索引存储时单条膨胀至约40字节（4倍膨胀）
• 无索引存储时单条约20字节
  已排除几种Windows环境下难以部署或成本过高的SQLite VFS压缩方案。

可行替代方案 #

1. 字典编码（Dictionary Encoding） #

利用数据中存在的重复模式（如foo、foo1234），将所有唯一短文本映射为整数ID，实现存储压缩：

• 创建两个表：
  • 字典表：CREATE TABLE dict(text_id INTEGER PRIMARY KEY AUTOINCREMENT, original_text TEXT UNIQUE);
  • 主数据表：CREATE TABLE data(id INTEGER PRIMARY KEY AUTOINCREMENT, text_id INTEGER REFERENCES dict(text_id));
• 插入数据时，先检查字典表是否存在该文本，不存在则插入并获取text_id，再将text_id存入主数据表
• 前缀查询逻辑：先查询字典表中符合前缀条件的所有text_id，再用这些ID关联主数据表获取结果
• 优势：主数据表仅存整数，磁盘占用极低；索引可建在字典表的original_text上，前缀查询效率不受影响；重复率越高，压缩效果越显著

2. SQLite内置存储优化 #

通过调整表结构减少冗余：

• 使用VARCHAR(20)替代TEXT：虽然SQLite本质上不区分两者，但明确长度约束会触发内部存储优化，减少额外开销
• 启用WITHOUT ROWID：由于你的主表已经有自增整数主键，可改为CREATE TABLE data(id INTEGER PRIMARY KEY AUTOINCREMENT, a VARCHAR(20)) WITHOUT ROWID;，避免SQLite默认生成的ROWID额外存储，进一步降低单条数据占用

3. 自定义压缩+分离索引表 #

针对你提出的「存储lz4压缩字符串，保留原文本索引」的需求，可通过双表分离实现：

• 创建存储表：CREATE TABLE compressed_data(id INTEGER PRIMARY KEY AUTOINCREMENT, a_compressed BLOB);（用BLOB存储压缩后的二进制数据）
• 创建索引表：CREATE TABLE text_index(a TEXT UNIQUE, id INTEGER REFERENCES compressed_data(id));，并在a字段建前缀索引：CREATE INDEX idx_prefix ON text_index(a);
• 插入流程：对原文本s执行lz4.compress(s)，将压缩结果存入compressed_data，同时将原文本s和对应id存入text_index
• 查询流程：先通过text_index的前缀条件（如a >= 'd' AND a < 'e'）获取匹配的id，再从compressed_data中取出压缩数据解压后返回
• 注意：短文本的lz4压缩率可能有限（如11字节文本压缩后可能仅节省2-3字节），但结合重复数据的去重（通过text_index的UNIQUE约束），仍能有效降低整体磁盘占用

优化测试示例（字典编码方案） #

N = 100_000
import sqlite3, random, time, os

db = sqlite3.connect('test_dict.db')
# 创建字典表和主数据表
db.execute("CREATE TABLE dict(text_id INTEGER PRIMARY KEY AUTOINCREMENT, original_text TEXT UNIQUE);")
db.execute("CREATE TABLE data(id INTEGER PRIMARY KEY AUTOINCREMENT, text_id INTEGER);")
# 给字典表的文本字段建索引，支持前缀查询
db.execute("CREATE INDEX idx_dict_text ON dict(original_text);")

total_size = 0
for i in range(N):
    s = "".join(random.choice(["foo", "foo1234", "bar", "hello", "whatsup"] + list("abcdefghijk")) for _ in range(5))
    total_size += len(s)
    # 先插入字典表（忽略重复），获取text_id
    db.execute("INSERT OR IGNORE INTO dict(original_text) VALUES (?)", (s,))
    # 查询获取对应的text_id
    text_id = db.execute("SELECT text_id FROM dict WHERE original_text = ?", (s,)).fetchone()[0]
    # 插入主数据表
    db.execute("INSERT INTO data(text_id) VALUES (?)", (text_id,))

db.commit()

# 计算存储占用
print(f"{N} strings, raw data: ~ {total_size / N} bytes per string")
print(f"DB on disk: ~ {os.path.getsize('test_dict.db') / N} bytes per string")

# 测试前缀查询
t0 = time.time()
# 先查字典表得到符合前缀的text_id，再关联主表
query_result = list(db.execute("""
    SELECT d.id, dt.original_text 
    FROM data d
    JOIN dict dt ON d.text_id = dt.text_id
    WHERE dt.original_text >= 'd' AND dt.original_text < 'e';
"""))
print(f"query returned {len(query_result)} results in {time.time() - t0} sec")

内容的提问来源于stack exchange，提问作者Basj