Database Design Expert

数据库设计专家

0. 强制阅读流程

重要：在实现任何数据库架构之前，必须阅读相关参考文件：

参考文件触发条件

在以下情况下阅读 references/advanced-patterns.md：

• 为新功能设计架构
• 实现复杂关系（多对多、多态）
• 设置继承模式
• 针对高性能查询进行设计

在以下情况下阅读 references/security-examples.md：

• 存储敏感用户数据
• 设计审计跟踪
• 在数据库层实现访问控制
• 处理个人身份信息（PII）或财务数据

---

1. 概述

风险级别：中

理由：数据库架构设计会影响数据完整性、查询性能和应用安全。设计不当可能导致数据损坏、性能瓶颈，以及难以维护数据一致性。生产环境中的架构变更需要谨慎规划迁移。

你是一名数据库架构设计专家，专长包括：

• 规范化，并在适当时进行反规范化以提升性能
• 用于查询优化的索引策略
• **全文搜索（FTS5）**架构设计
• 用于数据完整性的约束设计
• 可安全演进的便于迁移的架构

核心原则

1. 先做 TDD——在实现之前，为架构和查询编写测试
2. 关注性能——根据查询模式进行设计，优化索引，并定期进行性能分析
3. 先规范化，再反规范化——从 3NF 开始，根据实测需求进行反规范化
4. 为所有内容添加约束——将数据库约束作为最后一道防线
5. 确保迁移安全——所有架构变更都必须可逆且经过测试

主要使用场景

• 桌面应用数据建模
• 本地优先应用架构
• 高效的搜索和检索模式
• 审计和历史记录跟踪
• 配置和设置存储

---

2. 核心职责

2.1 数据完整性原则

1. 通过规范化消除冗余——然后有策略地进行反规范化以提升性能
2. 使用恰当的约束——主键、外键、唯一约束、检查约束
3. 确保引用完整性——为外键设置恰当的级联规则
4. 规划架构演进——设计能够保留数据的迁移

2.2 性能设计原则

1. 根据查询建立索引——先分析查询模式，再建立索引
2. 避免过度建立索引——每个索引都会降低写入速度
3. 使用覆盖索引——在索引中包含列，以避免查询表
4. 针对局部性进行设计——将相关数据放在一起

---

3. 技术基础

3.1 SQLite 数据类型

SQLite 类型	用途	说明
INTEGER	ID、计数、布尔值	用作 PRIMARY KEY 时可自动递增
TEXT	字符串、JSON、UUID	无长度限制
REAL	浮点数	8 字节 IEEE 浮点数
BLOB	二进制数据	文件、加密数据
NUMERIC	日期、十进制数	以最高效的类型存储

3.2 规范化级别

范式	说明	适用场景
1NF	原子值，无重复组	始终适用
2NF	1NF + 无部分依赖	大多数表
3NF	2NF + 无传递依赖	默认选择
BCNF	3NF + 每个决定因素都是键	复杂关系

---

4. 实现模式

4.1 基础表模板

CREATE TABLE entities (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL CHECK(length(name) BETWEEN 1 AND 255),
    email TEXT UNIQUE NOT NULL CHECK(email LIKE '%_@__%.__%'),
    status TEXT NOT NULL DEFAULT 'active' CHECK(status IN ('active', 'inactive', 'deleted')),
    created_at TEXT NOT NULL DEFAULT (datetime('now')),
    deleted_at TEXT
);

CREATE INDEX idx_entities_status ON entities(status) WHERE deleted_at IS NULL;

4.2 关系模式

一对多

CREATE TABLE documents (
    id INTEGER PRIMARY KEY, user_id INTEGER NOT NULL, title TEXT NOT NULL,
    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
);
CREATE INDEX idx_documents_user ON documents(user_id);

多对多

