构建一个最小 SQL 数据库的原型有助于开发者深入理解数据库的核心机制，包括查询解析、执行优化和数据持久化。这种从零开始的实现方式，能揭示 SQL 引擎的内部工作原理，避免将数据库视为黑盒子。在实际工程中，这样的原型可用于快速验证查询逻辑或教育目的，而非生产环境。
SQL 数据库的基本架构通常分为前端（Parser）和后端（Executor 与 Storage）。Parser 负责将用户输入的 SQL 字符串转换为抽象语法树（AST），Executor 则根据 AST 生成执行计划并操作数据，Storage 提供数据读写接口。借鉴经典教程如 SQLite 从零实现，我们可以简化这些组件，支持基本的 CREATE TABLE、INSERT、SELECT 语句。
首先，聚焦 Parser 的实现。Parser 是 SQL 引擎的入口，需要处理词法分析（Lexing）和语法分析（Parsing）。词法分析将 SQL 字符串拆分为 Token 序列，例如关键字（如 SELECT）、标识符（如表名）、操作符（如 =）和字面量（如 'value'）。一个简单的 Lexer 可以使用状态机遍历字符串，识别这些 Token。
例如，在 Python 中实现 Lexer：
class Token:
def __init__(self, type, value):
self.type = type
self.value = value
def lexer(sql):
tokens = []
i = 0
while i < len(sql):
if sql[i].isspace():
i += 1
continue
if sql[i].isalpha():
word = ''
while i < len(sql) and sql[i].isalnum():
word += sql[i]
i += 1
tokens.append(Token('KEYWORD' if word.upper() in ['SELECT', 'INSERT', 'CREATE', 'TABLE'] else 'ID', word))
continue
# 处理操作符、字符串等...
if sql[i] == "'":
i += 1
val = ''
while i < len(sql) and sql[i] != "'":
val += sql[i]
i += 1
i += 1 # 跳过结束引号
tokens.append(Token('STRING', val))
continue
# 类似处理数字、逗号等
return tokens
这个 Lexer 支持基本 Token 识别。接下来是语法分析，使用递归下降解析器构建 AST。对于 CREATE TABLE 语句，AST 可能表示为树状结构：CreateTableNode (table_name, columns=[ColumnNode (name, type)])。
对于 SELECT 语句，AST 包括 SelectNode (columns, table, where_conditions)。一个简单的 Parser 函数可以消费 Token 列表，匹配语法规则：
def parse_select(tokens):
if tokens[0].type == 'KEYWORD' and tokens[0].value.upper() == 'SELECT':
# 解析列、FROM、WHERE
pass
# 返回AST节点
这种实现仅支持子集 SQL，避免复杂如 JOIN 或子查询。实际参数：Token 类型限制为 10 种以内，AST 节点类不超过 5 种，确保解析时间 O (n)。
证据显示，这种 Parser 设计在教程中已被验证有效。例如，在 cstack 的 SQLite 教程中，Lexer 使用有限状态机处理 SQL，Parser 生成字节码前体，支持高效执行。测试时，可用单元测试验证如 "SELECT * FROM users" 解析为正确 AST。
其次，Executor 负责执行 AST。它将 AST 转换为逻辑执行计划，然后物理执行。简单 Executor 可直接遍历 AST：对于 SELECT，扫描 Storage 中的表，应用 WHERE 过滤，投影所需列。
实现一个内存 - based Executor：
class Executor:
def execute(self, ast, storage):
if isinstance(ast, SelectNode):
table = storage.get_table(ast.table)
result = []
for row in table.rows:
if self.eval_where(ast.where, row):
projected = {col: row.get(col, None) for col in ast.columns}
result.append(projected)
return result
def eval_where(self, condition, row):
# 简单条件评估，如 col = value
return True # 占位
这里，eval_where 处理基本谓词。优化点：对于常量过滤，先推下到 Storage 扫描。但在最小实现中，无需查询优化器，仅顺序扫描。参数建议：行缓冲区大小为 1000 行，WHERE 条件支持 =、>、< 三种操作符，回滚策略为简单事务（内存快照）。
从 key-value 存储启发，Executor 可集成简单索引加速扫描，但 prototype 中用全表扫描。风险：无索引时，O (n) 扫描对大表慢；限制造成：忽略 ACID，仅内存一致性。
最后，Storage 层管理数据持久化。最小设计使用内存字典存储表：{table_name: {columns: dict, rows: list of dicts}}。为持久化，序列化到 JSON 文件或二进制。
示例 Storage：
class Storage:
def __init__(self, file_path):
self.tables = {}
self.file_path = file_path
self.load()
def create_table(self, name, columns):
self.tables[name] = {'columns': columns, 'rows': []}
self.persist()
def insert(self, table, row):
self.tables[table]['rows'].append(row)
self.persist()
def get_table(self, name):
return self.tables.get(name)
def persist(self):
with open(self.file_path, 'w') as f:
json.dump(self.tables, f)
def load(self):
try:
with open(self.file_path, 'r') as f:
self.tables = json.load(f)
except FileNotFoundError:
self.tables = {}
这个 Storage 支持基本 CRUD，文件路径如 'db.json'。参数：每 100 次写操作 flush 一次，减少 IO；支持 TEXT 和 INT 类型，长度限 20 字符。借鉴 nan.fyi 的 append-only 文件，可扩展为段文件，但 prototype 用 JSON 简单。
整合组件：REPL 循环读取 SQL，Parser 生成 AST，Executor 执行返回结果，Storage 持久数据。完整 prototype 约 500 行代码，用 Python 快速迭代。
可落地清单：
- 环境：Python 3.8+，无外部依赖。
- 测试用例：CREATE TABLE users (id INT, name TEXT); INSERT INTO users VALUES (1, 'Alice'); SELECT * FROM users;
- 监控点：解析时间 < 1ms，执行延迟 < 10ms for 1000 行。
- 回滚策略：内存事务，用 try-except 包装执行。
- 扩展：添加 B-tree 索引，支持 UPDATE/DELETE。
这种最小 SQL 数据库原型揭示了引擎本质：Parser 桥接用户与内部，Executor 桥接逻辑与物理，Storage 桥接内存与盘。虽简陋，却为深入如 LSM 树或优化器铺路。在工程中，用作教学工具或查询模拟器。
资料来源：
- https://nan.fyi/database (键值存储启发)
- https://cstack.github.io/db_tutorial/ (SQLite 从零教程)
- https://build-your-own.org/database/ (Go 实现参考)