thzxx
f97c5a6001
- KVStore 混合写单记录原子：新增 writeBatch（put+delete 同一条日志记录）， KVStoreEngine 全部混合写路径统一（兑现真原子宣称，崩溃无新旧行并存） - WAL full 模式写入失败抛错（此前 console.warn 吞错 → 崩溃即丢且无感知） - MemoryEngine SET NULL 级联索引残留：复用 removeIndexEntries（消除虚假 UNIQUE_VIOLATION） - delete 级联两阶段：先全量 RESTRICT 预检（沿 CASCADE 链递归）再执行，无部分级联 （Memory/Aria 对齐） - BufferPool 驱逐同步清理 pages Map（EvictionManager onRemove 回调，内存预算真实生效） - MVCC commit 清理已提交版本（版本链仅作事务内 undo，消除行数据双份常驻） - LSM.flush 重复入链修复（入链即置空 immutable）+ frozenMemtables 可见性时序 - rollbackToSavepoint 重建受影响表二级索引（消除过期索引条目） 测试 1114 → 1126（71 套件）；行覆盖率 89.7%；版本 0.6.3
530 lines
18 KiB
TypeScript
530 lines
18 KiB
TypeScript
/**
|
||
* KVStoreEngine — 基于自研 KVStore 的磁盘存储引擎（替代 IndexedDBEngine / OPFSEngine）
|
||
* @module engine/kvstore_engine
|
||
*
|
||
* v0.6.0: 完全移除 IndexedDB 后的 disk 模式引擎。
|
||
*
|
||
* 架构：MemoryEngine（内存热路径 + 事务快照）+ KVStore（持久化 + 原子写）
|
||
* - 读：始终走内存（写路径同步落盘，重启从 KVStore 恢复）
|
||
* - 写：内存先行 + KVStore 增量持久化（insert 增量 putMany；update/delete 受影响行重写；
|
||
* 主键变更/级联场景整表 diff；全部原子）
|
||
* - 事务：内存快照 + commit 时受影响表原子 flush（writeBatch 单记录 = 真原子，
|
||
* 此前 IndexedDBEngine 依赖 IDB 事务，现在完全自研）
|
||
*
|
||
* 数据布局（KVStore keys）:
|
||
* `__schema` — JSON { tableName: TableSchema }
|
||
* `__meta:{key}` — 库内元数据（迁移版本等）
|
||
* `t:{table}:{pk}` — 行数据（JSON）
|
||
*/
|
||
|
||
import type { IStorageEngine } from './interface';
|
||
import type { QueryPlan, TableSchema } from '../constants';
|
||
import { DatabaseError } from '../constants';
|
||
import { MemoryEngine } from './memory';
|
||
import { KVStore } from './kvstore/index';
|
||
import { SharedMemoryBackend } from './kvstore/shared_memory_medium';
|
||
import type { IStorageBackend } from './aria/store/backend';
|
||
|
||
const SCHEMA_KEY = '__schema';
|
||
const ROW_PREFIX = 't:';
|
||
|
||
const enc = (s: string) => new TextEncoder().encode(s).buffer as ArrayBuffer;
|
||
const dec = (b: ArrayBuffer) => new TextDecoder().decode(b);
|
||
|
||
export class KVStoreEngine implements IStorageEngine {
|
||
readonly name = 'kv';
|
||
|
||
private kv!: KVStore;
|
||
private memory: MemoryEngine = new MemoryEngine();
|
||
private dbName = '';
|
||
private version = 1;
|
||
private opened = false;
|
||
|
||
/** 活跃事务标记 */
|
||
private txActive = false;
|
||
/** 事务中写过的表（commit 时只 flush 这些表） */
|
||
private txDirtyTables: Set<string> = new Set();
|
||
/** 事务中发生 schema 变更（DDL）—— commit 时持久化 schema */
|
||
private txSchemaChanged = false;
|
||
|
||
constructor(medium?: IStorageBackend, checkpointThreshold?: number) {
|
||
this.kv = new KVStore(medium, checkpointThreshold);
|
||
}
|
||
|
||
// ---- 行 key 编解码 ----
|
||
|
||
private rowKey(table: string, pk: string): string {
|
||
return `${ROW_PREFIX}${table}:${pk}`;
|
||
}
|
||
|
||
private rowPrefix(table: string): string {
|
||
return `${ROW_PREFIX}${table}:`;
|
||
}
|
||
|
||
// ---- 生命周期 ----
|
||
|
||
async open(dbName: string, version: number): Promise<void> {
|
||
if (this.opened) return;
|
||
this.dbName = dbName;
|
||
this.version = version;
|
||
await this.kv.open(dbName);
|
||
await this.memory.open(dbName, version);
|
||
|
||
// 恢复 schema
|
||
const schemaRaw = await this.kv.get(SCHEMA_KEY);
|
||
if (schemaRaw) {
|
||
try {
|
||
const schemas = JSON.parse(dec(schemaRaw)) as Record<string, TableSchema>;
|
||
for (const schema of Object.values(schemas)) {
|
||
await this.memory.createTable(schema);
|
||
}
|
||
} catch {
|
||
throw new DatabaseError('Corrupted schema in KVStore', 'KV_SCHEMA_ERROR');
|
||
}
|
||
}
|
||
|
||
// 恢复行数据 + 重建索引
|
||
const all = await this.kv.getAll();
|
||
for (const [key, value] of all) {
|
||
if (!key.startsWith(ROW_PREFIX)) continue;
|
||
const sep = key.indexOf(':', ROW_PREFIX.length);
|
||
if (sep < 0) continue;
|
||
const table = key.slice(ROW_PREFIX.length, sep);
|
||
if (!(await this.memory.hasTable(table))) continue;
|
||
try {
|
||
const row = JSON.parse(dec(value));
|
||
await this.memory.insert(table, [row]);
|
||
} catch {
|
||
// 单行损坏跳过（repair 可清理）
|
||
}
|
||
}
|
||
// 重建二级索引（schema 标记的索引列）
|
||
const tables = await this.memory.getTableNames();
|
||
for (const table of tables) {
|
||
const schema = await this.memory.getTableSchema(table);
|
||
if (!schema) continue;
|
||
for (const [col, colDef] of Object.entries(schema.columns)) {
|
||
if (colDef.index || colDef.unique) {
|
||
await this.memory.createIndex(table, col, colDef.unique);
|
||
}
|
||
}
|
||
}
|
||
|
||
this.opened = true;
|
||
}
|
||
|
||
async close(): Promise<void> {
|
||
if (!this.opened) return;
|
||
// 活跃事务先回滚
|
||
if (this.txActive) {
|
||
try { await this.rollbackTransaction(); } catch { /* ignore */ }
|
||
}
|
||
await this.kv.close();
|
||
await this.memory.close();
|
||
this.opened = false;
|
||
}
|
||
|
||
isOpen(): boolean {
|
||
return this.opened;
|
||
}
|
||
|
||
/**
|
||
* v0.6.0: 从 KVStore 重新加载全部数据到内存（多标签页同步重载用）。
|
||
* Hybrid 引擎的 reloadMemoryFromDisk 依赖磁盘引擎"读穿透"，
|
||
* KVStoreEngine 读内存 → 提供 reload 重新加载磁盘最新数据。
|
||
*/
|
||
async reload(): Promise<void> {
|
||
if (!this.opened) return;
|
||
// 1. KVStore 重新从介质加载（外部写入可见）
|
||
await this.kv.reload();
|
||
// 2. 内存缓存重载
|
||
await this.memory.close();
|
||
await this.memory.open(this.dbName, this.version);
|
||
this.opened = false;
|
||
await this.open(this.dbName, this.version);
|
||
}
|
||
|
||
/** v0.4.2-fix: 自愈 — 校验 KVStore 日志/快照完整性并重建内存 */
|
||
async repair(): Promise<void> {
|
||
this.ensureOpen();
|
||
await this.kv.repair();
|
||
await this.memory.close();
|
||
await this.memory.open(this.dbName, this.version);
|
||
// 重新恢复（复用 open 的恢复逻辑）
|
||
this.opened = false;
|
||
await this.open(this.dbName, this.version);
|
||
}
|
||
|
||
async clearAll(): Promise<void> {
|
||
this.ensureOpen();
|
||
await this.kv.clear();
|
||
await this.memory.clearAll();
|
||
}
|
||
|
||
async getMeta(key: string): Promise<string | null> {
|
||
const raw = await this.kv.get(`__meta:${key}`);
|
||
return raw ? dec(raw) : null;
|
||
}
|
||
|
||
async setMeta(key: string, value: string): Promise<void> {
|
||
await this.kv.put(`__meta:${key}`, enc(value));
|
||
}
|
||
|
||
// ---- 表管理 ----
|
||
|
||
async createTable(schema: TableSchema): Promise<void> {
|
||
this.ensureOpen();
|
||
await this.memory.createTable(schema);
|
||
if (this.txActive) {
|
||
this.txDirtyTables.add(schema.name);
|
||
this.txSchemaChanged = true;
|
||
return;
|
||
}
|
||
await this.persistSchema();
|
||
}
|
||
|
||
async dropTable(tableName: string): Promise<void> {
|
||
this.ensureOpen();
|
||
await this.memory.dropTable(tableName);
|
||
if (this.txActive) {
|
||
this.txDirtyTables.add(tableName);
|
||
this.txSchemaChanged = true;
|
||
return;
|
||
}
|
||
await this.persistSchema();
|
||
// 删除该表全部行（KV 中残留清理）
|
||
const diff = await this.collectTableDiff(tableName);
|
||
await this.kv.writeBatch(diff.puts, diff.deletes);
|
||
}
|
||
|
||
async hasTable(tableName: string): Promise<boolean> {
|
||
this.ensureOpen();
|
||
return this.memory.hasTable(tableName);
|
||
}
|
||
|
||
async getTableNames(): Promise<string[]> {
|
||
this.ensureOpen();
|
||
return this.memory.getTableNames();
|
||
}
|
||
|
||
async getTableSchema(tableName: string): Promise<TableSchema | null> {
|
||
this.ensureOpen();
|
||
return this.memory.getTableSchema(tableName);
|
||
}
|
||
|
||
async alterTable(
|
||
tableName: string,
|
||
action: 'ADD' | 'DROP',
|
||
column: import('../constants').ColumnDef & { name: string },
|
||
): Promise<void> {
|
||
this.ensureOpen();
|
||
await this.memory.alterTable(tableName, action, column);
|
||
if (this.txActive) {
|
||
this.txDirtyTables.add(tableName);
|
||
this.txSchemaChanged = true;
|
||
return;
|
||
}
|
||
await this.persistSchema();
|
||
if (action === 'DROP') {
|
||
// 重写存储行（移除该列）
|
||
const diff = await this.collectTableDiff(tableName);
|
||
await this.kv.writeBatch(diff.puts, diff.deletes);
|
||
}
|
||
}
|
||
|
||
// ---- CRUD ----
|
||
|
||
async insert(tableName: string, rows: Record<string, unknown>[]): Promise<string[]> {
|
||
this.ensureOpen();
|
||
const pks = await this.memory.insert(tableName, rows);
|
||
if (this.txActive) {
|
||
this.txDirtyTables.add(tableName);
|
||
return pks;
|
||
}
|
||
// 增量持久化（原子 putMany）
|
||
const schema = await this.memory.getTableSchema(tableName);
|
||
if (!schema) throw new DatabaseError(`Table "${tableName}" does not exist`, 'TABLE_NOT_FOUND');
|
||
const pkCol = this.getPK(schema);
|
||
const puts: Record<string, ArrayBuffer> = {};
|
||
rows.forEach((row, i) => {
|
||
puts[this.rowKey(tableName, String(pks[i] ?? row[pkCol]))] = enc(JSON.stringify(row));
|
||
});
|
||
await this.kv.putMany(puts);
|
||
return pks;
|
||
}
|
||
|
||
async find(tableName: string, query: QueryPlan): Promise<Record<string, unknown>[]> {
|
||
this.ensureOpen();
|
||
return this.memory.find(tableName, query);
|
||
}
|
||
|
||
async findStream(tableName: string, query: QueryPlan, onRow: (row: Record<string, unknown>) => void): Promise<number> {
|
||
this.ensureOpen();
|
||
return this.memory.findStream(tableName, query, onRow);
|
||
}
|
||
|
||
async update(
|
||
tableName: string,
|
||
query: QueryPlan,
|
||
updates: Record<string, unknown>,
|
||
): Promise<number> {
|
||
this.ensureOpen();
|
||
const schema = await this.memory.getTableSchema(tableName);
|
||
if (!schema) throw new DatabaseError(`Table "${tableName}" does not exist`, 'TABLE_NOT_FOUND');
|
||
const pkCol = this.getPK(schema);
|
||
const pkChanged = pkCol in updates;
|
||
|
||
// 收集受影响旧主键（内存匹配）
|
||
const affected = pkChanged ? [] : await this.collectMatchingPks(tableName, query);
|
||
const count = await this.memory.update(tableName, query, updates);
|
||
if (this.txActive) {
|
||
this.txDirtyTables.add(tableName);
|
||
return count;
|
||
}
|
||
|
||
const puts: Record<string, ArrayBuffer> = {};
|
||
const deletes: string[] = [];
|
||
if (pkChanged) {
|
||
// 主键变更：相关表整表 diff（罕见操作，可靠性优先）
|
||
for (const t of await this.affectedTables(tableName)) {
|
||
const diff = await this.collectTableDiff(t);
|
||
Object.assign(puts, diff.puts);
|
||
deletes.push(...diff.deletes);
|
||
}
|
||
} else {
|
||
// v0.6.2-fix(P0): 受影响主键经 String() 化后按 `where { [pkCol]: pk }` 回查内存行，
|
||
// 数值型主键（123 !== "123"）不命中 → 行被误判删除 → 重启丢数据。
|
||
// 改为单次全表扫描 + 受影响集合过滤（同时消除此前 O(N×M) 逐主键回查开销）。
|
||
const affectedSet = new Set(affected);
|
||
const allRows = await this.memory.find(tableName, { table: tableName });
|
||
for (const row of allRows) {
|
||
const pkStr = String(row[pkCol]);
|
||
if (affectedSet.has(pkStr)) {
|
||
puts[this.rowKey(tableName, pkStr)] = enc(JSON.stringify(row));
|
||
affectedSet.delete(pkStr);
|
||
}
|
||
}
|
||
// 剩余主键（内存中已不存在，如被级联移除）→ 删除对应 KV 行
|
||
for (const pk of affectedSet) {
|
||
deletes.push(this.rowKey(tableName, pk));
|
||
}
|
||
// 级联影响表（SET NULL/CASCADE 外键）整表 diff
|
||
for (const t of await this.affectedTables(tableName)) {
|
||
if (t === tableName) continue;
|
||
const diff = await this.collectTableDiff(t);
|
||
Object.assign(puts, diff.puts);
|
||
deletes.push(...diff.deletes);
|
||
}
|
||
}
|
||
// 单次原子写（一条日志记录 = 真原子，v0.6.1）
|
||
await this.kv.writeBatch(puts, deletes);
|
||
return count;
|
||
}
|
||
|
||
async delete(tableName: string, query: QueryPlan): Promise<number> {
|
||
this.ensureOpen();
|
||
// 收集受影响主键（内存匹配）
|
||
const pks = await this.collectMatchingPks(tableName, query);
|
||
const count = await this.memory.delete(tableName, query);
|
||
if (this.txActive) {
|
||
this.txDirtyTables.add(tableName);
|
||
return count;
|
||
}
|
||
const puts: Record<string, ArrayBuffer> = {};
|
||
const deletes = pks.map((pk) => this.rowKey(tableName, pk));
|
||
// 级联影响表整表 diff（合并到单次原子写，v0.6.1）
|
||
for (const t of await this.affectedTables(tableName)) {
|
||
if (t === tableName) continue;
|
||
const diff = await this.collectTableDiff(t);
|
||
Object.assign(puts, diff.puts);
|
||
deletes.push(...diff.deletes);
|
||
}
|
||
await this.kv.writeBatch(puts, deletes);
|
||
return count;
|
||
}
|
||
|
||
async count(tableName: string, query?: QueryPlan): Promise<number> {
|
||
this.ensureOpen();
|
||
return this.memory.count(tableName, query);
|
||
}
|
||
|
||
async clear(tableName: string): Promise<void> {
|
||
this.ensureOpen();
|
||
await this.memory.clear(tableName);
|
||
if (this.txActive) {
|
||
this.txDirtyTables.add(tableName);
|
||
return;
|
||
}
|
||
const diff = await this.collectTableDiff(tableName);
|
||
await this.kv.writeBatch(diff.puts, diff.deletes);
|
||
}
|
||
|
||
// ---- 动态索引 ----
|
||
|
||
async createIndex(tableName: string, column: string, unique?: boolean): Promise<void> {
|
||
this.ensureOpen();
|
||
await this.memory.createIndex(tableName, column, unique);
|
||
if (this.txActive) {
|
||
this.txDirtyTables.add(tableName);
|
||
this.txSchemaChanged = true;
|
||
return;
|
||
}
|
||
await this.persistSchema();
|
||
}
|
||
|
||
async dropIndex(tableName: string, column: string, indexName?: string): Promise<void> {
|
||
this.ensureOpen();
|
||
await this.memory.dropIndex(tableName, column, indexName);
|
||
if (this.txActive) {
|
||
this.txDirtyTables.add(tableName);
|
||
this.txSchemaChanged = true;
|
||
return;
|
||
}
|
||
await this.persistSchema();
|
||
}
|
||
|
||
// ---- 事务（原子 flush） ----
|
||
|
||
async beginTransaction(): Promise<void> {
|
||
this.ensureOpen();
|
||
await this.memory.beginTransaction();
|
||
this.txActive = true;
|
||
this.txDirtyTables = new Set();
|
||
this.txSchemaChanged = false;
|
||
}
|
||
|
||
async commitTransaction(): Promise<void> {
|
||
this.ensureOpen();
|
||
if (!this.txActive) throw new DatabaseError('No active transaction', 'TX_NONE');
|
||
// v0.6.1: 全部 dirty 表合并为单次原子 flush（一条日志记录 = 真原子，
|
||
// 多表事务中途崩溃/失败不会出现"部分表已提交"）
|
||
const puts: Record<string, ArrayBuffer> = {};
|
||
const deletes: string[] = [];
|
||
for (const table of this.txDirtyTables) {
|
||
if (await this.memory.hasTable(table)) {
|
||
const diff = await this.collectTableDiff(table);
|
||
Object.assign(puts, diff.puts);
|
||
deletes.push(...diff.deletes);
|
||
} else {
|
||
// 事务内 drop 的表：清理 KV 残留行
|
||
const all = await this.kv.getAll();
|
||
const prefix = this.rowPrefix(table);
|
||
for (const [key] of all) {
|
||
if (key.startsWith(prefix)) deletes.push(key);
|
||
}
|
||
}
|
||
}
|
||
await this.kv.writeBatch(puts, deletes);
|
||
// 事务内 DDL 的 schema 一并持久化
|
||
if (this.txSchemaChanged) {
|
||
await this.persistSchema();
|
||
}
|
||
await this.kv.checkpoint();
|
||
await this.memory.commitTransaction();
|
||
this.txActive = false;
|
||
this.txDirtyTables = new Set();
|
||
}
|
||
|
||
async rollbackTransaction(): Promise<void> {
|
||
this.ensureOpen();
|
||
if (!this.txActive) throw new DatabaseError('No active transaction', 'TX_NONE');
|
||
await this.memory.rollbackTransaction();
|
||
this.txActive = false;
|
||
this.txDirtyTables = new Set();
|
||
this.txSchemaChanged = false;
|
||
}
|
||
|
||
// ---- 内部 ----
|
||
|
||
private ensureOpen(): void {
|
||
if (!this.opened) throw new DatabaseError('Database not opened', 'DB_NOT_OPEN');
|
||
}
|
||
|
||
private getPK(schema: TableSchema): string {
|
||
for (const [name, col] of Object.entries(schema.columns)) {
|
||
if (col.primaryKey) return name;
|
||
}
|
||
return Object.keys(schema.columns)[0];
|
||
}
|
||
|
||
/** 收集匹配查询的内存行主键（持久化差异计算用） */
|
||
private async collectMatchingPks(tableName: string, query: QueryPlan): Promise<string[]> {
|
||
const schema = await this.memory.getTableSchema(tableName);
|
||
if (!schema) throw new DatabaseError(`Table "${tableName}" does not exist`, 'TABLE_NOT_FOUND');
|
||
const pkCol = this.getPK(schema);
|
||
const rows = await this.memory.find(tableName, query);
|
||
return rows.map((r) => String(r[pkCol]));
|
||
}
|
||
|
||
/**
|
||
* 计算外键级联影响的表集合（传递闭包：A 被 B 引用，B 被 C 引用 → {A, B, C}）。
|
||
* 级联操作（delete/update 主键）需要把这些表一并重写持久化。
|
||
*/
|
||
private async affectedTables(tableName: string): Promise<Set<string>> {
|
||
const set = new Set<string>([tableName]);
|
||
let changed = true;
|
||
while (changed) {
|
||
changed = false;
|
||
for (const table of await this.memory.getTableNames()) {
|
||
if (set.has(table)) continue;
|
||
const schema = await this.memory.getTableSchema(table);
|
||
if (!schema) continue;
|
||
for (const col of Object.values(schema.columns)) {
|
||
if (col.references) {
|
||
const ref = col.references.split('.')[0];
|
||
if (set.has(ref)) {
|
||
set.add(table);
|
||
changed = true;
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
return set;
|
||
}
|
||
|
||
/** 持久化 schema（全部表） */
|
||
private async persistSchema(): Promise<void> {
|
||
const schemas: Record<string, TableSchema> = {};
|
||
for (const table of await this.memory.getTableNames()) {
|
||
const schema = await this.memory.getTableSchema(table);
|
||
if (schema) schemas[table] = schema;
|
||
}
|
||
await this.kv.put(SCHEMA_KEY, enc(JSON.stringify(schemas)));
|
||
}
|
||
|
||
/**
|
||
* v0.6.1: 整表 diff 收集（不落盘）：内存行全部 put + KV 残留行删除。
|
||
* 调用方合并到单次原子 writeBatch（put+delete 同一条日志记录，多表操作真原子）。
|
||
*/
|
||
private async collectTableDiff(tableName: string): Promise<{ puts: Record<string, ArrayBuffer>; deletes: string[] }> {
|
||
const prefix = this.rowPrefix(tableName);
|
||
const puts: Record<string, ArrayBuffer> = {};
|
||
const deletes: string[] = [];
|
||
// 表已删除：仅收集 KV 残留行删除
|
||
const schema = await this.memory.getTableSchema(tableName);
|
||
if (schema) {
|
||
const pkCol = this.getPK(schema);
|
||
const rows = await this.memory.find(tableName, { table: tableName });
|
||
const current = new Set<string>();
|
||
for (const row of rows) {
|
||
const key = this.rowKey(tableName, String(row[pkCol]));
|
||
current.add(key);
|
||
puts[key] = enc(JSON.stringify(row));
|
||
}
|
||
// KV 残留行（内存中已不存在）
|
||
const all = await this.kv.getAll();
|
||
for (const [key] of all) {
|
||
if (key.startsWith(prefix) && !current.has(key)) deletes.push(key);
|
||
}
|
||
} else {
|
||
const all = await this.kv.getAll();
|
||
for (const [key] of all) {
|
||
if (key.startsWith(prefix)) deletes.push(key);
|
||
}
|
||
}
|
||
return { puts, deletes };
|
||
}
|
||
}
|