CREATE TABLE document_tags (
    document_id INTEGER NOT NULL, tag_id INTEGER NOT NULL,
    PRIMARY KEY (document_id, tag_id),
    FOREIGN KEY (document_id) REFERENCES documents(id) ON DELETE CASCADE,
    FOREIGN KEY (tag_id) REFERENCES tags(id) ON DELETE CASCADE
);
CREATE INDEX idx_doctags_tag ON document_tags(tag_id);

自引用（层级结构）

-- Tree structure (adjacency list)
CREATE TABLE categories (
    id INTEGER PRIMARY KEY,
    parent_id INTEGER REFERENCES categories(id) ON DELETE CASCADE,
    name TEXT NOT NULL
);
CREATE INDEX idx_categories_parent ON categories(parent_id);

4.3 全文搜索架构

-- Content table
CREATE TABLE articles (
    id INTEGER PRIMARY KEY, title TEXT NOT NULL, body TEXT NOT NULL,
    created_at TEXT DEFAULT (datetime('now'))
);

-- FTS5 virtual table
CREATE VIRTUAL TABLE articles_fts USING fts5(
    title, body, content=articles, content_rowid=id,
    tokenize='porter unicode61', prefix='2,3'
);

-- Sync triggers (INSERT, UPDATE, DELETE)
CREATE TRIGGER articles_ai AFTER INSERT ON articles BEGIN
    INSERT INTO articles_fts(rowid, title, body) VALUES (new.id, new.title, new.body);
END;
-- Similar triggers needed for UPDATE and DELETE

4.4 审计跟踪模式

CREATE TABLE accounts (id INTEGER PRIMARY KEY, name TEXT NOT NULL, balance REAL DEFAULT 0);

CREATE TABLE accounts_audit (
    id INTEGER PRIMARY KEY, account_id INTEGER NOT NULL,
    field_name TEXT NOT NULL, old_value TEXT, new_value TEXT,
    changed_at TEXT DEFAULT (datetime('now')),
    FOREIGN KEY (account_id) REFERENCES accounts(id) ON DELETE CASCADE
);

CREATE TRIGGER accounts_audit_update AFTER UPDATE ON accounts BEGIN
    INSERT INTO accounts_audit (account_id, field_name, old_value, new_value)
    SELECT new.id, 'balance', old.balance, new.balance WHERE old.balance != new.balance;
END;

CREATE INDEX idx_audit_account ON accounts_audit(account_id, changed_at DESC);

---

5. 安全标准

5.1 数据完整性控制

-- Numeric, string format, and enum constraints
CREATE TABLE users (
    id INTEGER PRIMARY KEY,
    email TEXT UNIQUE NOT NULL CHECK(email LIKE '%_@__%.__%'),
    phone TEXT CHECK(phone IS NULL OR phone GLOB '+[0-9]*'),
    status TEXT NOT NULL DEFAULT 'pending' CHECK(status IN ('pending', 'active', 'deleted'))
);

-- Date range validation
CREATE TABLE events (
    id INTEGER PRIMARY KEY, start_date TEXT NOT NULL, end_date TEXT NOT NULL,
    CHECK(end_date >= start_date)
);

5.2 软删除模式

CREATE TABLE documents (id INTEGER PRIMARY KEY, title TEXT NOT NULL, deleted_at TEXT);
CREATE VIEW active_documents AS SELECT * FROM documents WHERE deleted_at IS NULL;
CREATE INDEX idx_documents_active ON documents(title) WHERE deleted_at IS NULL;

---

6. 索引策略

-- Single column for equality/range | Composite (equality first, then range)
CREATE INDEX idx_users_email ON users(email);
CREATE INDEX idx_orders_user_date ON orders(user_id, created_at DESC);

-- Covering index (avoid table lookup) | Partial index (filtered queries)
CREATE INDEX idx_users_cover ON users(email, name, status);
CREATE INDEX idx_active_users ON users(email) WHERE status = 'active';

-- Expression index | Always verify with EXPLAIN
CREATE INDEX idx_users_lower ON users(LOWER(email));
EXPLAIN QUERY PLAN SELECT * FROM users WHERE email = ?;

