系统知识 · 数据库系统
数据库的四个关键词：B+ 树索引、事务 ACID、隔离级别、执行计划——从使用到底层。
知识简介
数据库系统 提供持久化存储与高效查询。理解重点：索引为何加速（B+ 树）、事务如何保证（ACID）、并发隔离级别如何取舍、慢查询如何优化（EXPLAIN）。
核心概念
- B+ 树索引：非叶子存键、叶子存数据且有链表，适合范围查询；聚簇索引=主键即数据。
- 事务 ACID：原子性（undo log）、一致性（应用约束）、隔离性（锁/MVCC）、持久性（redo log）。
- 隔离级别：读未提交（脏读）/读已提交（不可重复读）/可重复读（幻读）/串行化。
- MVCC：多版本并发控制，快照读不加锁，写写才冲突。
- 执行计划：EXPLAIN 看 type / key / rows，全表扫描（ALL）是慢查询头号元凶。
代码示例：索引与慢查询分析
sql
-- 建表并加索引
CREATE TABLE users (
id INT PRIMARY KEY AUTO_INCREMENT,
email VARCHAR(64) NOT NULL,
city VARCHAR(32),
created_at DATETIME
);
ALTER TABLE users ADD INDEX idx_email (email); -- 等值查询加速
ALTER TABLE users ADD INDEX idx_city_time (city, created_at); -- 联合索引
-- 查看执行计划：type=ref/range 说明用到索引，ALL 是全表扫描
EXPLAIN SELECT * FROM users WHERE city = 'Shenzhen' ORDER BY created_at;
-- 反例：对索引列做函数运算会导致索引失效
EXPLAIN SELECT * FROM users WHERE YEAR(created_at) = 2026; -- 尽量改为范围查询代码示例：事务与隔离级别
sql
-- 查看/设置隔离级别（MySQL）
SELECT @@transaction_isolation;
SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;
-- 事务示例：转账
START TRANSACTION;
UPDATE accounts SET balance = balance - 100 WHERE id = 1;
UPDATE accounts SET balance = balance + 100 WHERE id = 2;
COMMIT; -- 失败时 ROLLBACK，保证原子性
-- 模拟脏读（读未提交级别下，事务 A 未提交的修改能被 B 读到）学习路径
- 用
EXPLAIN分析自己项目的慢查询，学会看索引命中。 - 读《高性能 MySQL》索引与事务章节。
- 亲手实现一个极简 B+ 树（Python），理解层高与扇出。
- 进阶：读 MySQL 源码 InnoDB 的 MVCC 实现。
完整可运行工程
可直接运行零第三方依赖Python 3.8+
运行环境：Python 3.8+（sqlite3 为标准库）。把 5 个文件放进同一目录后，先执行 python index_lab.py --selftest 零配置验证索引命中与事务回滚；再执行 python index_lab.py --rows 2000，会真实建库并打印"有索引 / 无索引"两条查询计划，可直接看到 SEARCH ... USING INDEX 与 SCAN 的差别。
knowledge-db-lab/
├── index_lab.py 造数 / 查询计划对比 / 事务转账演示（含 --selftest）
├── test_index_lab.py 5 个 unittest 回归用例
├── schema.sql 建表与组合索引定义（可被 sqlite3 直接执行）
├── requirements.txt 依赖清单（仅标准库）
└── README.md 运行步骤 / 索引与隔离级别要点
index_lab.py
主脚本：建库造数、EXPLAIN QUERY PLAN 对比、单事务转账与回滚，含 --selftest 与 CLI。
python
# index_lab.py：索引与事务实验室（SQLite，标准库即可运行）
# 用法：
# python index_lab.py --selftest 零配置自测
# python index_lab.py --rows 2000 建库造数 + 对比索引命中/全表扫描 + 事务演示
import argparse
import os
import sqlite3
import tempfile
HERE = os.path.dirname(os.path.abspath(__file__))
SCHEMA = os.path.join(HERE, "schema.sql")
FIND_SQL = ("SELECT id, amount FROM orders WHERE user_id = ? AND status = ? "
"ORDER BY created_at DESC LIMIT 20")
DROP_ALL = ("DROP INDEX IF EXISTS idx_orders_user_status;\n"
"DROP INDEX IF EXISTS idx_orders_user_created;\n")
def connect(path):
con = sqlite3.connect(path)
con.row_factory = sqlite3.Row
return con
def init(con, rows=2000):
con.executescript(open(SCHEMA, encoding="utf-8").read())
con.execute("DELETE FROM orders")
con.executemany(
"INSERT INTO orders (user_id, status, amount, created_at) VALUES (?, ?, ?, ?)",
((i % 50, ("paid", "pending", "refund")[i % 3], 100 + i,
"2026-09-%02d" % (i % 28 + 1)) for i in range(rows)),
)
con.commit()
return con.execute("SELECT COUNT(*) AS c FROM orders").fetchone()["c"]
def plan(con, sql, params=()):
return [r["detail"] for r in con.execute("EXPLAIN QUERY PLAN " + sql, params)]
def plan_once(db, sql, params=()):
# 单次连接的执行计划：换 schema 后必须用新连接，避免 sqlite3 语句缓存返回旧计划
con = connect(db)
try:
return plan(con, sql, params)
finally:
con.close()
def indexes(con):
return sorted(r["name"] for r in con.execute(
"SELECT name FROM sqlite_master WHERE type = 'index' AND name LIKE 'idx_%'"))
def exec_sql_script(db, script):
con = connect(db)
try:
con.executescript(script)
con.commit()
finally:
con.close()
def find_plan(db, user_id=7):
# 同一 SQL 的两种 schema：索引命中 / 索引全部移除后的全表扫描
with_index = plan_once(db, FIND_SQL, (user_id, "paid"))
exec_sql_script(db, DROP_ALL)
without_index = plan_once(db, FIND_SQL, (user_id, "paid"))
exec_sql_script(db, open(SCHEMA, encoding="utf-8").read()) # 恢复索引
return with_index, without_index
def transfer(con, src, dst, amount):
# 单事务内两笔 UPDATE：任一步失败整体回滚，转账前后总额守恒
try:
con.execute("BEGIN IMMEDIATE")
row = con.execute("SELECT balance FROM accounts WHERE id = ?", (src,)).fetchone()
if row is None or row["balance"] < amount:
raise ValueError("余额不足")
con.execute("UPDATE accounts SET balance = balance - ? WHERE id = ?", (amount, src))
con.execute("UPDATE accounts SET balance = balance + ? WHERE id = ?", (amount, dst))
con.commit()
return True
except Exception:
con.rollback()
raise
def total(con):
return con.execute("SELECT COALESCE(SUM(balance), 0) AS s FROM accounts").fetchone()["s"]
SEED_ACC = ("INSERT OR IGNORE INTO accounts (id, name, balance) "
"VALUES (1, 'alice', 500), (2, 'bob', 300);")
def selftest():
with tempfile.TemporaryDirectory() as td:
db = os.path.join(td, "lab.db")
con = connect(db)
try:
assert init(con, 2000) == 2000
finally:
con.close() # Windows 下先关连接，TemporaryDirectory 才能删掉 db 文件
with_index, without_index = find_plan(db, 7)
assert any("USING INDEX" in x.upper() for x in with_index), with_index
assert any("SCAN" in x.upper() for x in without_index), without_index
con = connect(db)
try:
assert "idx_orders_user_status" in indexes(con), "索引未恢复"
con.executescript(SEED_ACC)
con.commit()
before = total(con)
assert transfer(con, 1, 2, 200) is True
assert total(con) == before, (total(con), before)
try:
transfer(con, 2, 1, 10 ** 9)
raise AssertionError("余额不足应当抛错")
except ValueError:
pass
assert total(con) == before, "回滚后总额应不变"
assert con.execute("SELECT balance FROM accounts WHERE id = 1").fetchone()["balance"] == 300
finally:
con.close()
print("selftest OK：索引命中 / 无索引全扫对比 / 事务守恒与回滚 用例通过")
def main(argv=None):
ap = argparse.ArgumentParser(description="索引与事务实验室（SQLite）")
ap.add_argument("--db", default="orders.db", help="数据库文件路径")
ap.add_argument("--rows", type=int, default=2000, help="造数行数")
ap.add_argument("--selftest", action="store_true", help="零配置自测")
a = ap.parse_args(argv)
if a.selftest:
selftest()
return 0
con = connect(a.db)
try:
n = init(con, a.rows)
finally:
con.close()
with_index, without_index = find_plan(a.db)
print("已建库 %s，orders 行数：%d" % (a.db, n))
print("命中索引计划：", " / ".join(with_index))
print("移除索引计划：", " / ".join(without_index))
con = connect(a.db)
try:
con.executescript(SEED_ACC)
con.commit()
before = total(con)
transfer(con, 1, 2, 100)
print("转账 100 前后总额：%d -> %d（应相等）" % (before, total(con)))
finally:
con.close()
return 0
if __name__ == "__main__":
raise SystemExit(main())
test_index_lab.py
unittest 回归：行数校验、索引命中、无索引全扫、转账总额守恒、余额不足回滚。
python
# 回归测试：python -m unittest -v test_index_lab
import os
import tempfile
import unittest
import index_lab as lab
class TestIndexLab(unittest.TestCase):
def setUp(self):
self.td = tempfile.TemporaryDirectory()
self.db = os.path.join(self.td.name, "t.db")
self.con = lab.connect(self.db)
lab.init(self.con, 500)
def tearDown(self):
self.con.close()
self.td.cleanup()
def test_rows_loaded(self):
self.assertEqual(self.con.execute("SELECT COUNT(*) c FROM orders").fetchone()["c"], 500)
def test_index_used(self):
detail = " ".join(lab.plan(self.con, lab.FIND_SQL, (3, "paid"))).upper()
self.assertIn("USING INDEX", detail)
def test_plan_differs_without_index(self):
with_index, without_index = lab.find_plan(self.db)
self.assertTrue(any("USING INDEX" in x.upper() for x in with_index), with_index)
self.assertTrue(any("SCAN" in x.upper() for x in without_index), without_index)
self.assertIn("idx_orders_user_status", lab.indexes(self.con))
def test_transfer_keeps_total(self):
self.con.executescript("INSERT INTO accounts (id,name,balance) VALUES (1,'a',500),(2,'b',100);")
self.con.commit()
before = lab.total(self.con)
lab.transfer(self.con, 1, 2, 50)
self.assertEqual(lab.total(self.con), before)
self.assertEqual(self.con.execute("SELECT balance FROM accounts WHERE id=2").fetchone()["balance"], 150)
def test_rollback_on_insufficient(self):
self.con.executescript("INSERT INTO accounts (id,name,balance) VALUES (1,'a',50),(2,'b',10);")
self.con.commit()
with self.assertRaises(ValueError):
lab.transfer(self.con, 1, 2, 999)
self.assertEqual(self.con.execute("SELECT balance FROM accounts WHERE id=1").fetchone()["balance"], 50)
if __name__ == "__main__":
unittest.main(verbosity=2)
schema.sql
DDL：orders / accounts 两表 + 组合索引与排序索引，含 CHECK 约束。
sql
-- knowledge-db-lab / schema.sql：订单表 + 账户表 + 覆盖索引
PRAGMA journal_mode = WAL;
CREATE TABLE IF NOT EXISTS orders (
id INTEGER PRIMARY KEY,
user_id INTEGER NOT NULL,
status TEXT NOT NULL,
amount INTEGER NOT NULL,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS accounts (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
balance INTEGER NOT NULL CHECK (balance >= 0)
);
-- 组合索引：按 user_id + status 过滤时可直接命中，不再全表扫描
CREATE INDEX IF NOT EXISTS idx_orders_user_status ON orders (user_id, status);
-- 排序索引：ORDER BY created_at 时避免临时排序
CREATE INDEX IF NOT EXISTS idx_orders_user_created ON orders (user_id, created_at);
requirements.txt
依赖清单：仅 Python 标准库（sqlite3），无第三方包。
text
# 仅标准库（sqlite3 为 Python 内置模块），无需 pip install
README.md
运行步骤、索引与事务要点表、与页面正文的对应关系。
markdown
# knowledge-db-lab
最小可跑的"索引 + 事务"实验工程，零第三方依赖。
## 快速开始
```bash
python index_lab.py --selftest # 零配置自测（临时库）
python index_lab.py --rows 2000 # 生成 orders.db 并打印查询计划
python -m unittest -v test_index_lab # 5 个回归用例
```
## 文件
| 文件 | 作用 |
| --- | --- |
| index_lab.py | 造数、查询计划对比、事务转账 |
| test_index_lab.py | 回归测试 |
| schema.sql | 建表与索引 DDL |
## 要点
- 组合索引 `(user_id, status)` 的列顺序要与 WHERE 的条件顺序对齐，才能同时命中过滤与排序。
- `EXPLAIN QUERY PLAN` 中 `SEARCH ... USING INDEX` 表示走索引，`SCAN` 表示全表扫描。
- 事务用 `BEGIN IMMEDIATE` 提前拿写锁，避免升级锁导致的死锁；`CHECK (balance >= 0)` 在数据库层兜底。
运行标注：①可直接运行——python index_lab.py --selftest 与 python -m unittest -v test_index_lab 均零配置通过；②需按需修改——实际业务表请替换 schema.sql 中的字段与索引，本工程不对接任何线上库；③换数据库——页面正文同时给出 MySQL 的 EXPLAIN 与 PostgreSQL 的 EXPLAIN ANALYZE 写法，语义一致，语法略有差异。
依赖清单：零第三方依赖 —— 仅 Python 标准库 sqlite3（Python 3.8+）；命令行工具 sqlite3 可选
可直接运行：可直接运行 python index_lab.py --selftest 零配置自测；python index_lab.py --rows 2000 建库并打印「有索引 / 无索引」查询计划
需先改的配置项：无配置项需要修改（默认在临时目录建库）。想调数据量改 --rows，想换库名改 --db