目录
- 一致性状态机
- 预写日志记录（Write-Ahead Logging, WAL）
[TOC]
一致性状态机
对于一个无限增长的序列a[1, 2, 3…], 如果对于任意整数i, a[i]的值满足分布式一致性，这个系统就满足一致性状态机的要求。
基本上所有的系统都会有源源不断的操作, 这时候单独对某个特定的值达成一致是不够的。为了真实系统保证所有的副本的一致性，通常会把操作转化为write-ahead-log(简称WAL). 然后让系统的所有副本对WAL保持一致， 这样每个进程按照顺序执行WAL里的操作，就能保证最终的状态是一致的。
预写日志记录（Write-Ahead Logging, WAL）
在计算机科学中，预写日志记录 （WAL） 是一系列技术，用于在数据库系统中提供原子性和持久性（ACID 的两个属性）
预写日志是一种仅用于附加的辅助磁盘驻留结构，用于崩溃和事务恢复。更改首先记录在日志中，必须先将日志写入稳定存储，然后再将更改写入数据库
让我们通过一个简单的银行转账例子来理解WAL的重要性：
- Alice账户余额：1000元
- Bob账户余额：500元
不使用WAL的情况：
1. 从Alice账户扣除100元 (Alice: 900元)
2. 系统崩溃
3. Bob没有收到钱，但Alice的钱已经扣除！
使用WAL的情况：
1. 写入日志：[事务开始, Alice扣款100元, Bob收款100元]
2. 系统确认日志写入磁盘
3. 执行实际操作：
- 从Alice账户扣除100元
- 给Bob账户增加100元
4. 写入日志：[事务完成]
如果在任何步骤发生崩溃，系统都可以通过查看日志来恢复正确状态。
主要功能
预写日志的主要功能可以概括为:
事务开始
-- 示例：转账操作 BEGIN TRANSACTION; UPDATE accounts SET balance = balance - 100 WHERE user = 'Alice'; UPDATE accounts SET balance = balance + 100 WHERE user = 'Bob'; COMMIT;日志记录生成
LSN: 1001 TransID: T123 Type: BEGIN Timestamp: 2024-03-21 10:00:00 LSN: 1002 TransID: T123 Type: UPDATE Table: accounts Before: {user: 'Alice', balance: 1000} After: {user: 'Alice', balance: 900} LSN: 1003 TransID: T123 Type: UPDATE Table: accounts Before: {user: 'Bob', balance: 500} After: {user: 'Bob', balance: 600} LSN: 1004 TransID: T123 Type: COMMIT恢复示例
如果系统在LSN 1002后崩溃：
1. 分析日志：发现未完成的事务T123 2. 回滚操作：将Alice的余额恢复到1000 3. 记录回滚日志（CLR）- LSN (Log Sequence Number): 唯一标识日志记录
- 事务ID：标识产生该日志记录的事务
- 操作类型：INSERT、UPDATE、DELETE等
- 表ID：受影响的表标识
- 前镜像：修改前的数据
- 后镜像：修改后的数据
- 时间戳：日志记录的创建时间
- 模糊检查点：允许在检查点过程中继续处理新的事务
- 精确检查点：暂停所有新事务直到检查点完成
- 增量检查点：将脏页分批写入磁盘，减少I/O峰值
- 写入性能更好（顺序写入）
- 空间利用率高
- 实现相对简单
- 支持细粒度的并发控制
- 需要定期执行检查点操作
- 恢复时间可能较长
- 日志管理的开销
- 写放大问题
- PostgreSQL
- SQLite
- MySQL (InnoDB引擎)
- 分布式一致性协议（如Raft）
- 消息队列系统
- 分布式缓存
- 日志结构文件系统
- 事务性文件系统
- 写前日志（Write-Ahead Logging）
- 所有修改必须先写入日志，再写入数据页
- 日志记录必须在事务提交前写入稳定存储
- 重复历史（Repeating History）
- 系统崩溃后的恢复过程必须重做所有操作
- 即使某些事务最终会被回滚，也需要先重做
- 存储前后镜像（Store Before- and After-Images）
- 每个更新操作都记录修改前的值（前镜像）和修改后的值（后镜像）
- 用于支持事务回滚和恢复操作
更新日志记录
补偿日志记录(CLR)
<LSN, TransID, PageID, UndoNextLSN, RedoInfo>- 分析阶段（Analysis Pass）
- 从最后一个检查点开始扫描日志
- 重建事务表（Transaction Table）和脏页表（Dirty Page Table）
- 确定活跃事务集合和脏页集合
- 确定RedoLSN（重做起点）
- 重做阶段（Redo Pass）
- 从RedoLSN开始扫描日志
- 对所有日志记录（包括CLR）进行重做
- 重做条件：
- 页面LSN < 日志记录LSN
- 页面在脏页表中或当前LSN >= RecLSN
- 回滚阶段（Undo Pass）
- 回滚所有分析阶段确定的未完成事务
- 从后向前处理，生成补偿日志记录（CLR）
- 按照UndoNextLSN链进行回滚
- 分析阶段：
- T1处于提交状态
- T2处于活跃状态
- P1和P2在脏页表中
- 重做阶段：
- 重做LSN 110和130的更新操作
- 回滚阶段：
- 只需回滚T2的操作
- 生成CLR回滚LSN 130的操作
- 日志管理
- 合理配置日志大小
- 实现日志压缩/归档机制
- 监控日志增长速度
- 性能优化
- 批量写入日志
- 异步刷盘策略
- 使用预写缓冲区
- 监控与维护
- 监控检查点频率
- 跟踪日志空间使用
- 定期验证恢复功能
详细恢复步骤
graph TD A[系统崩溃/启动恢复] --> B[分析阶段] B --> C[重建事务表和脏页表] C --> D[确定RedoLSN] D --> E[重做阶段] E --> F{是否需要重做?} F -->|是| G[重做操作] F -->|否| H[跳过] G --> I[下一个日志记录] H --> I I --> J{日志结束?} J -->|否| F J -->|是| K[回滚阶段] K --> L[处理未完成事务] L --> M{需要回滚?} M -->|是| N[生成CLR并回滚] M -->|否| O[恢复完成] N --> P[更新UndoNextLSN] P --> L示例场景
假设有以下日志记录序列：
LSN TransID Type PrevLSN PageID Data
100 T1 BEGIN NULL - -
110 T1 UPDATE 100 P1 x=10->20
120 T2 BEGIN NULL - -
130 T2 UPDATE 120 P2 y=5->15
140 T1 COMMIT 110 - -
如果在LSN 135时系统崩溃：
最佳实践
Python实现简单的WAL
这里的实现比较简单，并且有些逻辑上的漏洞
1、只考虑了update的情况
2、commit是假的commit,只要进行了update都会写wal.log，所以就都认为是一定需要recover的，没有区分undo.log和redo.log，可以留着有空再改进
一些疑问
归档之前记录检查点
在归档之前记录检查点是WAL机制中的一个重要设计，主要有以下几个原因：
1. 确保数据一致性def create_checkpoint(self) -> Dict[str, Any]: """创建检查点，标记当前系统的一致性状态""" checkpoint = { "timestamp": time.time(), "lsn": self.current_lsn, # 记录当前的LSN "records_count": self.records_count } # ... 保存检查点信息 ...- 优化恢复过程
- 没有检查点的情况：
- 有检查点的情况
sequenceDiagram participant DB as 数据库 participant WAL as WAL日志 participant Disk as 磁盘 Note over DB,Disk: 系统崩溃后重启 DB->>WAL: 读取最近的检查点 WAL->>Disk: 只需读取检查点之后的日志 Note over DB: 只需重放检查点之后的操作 Note over DB: 显著提高恢复效率- 安全归档的前提
def archive_logs(self, checkpoint_lsn: int): """归档检查点之前的日志""" # 1. 检查点确保了这些数据已经写入磁盘 # 2. 检查点之前的日志可以安全归档 for record in records: if record.lsn <= checkpoint_lsn: # 检查点之前的记录 archive_records.append(record) else: keep_records.append(record)# 检查点提供了安全归档的界限 if self.records_count >= self.checkpoint_size: checkpoint = self.create_checkpoint() # 先创建检查点 self.archive_logs(checkpoint["lsn"]) # 再根据检查点LSN归档- 数据库恢复的保障
def recover_from_wal(self): """从WAL恢复数据库状态""" # 1. 找到最近的检查点 checkpoint = self.find_latest_checkpoint() # 2. 只需要处理检查点之后的日志 records = self.wal.recover_with_archives() recent_records = [r for r in records if r.lsn > checkpoint["lsn"]] # 3. 重放操作 for record in recent_records: if record.operation == "UPDATE": self.apply_update(record)检查点机制在归档前是必要的，因为它
- 标记了一个已知的一致性状态
- 提供了安全归档的边界
- 优化了恢复流程
- 保证了数据的完整性
- 提供了系统状态的快照
- 支持增量备份和恢复
这就是在WAL中归档操作总是与检查点机制配合使用
设计
prompt
请你结合database.py和wal.py实现，用mermaid语法绘制1. 展示系统的静态结构 2. 展示更新操作的流程 3. 展示恢复操作的流程： 4. 展示WAL记录的生命周期1. 展示系统的静态结构
classDiagram class Database { -Dict data -WAL wal +update(trans_id, table, key, value) +get(table, key) +commit() +recover_from_wal() } class WAL { -str log_path -int current_lsn -int checkpoint_size -str archive_dir -str checkpoint_dir -int records_count +append(trans_id, operation, table, before, after) +flush() +recover() +recover_with_archives() +create_checkpoint() +archive_logs(checkpoint_lsn) } class WALRecord { +int lsn +str trans_id +str operation +str table +Dict before +Dict after +float timestamp +to_dict() +from_dict() } Database --> WAL : uses WAL --> WALRecord : creates2. 展示更新操作的流程
sequenceDiagram participant Client participant Database participant WAL participant Disk Client->>Database: update(trans_id, table, key, value) Database->>WAL: append(trans_id, "UPDATE", table, before, after) WAL->>Disk: 写入WAL记录 Disk-->>WAL: 确认写入 WAL-->>Database: 返回LSN Database->>Database: 更新内存数据 alt records_count >= checkpoint_size WAL->>WAL: create_checkpoint() WAL->>Disk: 写入检查点 WAL->>WAL: archive_logs() WAL->>Disk: 写入归档文件 WAL->>WAL: 清理WAL文件 end Database-->>Client: 完成更新3. 展示恢复操作的流程：
sequenceDiagram participant Database participant WAL participant Archive participant Checkpoint participant WALFile Database->>WAL: recover_from_wal() WAL->>Archive: 读取最新归档状态 Archive-->>WAL: 返回归档状态 WAL->>Checkpoint: 查找最新检查点 Checkpoint-->>WAL: 返回检查点信息 WAL->>WALFile: 读取增量记录 WALFile-->>WAL: 返回WAL记录 WAL->>WAL: 合并状态 WAL->>Database: 返回恢复记录 Database->>Database: 重建内存数据4. 展示WAL记录的生命周期：
stateDiagram-v2 [*] --> 活跃记录: 写入WAL文件 活跃记录 --> 检查点前记录: 达到checkpoint_size 检查点前记录 --> 归档状态: archive_logs() 归档状态 --> 最终状态: 合并到归档文件 活跃记录 --> 恢复数据: 系统重启 归档状态 --> 恢复数据: 系统重启 恢复数据 --> [*]: 加载到内存 note right of 活跃记录: WAL文件中的记录 note right of 检查点前记录: LSN <= checkpoint_lsn note right of 归档状态: 转换为最终数据状态 note right of 最终状态: state_{timestamp}.json代码
database.py
from typing import Dict, Any from wal import WAL import os class Database: """简单的内存数据库实现，支持WAL机制 主要功能： 1. 基本的键值对存储 2. 使用WAL确保数据持久性 3. 系统重启时的数据恢复 """ def __init__(self, wal_path: str, checkpoint_dir=None): """初始化数据库 Args: wal_path: WAL日志文件路径 checkpoint_dir: checkpoint目录路径（可选） """ self.wal_path = wal_path self.checkpoint_dir = checkpoint_dir or os.path.dirname(wal_path) self.data: Dict[str, Dict[str, Any]] = {} # 内存中的数据存储 self.wal = WAL(wal_path) # WAL管理器 self.recover_from_wal() # 启动时从WAL恢复数据 def recover_from_wal(self): """从WAL（包括归档）恢复数据库状态""" # 使用recover_with_archives替代recover records = self.wal.recover_with_archives() for record in records: if record.operation == "UPDATE": if record.table not in self.data: self.data[record.table] = {} self.data[record.table].update(record.after) def update(self, trans_id: str, table: str, key: str, value: Any): """更新数据库中的值 遵循WAL原则：先写日志，再更新数据 Args: trans_id: 事务ID table: 表名 key: 键 value: 值 """ if table not in self.data: self.data[table] = {} # 记录修改前的值 before = {key: self.data[table].get(key)} after = {key: value} # 先写日志 self.wal.append(trans_id, "UPDATE", table, before, after) # 再更新数据 self.data[table][key] = value def get(self, table: str, key: str) -> Any: """获取数据库中的值 Args: table: 表名 key: 键 Returns: 对应的值，如果不存在则返回None """ return self.data.get(table, {}).get(key) def commit(self): """提交事务，确保日志写入磁盘""" self.wal.flush()wal.py
import os import json import time from typing import Dict, Any, List, Optional from dataclasses import dataclass, asdict, field @dataclass class WALRecord: """使用dataclass简化日志记录类的定义""" lsn: int trans_id: str operation: str table: str before: Dict[str, Any] after: Dict[str, Any] timestamp: float = field(default_factory=time.time) def to_dict(self) -> Dict[str, Any]: return asdict(self) @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'WALRecord': return cls(**data) class WAL: """WAL管理器""" def __init__(self, log_path: str): """初始化WAL管理器 Args: log_path: 日志文件路径 """ self.log_path = log_path self.current_lsn = 0 self.checkpoint_size = 200 # 修改为较小的值以便测试 # 将archive_dir和checkpoint_dir放在log_path同级目录下 base_dir = os.path.dirname(log_path) self.archive_dir = os.path.join(base_dir, "archive") self.checkpoint_dir = os.path.join(base_dir, "checkpoint") self.records_count = 0 # 确保目录存在 self._init_directories() # 创建文件（如果不存在） if not os.path.exists(self.log_path): with open(self.log_path, 'w') as f: pass self._init_records_count() def _write_to_log(self, data: str): """同步写入日志文件 Args: data: 要写入的数据 """ with open(self.log_path, 'a') as f: f.write(data + '\n') f.flush() os.fsync(f.fileno()) def append(self, trans_id: str, operation: str, table: str, before: Dict[str, Any], after: Dict[str, Any]) -> int: """追加新的日志记录""" self.current_lsn += 1 record = WALRecord(self.current_lsn, trans_id, operation, table, before, after) # 同步写入 self._write_to_log(json.dumps(record.to_dict())) # 更新计数并检查是否需要创建检查点 self.records_count += 1 if self.records_count >= self.checkpoint_size: self._create_checkpoint_and_archive() return self.current_lsn def flush(self): """刷新日志到磁盘（现在是空操作，因为每次写入都是同步的）""" pass def recover(self) -> List[WALRecord]: """从日志文件中恢复记录""" try: # 首先检查最新的检查点 checkpoint = self._find_latest_checkpoint() start_lsn = checkpoint["lsn"] if checkpoint else 0 # 获取归档的状态 final_state = {} archive_files = sorted([f for f in os.listdir(self.archive_dir) if f.startswith("state_") and f.endswith(".json")]) if archive_files: latest_archive = archive_files[-1] with open(os.path.join(self.archive_dir, latest_archive)) as f: archive_data = json.load(f) final_state = archive_data["state"] # 从检查点开始读取增量记录 with open(self.log_path, 'r') as f: for line in f: if line.strip(): record = WALRecord.from_dict(json.loads(line)) if record.lsn > start_lsn: # 更新状态 if record.operation == "UPDATE": if record.table not in final_state: final_state[record.table] = {} final_state[record.table].update(record.after) # 将最终状态转换为WAL记录 result_records = [] for table, table_data in final_state.items(): for key, value in table_data.items(): record = WALRecord( lsn=self.current_lsn + 1, trans_id="RECOVERY", operation="UPDATE", table=table, before={}, after={key: value} ) result_records.append(record) return result_records except Exception as e: print("Error during recovery: {}".format(e)) return [] def _create_checkpoint_and_archive(self): """异步创建检查点和归档""" try: checkpoint = self.create_checkpoint() self.archive_logs(checkpoint["lsn"]) self.records_count = 0 except Exception as e: print(f"Error during checkpoint/archive: {e}") def create_checkpoint(self) -> Dict[str, Any]: """创建检查点，包含更多信息""" checkpoint = { "timestamp": time.time(), "lsn": self.current_lsn, "records_count": self.records_count, "pending_writes": 0 } # 在checkpoint_dir下创建checkpoint文件 checkpoint_path = os.path.join(self.checkpoint_dir, "checkpoint_{}.json".format(checkpoint['lsn'])) with open(checkpoint_path, 'w') as f: json.dump(checkpoint, f) return checkpoint def _find_latest_checkpoint(self) -> Optional[Dict[str, Any]]: """查找最新的检查点""" # 在checkpoint_dir下查找checkpoint文件 checkpoint_files = [f for f in os.listdir(self.checkpoint_dir) if f.startswith("checkpoint_")] if not checkpoint_files: return None latest_checkpoint = max(checkpoint_files) with open(os.path.join(self.checkpoint_dir, latest_checkpoint)) as f: return json.load(f) def __del__(self): """清理资源（不再需要停止线程）""" pass def recover_with_archives(self) -> List[WALRecord]: """从归档状态和当前日志中恢复所有记录 恢复流程： 1. 加载最新的归档状态 2. 读取当前WAL文件中的记录 3. 按LSN顺序应用所有更新 Returns: 转换后的WAL记录列表，表示最终状态 """ try: # 1. 获取最新归档状态 final_state = {} # {table: {key: value}} last_archive_lsn = 0 archive_files = sorted([f for f in os.listdir(self.archive_dir) if f.startswith("state_") and f.endswith(".json")]) if archive_files: latest_archive = archive_files[-1] with open(os.path.join(self.archive_dir, latest_archive)) as f: archive_data = json.load(f) final_state = archive_data["state"] last_archive_lsn = archive_data["metadata"]["checkpoint_lsn"] # 2. 读取并应用当前WAL文件中的所有记录 if os.path.exists(self.log_path): with open(self.log_path, 'r') as f: for line in f: if line.strip(): record = WALRecord.from_dict(json.loads(line)) if record.operation == "UPDATE": if record.table not in final_state: final_state[record.table] = {} final_state[record.table].update(record.after) # 3. 将最终状态转换为WAL记录 result_records = [] for table, table_data in final_state.items(): for key, value in table_data.items(): record = WALRecord( lsn=self.current_lsn + 1, trans_id="RECOVERY", operation="UPDATE", table=table, before={}, after={key: value} ) result_records.append(record) return result_records except Exception as e: print("Error during recovery with archives: {}".format(e)) return [] def _get_relevant_archives(self, checkpoint_lsn: int) -> List[str]: """获取与恢复相关的归档文件列表 Args: checkpoint_lsn: 检查点的LSN Returns: 相关归档文件的完整路径列表 """ if not os.path.exists(self.archive_dir): return [] archive_files = [] for filename in os.listdir(self.archive_dir): if not filename.startswith("wal_") or not filename.endswith(".log"): continue # 从文件名解析LSN范围 try: # 文件名格式：wal_START-LSN_END-LSN.log _, start_lsn, end_lsn = filename.replace(".log", "").split("_") end_lsn = int(end_lsn) # 只选择包含检查点LSN之后记录的文件 if end_lsn >= checkpoint_lsn: archive_files.append(os.path.join(self.archive_dir, filename)) except ValueError: continue # 按文件名排序，确保按时间顺序处理 return sorted(archive_files) def _read_log_file(self, file_path: str) -> List[WALRecord]: """读取日志文件中的所有记录 Args: file_path: 日志文件路径 Returns: 文件中的WALRecord列表 """ records = [] if not os.path.exists(file_path): return records with open(file_path, 'r') as f: for line in f: if line.strip(): try: data = json.loads(line) record = WALRecord.from_dict(data) records.append(record) except json.JSONDecodeError: print(f"Error decoding log record: {line}") except Exception as e: print(f"Error processing log record: {e}") return records def _sort_and_deduplicate_records(self, records: List[WALRecord]) -> List[WALRecord]: """对记录进行排序和去重 Args: records: 原始记录列表 Returns: 排序和去重后的记录列表 """ # 按LSN排序 sorted_records = sorted(records, key=lambda r: r.lsn) # 使用字典去重，保留最新的记录 unique_records = {} for record in sorted_records: unique_records[record.lsn] = record return list(unique_records.values()) def _init_records_count(self): """初始化记录计数 包括当前WAL文件和所有归档文件中的记录 """ self.records_count = 0 # 计算归档文件中的记录数 if os.path.exists(self.archive_dir): for filename in os.listdir(self.archive_dir): if filename.startswith("wal_") and filename.endswith(".log"): file_path = os.path.join(self.archive_dir, filename) with open(file_path, 'r') as f: self.records_count += sum(1 for line in f if line.strip()) # 计算当前WAL文件中的记录数 if os.path.exists(self.log_path): with open(self.log_path, 'r') as f: self.records_count += sum(1 for line in f if line.strip()) def archive_logs(self, checkpoint_lsn: int): """归档检查点之前的日志记录，将其转换为最终数据状态 Args: checkpoint_lsn: 检查点的LSN """ try: # 读取当前WAL文件中需要归档的记录 records_to_archive = [] with open(self.log_path, 'r') as f: for line in f: if line.strip(): record = WALRecord.from_dict(json.loads(line)) if record.lsn <= checkpoint_lsn: records_to_archive.append(record) if not records_to_archive: return # 将WAL记录转换为最终数据状态 final_state = {} # 格式: {table: {key: value}} for record in records_to_archive: if record.operation == "UPDATE": if record.table not in final_state: final_state[record.table] = {} # 遍历after字典中的所有键值对 for key, value in record.after.items(): final_state[record.table][key] = value # 创建归档文件（使用JSON格式存储最终状态） timestamp = int(time.time()) archive_filename = "state_{}.json".format(timestamp) archive_path = os.path.join(self.archive_dir, archive_filename) # 保存最终状态和元数据 archive_data = { "metadata": { "checkpoint_lsn": checkpoint_lsn, "timestamp": timestamp, "record_count": len(records_to_archive) }, "state": final_state } # 写入归档文件 with open(archive_path, 'w') as f: json.dump(archive_data, f, indent=2) # 清理原WAL文件中已归档的记录 remaining_records = [] with open(self.log_path, 'r') as f: for line in f: if line.strip(): record = WALRecord.from_dict(json.loads(line)) if record.lsn > checkpoint_lsn: remaining_records.append(line) # 重写WAL文件，只保留未归档的记录 with open(self.log_path, 'w') as f: for line in remaining_records: f.write(line) except Exception as e: print("Error during log archiving: {}".format(e)) def _init_directories(self): """初始化所需的目录""" # 确保WAL文件所在目录存在 os.makedirs(os.path.dirname(self.log_path), exist_ok=True) # 确保归档目录存在 os.makedirs(self.archive_dir, exist_ok=True) # 确保checkpoint目录存在 os.makedirs(self.checkpoint_dir, exist_ok=True)UT测试
import unittest import os import json from database import Database class TestWAL(unittest.TestCase): """WAL机制的单元测试类 测试内容： 1. 基本的数据更新操作 2. 系统崩溃后的数据恢复 3. 归档功能 """ def __init__(self, *args, **kwargs): """初始化测试类，添加清理标志""" super().__init__(*args, **kwargs) # 默认不清理文件，方便观察 self.should_cleanup = False def setUp(self): """测试前的准备工作 1. 设置测试用的目录结构 2. 确保测试开始前相关文件不存在 3. 创建新的数据库实例 """ # 使用当前目录作为基础目录 self.test_dir = os.getcwd() # 创建WAL日志目录 self.wal_dir = os.path.join(self.test_dir, "wal_logs") if not os.path.exists(self.wal_dir): os.makedirs(self.wal_dir) self.wal_path = os.path.join(self.wal_dir, "test_wal.log") # 清理之前的测试文件 if os.path.exists(self.wal_path): os.remove(self.wal_path) # 清理之前的归档目录 archive_dir = os.path.join(self.test_dir, "archive") if os.path.exists(archive_dir): for f in os.listdir(archive_dir): os.remove(os.path.join(archive_dir, f)) os.rmdir(archive_dir) # 创建数据库实例时传入checkpoint目录 self.db = Database(self.wal_path) def tearDown(self): """测试后的清理工作 只有在should_cleanup为True时才清理文件 """ if not self.should_cleanup: return try: # 清理WAL日志文件和目录 if os.path.exists(self.wal_path): os.remove(self.wal_path) if os.path.exists(self.wal_dir): os.rmdir(self.wal_dir) # 清理checkpoint文件和目录 if os.path.exists(self.checkpoint_dir): for f in os.listdir(self.checkpoint_dir): os.remove(os.path.join(self.checkpoint_dir, f)) os.rmdir(self.checkpoint_dir) # 清理归档目录 archive_dir = os.path.join(self.test_dir, "archive") if os.path.exists(archive_dir): for f in os.listdir(archive_dir): os.remove(os.path.join(archive_dir, f)) os.rmdir(archive_dir) except Exception as e: print("Error during cleanup: {}".format(e)) def test_basic_operations(self): """测试基本的更新操作，使用更多数据 验证： 1. 数据更新是否成功 2. 更新后的值是否正确 3. 多表操作 4. 多种数据类型 """ # 账户表操作 accounts_data = { "Alice": 1000, "Bob": 500, "Charlie": 2000, "David": 1500, "Eve": 3000 } # 用户信息表操作 user_info_data = { "Alice": {"age": 25, "city": "New York", "active": True}, "Bob": {"age": 30, "city": "London", "active": False}, "Charlie": {"age": 35, "city": "Paris", "active": True}, "David": {"age": 28, "city": "Tokyo", "active": True}, "Eve": {"age": 32, "city": "Berlin", "active": True} } # 商品表操作 products_data = { "prod1": {"name": "Laptop", "price": 1200, "stock": 50}, "prod2": {"name": "Phone", "price": 800, "stock": 100}, "prod3": {"name": "Tablet", "price": 600, "stock": 75}, "prod4": {"name": "Watch", "price": 300, "stock": 150}, "prod5": {"name": "Headphones", "price": 200, "stock": 200} } # 更新账户表 for user, balance in accounts_data.items(): self.db.update("T1", "accounts", user, balance) # 更新用户信息表 for user, info in user_info_data.items(): self.db.update("T1", "user_info", user, info) # 更新商品表 for prod_id, info in products_data.items(): self.db.update("T1", "products", prod_id, info) self.db.commit() # 验证账户表数据 for user, balance in accounts_data.items(): self.assertEqual(self.db.get("accounts", user), balance) # 验证用户信息表数据 for user, info in user_info_data.items(): self.assertEqual(self.db.get("user_info", user), info) # 验证商品表数据 for prod_id, info in products_data.items(): self.assertEqual(self.db.get("products", prod_id), info) def test_recovery(self): """测试数据库恢复功能 验证： 1. 写入数据后系统崩溃 2. 重启后是否能正确恢复数据 """ self.db.update("T1", "accounts", "Alice", 1000) self.db.update("T1", "accounts", "Bob", 500) self.db.commit() # 创建新的数据库实例，模拟重启 new_db = Database(self.wal_path) # 验证数据是否正确恢复 self.assertEqual(new_db.get("accounts", "Alice"), 1000) self.assertEqual(new_db.get("accounts", "Bob"), 500) def test_archiving(self): """测试归档功能，使用更多数据触发归档 验证： 1. 写入大量数据触发归档 2. 检查归档文件是否正确生成 3. 验证归档的数据是否正确 """ print("\nStarting archiving test...") print("Test directory:", self.test_dir) # 生成大量测试数据 for i in range(100): # 增加到100个用户 # 账户数据 self.db.update("T1", "accounts", f"user_{i}", 1000 + i * 100) # 用户信息 user_info = { "age": 20 + i % 40, "city": ["New York", "London", "Paris", "Tokyo", "Berlin"][i % 5], "active": i % 3 == 0, "score": i * 10, "registration_date": f"2024-{(i % 12) + 1:02d}-{(i % 28) + 1:02d}" } self.db.update("T1", "user_info", f"user_{i}", user_info) # 交易记录 transaction = { "amount": 100 + i * 50, "type": ["deposit", "withdrawal", "transfer"][i % 3], "timestamp": f"2024-03-{i % 30 + 1:02d} {i % 24:02d}:{i % 60:02d}:00" } self.db.update("T1", "transactions", f"tx_{i}", transaction) self.db.commit() print("Large dataset committed") # 检查归档目录 archive_dir = os.path.join(self.test_dir, "wal_logs", "archive") print("Looking for archives in:", archive_dir) # 确保目录存在 self.assertTrue(os.path.exists(archive_dir), "Archive directory does not exist") archive_files = [f for f in os.listdir(archive_dir) if f.startswith("state_") and f.endswith(".json")] print("Found archive files:", archive_files) # 验证是否生成了归档文件 self.assertTrue(len(archive_files) > 0, "No archive file was created") # 读取最新的归档文件 latest_archive = max(archive_files) archive_path = os.path.join(archive_dir, latest_archive) print("Reading archive file:", archive_path) with open(archive_path) as f: archive_data = json.load(f) print("Archive content sample:", json.dumps( {k: list(v.keys())[:5] for k, v in archive_data["state"].items()}, indent=2 )) # 验证归档的数据是否正确（抽样检查） self.assertIn("accounts", archive_data["state"]) self.assertIn("user_info", archive_data["state"]) self.assertIn("transactions", archive_data["state"]) # 检查几个随机样本 for i in [0, 42, 66]: user_key = f"user_{i}" self.assertEqual( archive_data["state"]["accounts"][user_key], 1000 + i * 100 ) self.assertEqual( archive_data["state"]["user_info"][user_key]["age"], 20 + i % 40 ) if __name__ == '__main__': unittest.main()测试显然是成功的
性能优化建议
1. 日志写入优化
# 不好的做法：每个操作都刷盘 for record in records: write_log(record) flush_to_disk() # 好的做法：批量写入 batch = [] for record in records: batch.append(record) if len(batch) >= BATCH_SIZE: write_log_batch(batch) flush_to_disk() batch = []2. 检查点优化
- WAL生成速率
- 检查点耗时
- 恢复时间
- 日志空间使用率
3. 监控指标
- 分析阶段（Analysis Pass）
工作原理
详细的工作流程
日志记录格式
典型的WAL日志记录包含以下字段：
示例说明
想象一下，当运行它的计算机断电时，一个程序正在执行某些操作。重新启动后，该程序可能需要知道它正在执行的操作是成功、部分成功还是失败。如果使用预写日志，程序可以检查此日志，并将意外断电时应该做什么与实际做什么进行比较。基于这种比较，程序可以决定撤消它已开始的内容、完成已开始的内容或保持原样
图解检查点过程
sequenceDiagram
participant DB as 数据库
participant WAL as WAL日志
participant Disk as 磁盘
Note over DB,Disk: 正常操作
DB->>WAL: 写入操作日志
WAL->>Disk: 定期刷新日志
Note over DB,Disk: 检查点开始
DB->>DB: 暂停新事务(精确检查点)
DB->>Disk: 将脏页写入磁盘
DB->>WAL: 记录检查点位置
DB->>DB: 恢复处理新事务时间点1: 系统正常运行
- WAL大小：1GB
- 脏页数量：1000页
- 活跃事务：10个
时间点2: 触发检查点
- 将1000个脏页写入磁盘
- 记录10个活跃事务的状态
- 标记检查点位置
时间点3: 检查点完成
- 可以清理旧的WAL日志
- 系统恢复只需要从该检查点开始
检查点类型
WAL vs 其他方案
时间点1: 系统正常运行
- WAL大小：1GB
- 脏页数量：1000页
- 活跃事务：10个
时间点2: 触发检查点
- 将1000个脏页写入磁盘
- 记录10个活跃事务的状态
- 标记检查点位置
时间点3: 检查点完成
- 可以清理旧的WAL日志
- 系统恢复只需要从该检查点开始
检查点类型
WAL vs 其他方案
WAL 允许就地完成数据库的更新。实现原子更新的另一种方法是使用影子分页，它不是就地的。就地更新的主要优点是它减少了修改索引和阻止列表的需要。
WAL的优势
WAL的劣势
应用场景
数据库系统
分布式系统
文件系统
ARIES算法
ARIES (Algorithms for Recovery and Isolation Exploiting Semantics) 是WAL最著名的实现之一，是IBM研究院开发的一种数据库恢复算法。
No comments yet.