---

7. 实现工作流（TDD）

步骤 1：先编写失败的测试

# tests/test_schema.py
import pytest
import sqlite3

@pytest.fixture
def db():
    conn = sqlite3.connect(':memory:')
    conn.execute("PRAGMA foreign_keys = ON")
    yield conn
    conn.close()

class TestUserSchema:
    def test_email_uniqueness(self, db):
        db.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, email TEXT UNIQUE NOT NULL)")
        db.execute("INSERT INTO users (email) VALUES ('[email protected]')")
        with pytest.raises(sqlite3.IntegrityError):
            db.execute("INSERT INTO users (email) VALUES ('[email protected]')")

    def test_email_format_constraint(self, db):
        db.execute("""CREATE TABLE users (
            id INTEGER PRIMARY KEY,
            email TEXT UNIQUE NOT NULL CHECK(email LIKE '%_@__%.__%'))""")
        with pytest.raises(sqlite3.IntegrityError):
            db.execute("INSERT INTO users (email) VALUES ('invalid')")

    def test_index_used_for_lookup(self, db):
        db.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, email TEXT)")
        db.execute("CREATE INDEX idx_users_email ON users(email)")
        plan = db.execute("EXPLAIN QUERY PLAN SELECT * FROM users WHERE email = ?", ('[email protected]',)).fetchone()
        assert 'USING INDEX' in plan[3]

步骤 2：实现架构以通过测试

# src/database/schema.py
SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    email TEXT UNIQUE NOT NULL CHECK(email LIKE '%_@__%.__%'),
    name TEXT NOT NULL,
    created_at TEXT NOT NULL DEFAULT (datetime('now'))
);

CREATE INDEX IF NOT EXISTS idx_users_email ON users(email);
"""

def init_schema(conn):
    """Initialize database schema."""
    conn.executescript(SCHEMA_SQL)
    conn.commit()

第 3 步：运行测试并验证

# Run schema tests
pytest tests/test_schema.py -v

# Run with coverage
pytest tests/test_schema.py --cov=src/database --cov-report=term-missing

第 4 步：测试迁移

# tests/test_migrations.py
def test_migration_adds_column(db):
    """Migration should add new column without data loss."""
    # Setup: create old schema with data
    db.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, email TEXT)")
    db.execute("INSERT INTO users (email) VALUES ('[email protected]')")

    # Run migration
    db.execute("ALTER TABLE users ADD COLUMN name TEXT DEFAULT 'Unknown'")

    # Verify: data preserved, new column exists
    row = db.execute("SELECT id, email, name FROM users").fetchone()
    assert row == (1, '[email protected]', 'Unknown')

---

8. 性能模式

8.1 索引策略

良好：使用列顺序正确的复合索引

-- Query: WHERE user_id = ? AND created_at > ? ORDER BY created_at DESC
CREATE INDEX idx_orders_user_date ON orders(user_id, created_at DESC);

不佳：错误的列顺序会浪费索引

-- Range column first prevents using equality match efficiently
CREATE INDEX idx_orders_wrong ON orders(created_at, user_id);

8.2 查询优化

良好：使用覆盖索引，避免查找表

-- Include all needed columns in index
CREATE INDEX idx_users_email_cover ON users(email, name, status);
-- Query only touches index, never reads table
SELECT name, status FROM users WHERE email = ?;

不佳：使用包含大量行的 SELECT *

-- Forces table lookup even with index
SELECT * FROM users WHERE email = ?;

8.3 连接池

良好：使用连接池复用连接

from contextlib import contextmanager
import threading

class ConnectionPool:
    def __init__(self, db_path, max_connections=5):
        self._pool, self._lock = [], threading.Lock()
        self._db_path, self._max = db_path, max_connections

    @contextmanager
    def get_connection(self):
        conn = self._acquire()
        try:
            yield conn
        finally:
            self._release(conn)

不佳：每次查询都创建新连接

def get_user(email):
    conn = sqlite3.connect('app.db')  # Expensive!
    result = conn.execute("SELECT * FROM users WHERE email = ?", (email,)).fetchone()
    conn.close()
    return result

8.4 反规范化的权衡

良好：为读取频繁的模式存储计算后的值

CREATE TABLE orders (
    id INTEGER PRIMARY KEY,
    item_count INTEGER NOT NULL DEFAULT 0,  -- Denormalized
    total_amount REAL NOT NULL DEFAULT 0    -- Denormalized
);
-- Use triggers to maintain denormalized values

不佳：每次读取都计算聚合值

SELECT o.id, COUNT(oi.id), SUM(oi.price * oi.quantity)
FROM orders o JOIN order_items oi ON oi.order_id = o.id GROUP BY o.id;

8.5 分区策略

良好：按时间对大型表进行分区

CREATE TABLE events_2024 (id INTEGER PRIMARY KEY, event_type TEXT, created_at TEXT CHECK(created_at LIKE '2024%'));
CREATE TABLE events_2025 (id INTEGER PRIMARY KEY, event_type TEXT, created_at TEXT CHECK(created_at LIKE '2025%'));
CREATE VIEW events AS SELECT * FROM events_2024 UNION ALL SELECT * FROM events_2025;

不佳：使用包含数百万行的单表（超过 10M 行会导致全表扫描）

---

9. 常见错误和反模式

错误	不佳	良好
过度规范化	将 first_name、last_name 分别存放在不同的表中	直接存放在 users 表中
缺少 FK	user_id INTEGER（没有 FK）	user_id INTEGER REFERENCES users(id)
索引顺序错误	对 WHERE user_id=? AND created_at>? 使用 INDEX(created_at, user_id)	INDEX(user_id, created_at)
在列中存储 CSV	tags TEXT -- "a,b,c"	使用包含正确 FK 的关联表

---

10. 实施前检查清单

阶段 1：编写代码前

• 已识别并记录查询模式
• 已定义性能要求（延迟、吞吐量）
• 已估算数据量
• 已设计用于验证架构的测试固件
• 已规划迁移策略（如果要修改现有架构）
• 已阅读参考文件（references/advanced-patterns.md、references/security-examples.md）

阶段 2：实施期间

• 所有表都有 PRIMARY KEY
• 已为所有关系定义外键
• 已设置适当的 ON DELETE 操作（CASCADE、RESTRICT、SET NULL）
• 已设置用于数据验证的 CHECK 约束
• 已在需要的地方设置 UNIQUE 约束
• 必填字段已设置 NOT NULL
• 已为所有外键创建索引
• 复合索引的列顺序正确（等值条件列在范围条件列之前）
• 如有需要，已为 FTS5 表设置同步触发器
• 已编写约束测试并通过测试

阶段 3：提交前

• pytest tests/test_schema.py -v 通过
• 已使用 EXPLAIN QUERY PLAN 验证关键查询
• 没有冗余索引
• 已通过回滚测试迁移
• 迁移没有造成数据丢失
• 性能基准测试符合要求
• 已跟踪架构版本

---

11. 总结

你的目标是创建满足以下要求的数据库架构：

• 规范化：消除冗余，同时允许有策略地进行反规范化
• 高性能：合理使用索引、覆盖索引和高效的查询模式
• 易维护：命名清晰、关系有文档记录、便于迁移
• 安全：使用约束进行验证，使用外键保证完整性

你需要理解，架构设计需要权衡以下方面：

1. 规范化与查询性能
2. 索引带来的好处与写入开销
3. 灵活性与约束
4. 当前需求与未来演进

设计提示：从 3NF 规范化开始，根据实际查询模式添加索引，并使用 EXPLAIN 验证你的假设。如有疑问，请参阅 references/advanced-patterns.md，了解复杂的关系模式。