Files
MetonaSqlark/src/engine/memory.ts
T
thzxx 2a109ef933 feat(B-1): 统一行校验 choke point —— 消除三份分叉的校验实现(A12/A17)
背景(PLAN-v0.7.5.md 根因 1):
修复前有**三份**行校验实现,覆盖面各不相同:

  位置                                        类型 required PK非空 maxLength min/max 未知列
  engine/memory.ts(disk/hybrid 共用)          ✓     ✓       ✓      ✗        ✗     静默丢弃
  engine/aria/index.ts → checkFieldType         ✓     ✓       ✓      ✓        ✓     静默丢弃
  table/schema.ts                               ✓     ✓       ✓      ✓        ✓     静默丢弃

后果一(A12):同一份 schema、同一条 INSERT 是否报约束错误取决于引擎选择 ——
`CREATE TABLE t (name STRING(3))` + 插入 'abcdef' 在 Aria 抛错,在
memory/disk/hybrid 静默写入超长值。

后果二(A17):四个引擎对未知列一律静默丢弃。`INSERT INTO t (id, nope) VALUES
('1',2)` 报成功,随后 `SELECT nope` 报 COLUMN_NOT_FOUND —— 同一列名在写路径与
读路径得到**相反结论**。TABLE API 直通路径尤其明显(executor 按 schema 列序
构造行,nope 那个位置根本没有值,所以连"校验 stmt.columns"都拦不住)。

根治方式:
1. 新增 src/table/validation.ts —— 唯一校验定义 `compileValidator(schema)`,
   约束覆盖面取三者并集,并把**规范化**(default 填充、undefined 跳过、
   __proto__ 防污染)与校验放在同一处。
   三种载荷形态刻意分成三个显式入口,不合成带 options 的函数:
     - validateRow(row, knownColumns?)  INSERT 语义(default 生效、缺列合法)
     - validatePartial(row)             UPDATE 语义(只校验出现的列)
     - assertNoUnknownColumns           独立可复用的列名存在性检查
   混成一个函数会让"required 是否生效"取决于调用方参数,重新引入跨路径差异。
2. MemoryEngine / AriaEngine 的私有 validateRow 改为委托;schema.ts 的公开
   validateRow 同样委托(API 不变,实现只剩一份)。
3. 四个引擎新增 validatePayload(table, rows, mode)(IStorageEngine 契约),
   Executor 在**任何副作用之前**调用:多行批量整体判定,错误消息一次列出全部
   未知列与已知列清单。
4. executeInsert 显式校验 stmt.columns 全部存在(A17)。
5. UPDATE 的外键级联写入(applyUpdateCascade)从"直接赋值"改为过
   validatePartial —— 此前 CASCADE 把新主键写进引用列时绕过 maxLength/min/max,
   与 A12 属同一类"校验只在部分写入路径生效"。

连带修正(测试夹具本身不忠实,B-1 使其暴露):
  - tests/engine/aria-cache.test.ts 的 makeRows 无条件返回 {id,name,age},
    部分用例的表只有 {id,name} —— 多余列被静默丢弃所以"通过"。新增 rowsFor()
    按 schema 裁剪,让夹具忠实反映表结构(而不是放宽校验)。
  - tests/v073-fixes.test.ts "schema 外列不持久化" 改为断言写路径即拒绝,
    并保留"合法行落盘后不含额外列"的检查。

验证:
  - 新增 tests/v080-unified-validation.test.ts:8 项 × 4 引擎 + 9 项校验器
    单元契约,共 41 断言;
  - 全量 84 套件 / 1499 测试通过;typecheck(src+tests) 与 lint 零错误。
2026-09-14 23:38:58 +08:00

969 lines
42 KiB
TypeScript
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
/**
* metona-sqlark Memory Engine — 基于 Map 的内存存储引擎
* @module engine/memory
*/
import type { IStorageEngine } from './interface';
import type { QueryPlan, TableSchema, WhereCondition } from '../constants';
import { DatabaseError } from '../constants';
import { cloneRow } from './interface';
import { matchWhere, applyOrderBy, projectColumns, containsUnresolvedSubqueries } from '../query/where-matcher';
import { stripUndefinedUpdates } from '../table/schema';
import { compileValidator, type RowValidator } from '../table/validation';
export class MemoryEngine implements IStorageEngine {
readonly name = 'memory';
private tables: Map<string, Map<string, Record<string, unknown>>> = new Map();
private schemas: Map<string, TableSchema> = new Map();
private indexes: Map<string, Map<string, Map<unknown, Set<string>>>> = new Map();
private opened = false;
/** v0.4.2-fix: 库内元数据(迁移版本持久化用) */
private metaStore: Map<string, string> = new Map();
/**
* v0.7.4: 由 CREATE UNIQUE INDEX 添加的 unique 列(table:col)。
* 与建表 UNIQUE 约束区分:DROP INDEX 只允许解除索引来源的 unique,
* 建表约束需重建表(对齐 SQLite 语义,此前静默解除且不可恢复)。
*/
private uniqueIndexCols: Set<string> = new Set();
// ---- 事务快照 ----
private snapshot: {
tables: Map<string, Map<string, Record<string, unknown>>>;
schemas: Map<string, TableSchema>;
indexes: Map<string, Map<string, Map<unknown, Set<string>>>>;
} | null = null;
// ---- 生命周期 ----
async open(_dbName: string, _version: number): Promise<void> {
if (this.opened) {
// 幂等:已打开则忽略
return;
}
this.opened = true;
}
async close(): Promise<void> {
this.tables.clear(); this.schemas.clear(); this.indexes.clear(); this.metaStore.clear(); this.opened = false;
}
isOpen(): boolean { return this.opened; }
// ---- v0.4.2-fix: 自愈 / 重置 / 元数据 ----
/** 内存引擎无需修复(无持久化损坏概念) */
async repair(): Promise<void> { return; }
/** 清空全部数据与表结构 */
async clearAll(): Promise<void> {
const names = Array.from(this.schemas.keys());
for (const name of names) {
await this.dropTable(name);
}
this.metaStore.clear();
}
async getMeta(key: string): Promise<string | null> {
return this.metaStore.get(key) ?? null;
}
async setMeta(key: string, value: string): Promise<void> {
this.metaStore.set(key, value);
}
// ---- 表管理 ----
async createTable(schema: TableSchema): Promise<void> {
if (this.schemas.has(schema.name)) throw new DatabaseError(`Table "${schema.name}" already exists`, 'TABLE_EXISTS');
// v0.4.2-fix: 存储 schema 深拷贝 — 此前 Hybrid.reloadMemoryFromDisk 直接存入
// disk 引擎的 schema 引用,内存/磁盘引擎共享同一对象,任一引擎 ALTER 都会污染对方
const copy: TableSchema = { name: schema.name, columns: {} };
for (const [colName, colDef] of Object.entries(schema.columns)) {
copy.columns[colName] = { ...colDef };
}
this.schemas.set(schema.name, copy);
this.tables.set(schema.name, new Map());
const tableIndexes = new Map<string, Map<unknown, Set<string>>>();
for (const [colName, colDef] of Object.entries(copy.columns)) {
if (colDef.index || colDef.unique) tableIndexes.set(colName, new Map());
}
this.indexes.set(schema.name, tableIndexes);
}
async dropTable(tableName: string): Promise<void> {
this.ensureTable(tableName);
this.schemas.delete(tableName); this.tables.delete(tableName); this.indexes.delete(tableName);
// v0.7.4: 清理该表的 unique 索引来源标记(重建同名表不残留)
const prefix = `${tableName}:`;
for (const key of this.uniqueIndexCols) {
if (key.startsWith(prefix)) this.uniqueIndexCols.delete(key);
}
}
async hasTable(tableName: string): Promise<boolean> { return this.schemas.has(tableName); }
async getTableNames(): Promise<string[]> { return Array.from(this.schemas.keys()); }
async getTableSchema(tableName: string): Promise<TableSchema | null> { return this.schemas.get(tableName) ?? null; }
/**
* v0.4.2-fix: 引擎级 ALTER TABLE — 直接修改内存 schema 引用并清理行数据。
* (此前走 executor 通用路径,行为相同;统一到引擎层保证 Hybrid/IndexedDB 委托一致性)
*/
async alterTable(
tableName: string,
action: 'ADD' | 'DROP',
column: import('../constants').ColumnDef & { name: string },
): Promise<void> {
// v0.7.2: 事务内 DDL 显式拒绝(与 AriaEngine 对齐)。此前事务快照对 schema
// 是浅拷贝,alterTable 直接修改共享 columns 对象 → ROLLBACK 后结构变更残留
// (三引擎行为不一致:Aria 拒绝 / Memory、KVStore 静默残留)
if (this.snapshot) {
throw new DatabaseError(
`ALTER TABLE is not supported inside a transaction (MemoryEngine DDL is not transactional)`,
'NOT_SUPPORTED',
);
}
this.ensureTable(tableName);
const schema = this.schemas.get(tableName)!;
if (action === 'ADD') {
if (schema.columns[column.name]) {
throw new DatabaseError(`Column "${column.name}" already exists in table "${tableName}"`, 'COLUMN_EXISTS');
}
schema.columns[column.name] = column;
// v0.8.0 根治:ALTER ADD 必须建立二级索引桶。
//
// 此前只写 schema.columns 而不建桶,而唯一性预检完全依赖索引桶
// `tableIndexes.get(colName)` 缺失即整段跳过)—— 于是
// `ALTER TABLE t ADD COLUMN email STRING UNIQUE` 之后插入重复 email
// **不会报错**close/reopen 时 createTable 依 schema 建桶、回灌第 2 行
// 触发 UNIQUE_VIOLATION 而异常被引擎 open 路径吞掉 → **行静默消失**。
if (column.index || column.unique) {
if (!this.indexes.has(tableName)) this.indexes.set(tableName, new Map());
const tableIndexes = this.indexes.get(tableName)!;
if (!tableIndexes.has(column.name)) tableIndexes.set(column.name, new Map());
// 已存在行:先校验存量唯一性(重复则回滚本次 ALTER),再回填索引桶
const colIndex = tableIndexes.get(column.name)!;
const table = this.tables.get(tableName)!;
const seen = new Set<unknown>();
for (const [pk, row] of table) {
const value = row[column.name];
if (value === null || value === undefined) continue; // null 不受唯一约束
if (column.unique && seen.has(value)) {
tableIndexes.delete(column.name);
delete schema.columns[column.name];
throw new DatabaseError(
`Duplicate value "${String(value)}" for UNIQUE column "${column.name}" in table "${tableName}"`,
'UNIQUE_VIOLATION',
);
}
seen.add(value);
let pks = colIndex.get(value);
if (!pks) { pks = new Set(); colIndex.set(value, pks); }
pks.add(pk);
}
}
return;
}
if (!schema.columns[column.name]) {
throw new DatabaseError(`Column "${column.name}" does not exist in table "${tableName}"`, 'COLUMN_NOT_FOUND');
}
// v0.7.3: 被删列是索引列 → 同步清理索引 Map —— 此前残留旧索引:
// 查询已删列仍走旧索引(不含新行)→ 结果不完整(对齐 AriaEngine cleanupTableIndexes
if (schema.columns[column.name].index || schema.columns[column.name].unique) {
this.indexes.get(tableName)?.delete(column.name);
}
delete schema.columns[column.name];
// 清理已有行中该列的值(find 返回行引用,直接删除生效)
const table = this.tables.get(tableName)!;
for (const row of table.values()) {
if (column.name in row) delete row[column.name];
}
}
// ---- CRUD ----
async insert(tableName: string, rows: Record<string, unknown>[]): Promise<string[]> {
this.ensureTable(tableName);
const schema = this.schemas.get(tableName)!;
const table = this.tables.get(tableName)!;
const pkColumn = this.getPrimaryKey(schema);
const pks: string[] = [];
// v0.7.3: 语句级原子性 —— 两阶段(先全量预检,后执行)。
// 此前逐行"校验+写入":第 N 行主键重复/唯一冲突抛错时,前 N-1 行已提交
// (无事务下语句级部分提交,与 v0.7.2 修复的 UPDATE 同类问题)。
const validated: Record<string, unknown>[] = [];
const pkSet = new Set<string>();
const batchUnique: Map<string, Set<unknown>> = new Map();
// 阶段 1:全量预检(任何一行失败 → 整条语句不执行)
for (const row of rows) {
const validatedRow = this.validateRow(schema, row);
const pkValue = String(validatedRow[pkColumn]);
// 批内主键互查(内存表尚未反映本批写入)
if (table.has(pkValue) || pkSet.has(pkValue)) {
throw new DatabaseError(`Duplicate primary key "${pkValue}" in table "${tableName}"`, 'DUPLICATE_KEY');
}
pkSet.add(pkValue);
// v0.7.3: 批内唯一互查 + 索引查(此前两行同批写入同一唯一值时,
// 第一行已写入索引 → 第二行 checkUniqueness 抛错 → 第一行残留)
this.checkInsertUniqueness(schema, tableName, validatedRow, batchUnique);
validated.push(validatedRow);
}
// 阶段 2:执行(预检已通过,此阶段不再抛校验类错误)
for (const validatedRow of validated) {
const pkValue = String(validatedRow[pkColumn]);
table.set(pkValue, validatedRow);
this.updateIndexes(tableName, validatedRow, pkValue);
pks.push(pkValue);
}
return pks;
}
/** v0.7.3: 按主键取已验证行(KVStoreEngine 持久化 validated 行用,含 default/类型归一) */
getRow(tableName: string, pkValue: string): Record<string, unknown> | null {
// v0.8.0: 返回副本(调用方用于持久化,不得持有内部引用)
const table = this.tables.get(tableName);
if (!table) return null;
const row = table.get(pkValue);
return row ? cloneRow(row) : null;
}
async find(tableName: string, query: QueryPlan): Promise<Record<string, unknown>[]> {
this.ensureTable(tableName);
const table = this.tables.get(tableName)!;
let results = this.tryIndexLookup(tableName, table, query);
if (query.where && Object.keys(query.where).length > 0) {
results = results.filter((row) => matchWhere(row, query.where!));
}
if (query.orderBy && query.orderBy.length > 0) {
results = applyOrderBy(results, query.orderBy);
}
const offset = query.offset ?? 0;
const limit = query.limit ?? results.length;
results = results.slice(offset, offset + limit);
if (query.columns && query.columns.length > 0 && query.columns[0] !== '*') {
results = results.map((row) => projectColumns(row, query.columns!));
}
// v0.8.0: 返回副本 —— 此前直接交出内部行对象,调用方原地修改即改写存储
// 并让索引与行失配(该行从此查不出来)。见 engine/interface.ts 的行所有权约定。
return results.map((row) => cloneRow(row));
}
/** v0.4.0: 流式查询 — 逐行回调(单次迭代,不物化结果数组) */
async findStream(tableName: string, query: QueryPlan, onRow: (row: Record<string, unknown>) => void): Promise<number> {
this.ensureTable(tableName);
const table = this.tables.get(tableName)!;
const hasWhere = !!(query.where && Object.keys(query.where).length > 0);
const limit = query.limit ?? Infinity;
const offset = query.offset ?? 0;
const project = query.columns && query.columns.length > 0 && query.columns[0] !== '*'
? (row: Record<string, unknown>) => projectColumns(row, query.columns!)
: null;
let count = 0;
let skipped = 0;
for (const row of table.values()) {
if (hasWhere && !matchWhere(row, query.where!)) continue;
if (skipped < offset) { skipped++; continue; }
onRow(project ? project(row) : cloneRow(row));
count++;
if (count >= limit) break;
}
return count;
}
async update(tableName: string, query: QueryPlan, updates: Record<string, unknown>): Promise<number> {
this.ensureTable(tableName);
const schema = this.schemas.get(tableName)!;
const table = this.tables.get(tableName)!;
const pkCol = this.getPrimaryKey(schema);
// v0.7.2: undefined 值视为"不更新该列"(保留旧值),null 显式置空
const cleanUpdates = stripUndefinedUpdates(updates);
// v0.7.4: 防御 —— QueryBuilder 直通引擎不经 Executor 子查询解析,
// 未解析的 $subquery/$col/$exists 在 matchWhere 中恒 false → 静默 0 行
if (containsUnresolvedSubqueries(query.where)) {
throw new DatabaseError(
'Unresolved subqueries/column references in UPDATE WHERE (use db.query() to execute subqueries)',
'NOT_SUPPORTED',
);
}
// v0.7.4: 未知列显式报错 —— 此前 SET nonexistent = ... 被静默写入存储行
// validateRow 只遍历 schema 列,脏列残留在行内并随持久化落盘)
for (const col of Object.keys(cleanUpdates)) {
if (!schema.columns[col]) {
throw new DatabaseError(`Column "${col}" does not exist in table "${tableName}"`, 'COLUMN_NOT_FOUND');
}
}
// v0.7.2: 语句级原子性 — 两阶段(先全量预检,后执行)。
// 此前逐行"校验+写入":第 N 行唯一冲突/校验失败抛错时,前 N-1 行已写入
// → 无事务下语句级部分提交(数据半更新且调用方已收到错误)。
const planned: { pk: string; row: Record<string, unknown>; updated: Record<string, unknown>; newPk: string }[] = [];
const batchUnique: Map<string, Set<unknown>> = new Map();
/**
* v0.8.0 根治:批内新主键互查。
*
* 此前阶段 1 只用 `table.has(newPk)` 与**语句执行前的表**比对,看不到同一语句内
* 其它行即将写入的新主键。于是 `UPDATE t SET id = 'X'`(匹配 2 行)在阶段 2
* 逐行 `table.set(newPk, ...)` 相互覆盖 —— 返回 affected=2,表中却只剩 1 行
* (静默丢行)。INSERT 路径在 v0.7.3 已做批内 PK Set 互查,UPDATE 漏了。
*
* 保守拒绝策略:同一语句内两行改到同一新主键必然互相覆盖,直接报错。
* 注意这也会拒绝"两行互换主键"(A:x→y, B:y→x)这种最终状态合法的写法 ——
* 那属于需要基于最终状态判定的场景,宁可显式报错也不静默丢行
* (与既有的"唯一值交换更新保守拒绝"语义一致)。
*/
const batchNewPks = new Set<string>();
// 阶段 1:全量预检(任何一行失败 → 整条语句不执行)
for (const [pk, row] of table) {
if (query.where && Object.keys(query.where).length > 0 && !matchWhere(row, query.where)) continue;
const updated = { ...row, ...cleanUpdates };
this.validateRow(schema, updated);
this.checkUpdateUniqueness(schema, tableName, pk, updated, batchUnique);
const newPk = String(updated[pkCol]);
// v0.6.2-fix(P0): 主键变更撞已有主键 → 抛 DUPLICATE_KEY(此前静默覆盖丢数据)
if (newPk !== pk && table.has(newPk)) {
throw new DatabaseError(
`Duplicate primary key "${newPk}" in table "${tableName}" (cannot update key to existing value)`,
'DUPLICATE_KEY',
);
}
// v0.8.0: 批内互查 —— 同一语句内两行改到同一新主键 → 整体拒绝(不得静默覆盖)
if (newPk !== pk) {
if (batchNewPks.has(newPk)) {
throw new DatabaseError(
`Duplicate primary key "${newPk}" in table "${tableName}" (multiple rows in the same statement update to the same key)`,
'DUPLICATE_KEY',
);
}
batchNewPks.add(newPk);
}
planned.push({ pk, row, updated, newPk });
}
// 阶段 1b:主键变更 RESTRICT 预检(引用表依赖行检查,任何修改前)
for (const p of planned) {
if (p.newPk !== p.pk) this.checkUpdateRestrict(tableName, p.pk);
}
// 阶段 2:执行(预检已通过,此阶段不再抛校验类错误)
let count = 0;
for (const { pk, row, updated, newPk } of planned) {
// v0.3.3: 先移除旧值索引条目(修复 update 后唯一约束被绕过、按新值查索引丢行)
this.removeIndexEntries(tableName, row, pk);
// v0.4.2-fix: 主键变更 — 删除旧键 + 级联更新引用表 + 新键落表
if (newPk !== pk) {
await this.applyUpdateCascade(tableName, pk, newPk);
}
table.delete(pk);
table.set(newPk, updated);
this.updateIndexes(tableName, updated, newPk);
count++;
}
return count;
}
/**
* v0.7.3: 插入唯一性预检 —— 批内互查(本批前几行写入同一唯一值)
* + 索引查(表中已有行)。与 update 的 checkUpdateUniqueness 对称,
* 两阶段 insert 预检阶段调用(索引尚未反映本批写入)。
*/
private checkInsertUniqueness(
schema: TableSchema,
tableName: string,
row: Record<string, unknown>,
batchUnique: Map<string, Set<unknown>>,
): void {
const tableIndexes = this.indexes.get(tableName);
for (const [colName, colDef] of Object.entries(schema.columns)) {
if (!colDef.unique) continue;
const value = row[colName];
if (value === undefined || value === null) continue;
let seen = batchUnique.get(colName);
if (!seen) {
seen = new Set<unknown>();
batchUnique.set(colName, seen);
}
if (seen.has(value)) {
throw new DatabaseError(
`Unique constraint violation on column "${colName}" in table "${schema.name}"`,
'UNIQUE_VIOLATION',
);
}
seen.add(value);
if (!tableIndexes) continue;
const colIndex = tableIndexes.get(colName);
if (colIndex && colIndex.has(value)) {
throw new DatabaseError(
`Unique constraint violation on column "${colName}" in table "${schema.name}"`,
'UNIQUE_VIOLATION',
);
}
}
}
/**
* v0.7.3: 更新唯一性预检 — 批内互查(多条行更新到同一唯一值)+ 索引查
* (排除自身旧条目)。阶段 1 中索引尚未更新,批内互查避免"两行同时改到
* 同一新值"绕过唯一约束。
*/
private checkUpdateUniqueness(
schema: TableSchema,
tableName: string,
pk: string,
updated: Record<string, unknown>,
batchUnique: Map<string, Set<unknown>>,
): void {
const tableIndexes = this.indexes.get(tableName);
for (const [colName, colDef] of Object.entries(schema.columns)) {
if (!colDef.unique) continue;
const value = updated[colName];
if (value === undefined || value === null) continue;
let seen = batchUnique.get(colName);
if (!seen) {
seen = new Set<unknown>();
batchUnique.set(colName, seen);
}
if (seen.has(value)) {
throw new DatabaseError(
`Unique constraint violation on column "${colName}" in table "${schema.name}"`,
'UNIQUE_VIOLATION',
);
}
seen.add(value);
if (!tableIndexes) continue;
const colIndex = tableIndexes.get(colName);
if (colIndex && colIndex.has(value)) {
const pks = colIndex.get(value)!;
// 值未变(新值 = 旧值)且索引中只有自身 → 允许
if (!(pks.size === 1 && pks.has(pk))) {
throw new DatabaseError(
`Unique constraint violation on column "${colName}" in table "${schema.name}"`,
'UNIQUE_VIOLATION',
);
}
}
}
}
/**
* v0.7.2: ON UPDATE RESTRICT 预检 — 从 applyUpdateCascade 提取,
* 两阶段 update 在任何修改前调用(整体拒绝语义)。
*/
private checkUpdateRestrict(tableName: string, oldPk: string): void {
for (const [refTableName, refSchema] of this.schemas) {
if (refTableName === tableName) continue;
for (const [colName, colDef] of Object.entries(refSchema.columns)) {
if (!colDef.references || !colDef.onUpdate) continue;
const [refTable] = colDef.references.split('.');
if (refTable !== tableName) continue;
const refTableData = this.tables.get(refTableName);
if (!refTableData) continue;
let hasDependents = false;
for (const [, refRow] of refTableData) {
if (String(refRow[colName]) !== oldPk) continue;
hasDependents = true;
if (colDef.onUpdate === 'RESTRICT') {
throw new DatabaseError(
`Cannot update "${tableName}" key "${oldPk}": foreign key "${colName}" in "${refTableName}" has dependent rows`,
'FOREIGN_KEY_VIOLATION',
);
}
}
// v0.7.2: SET NULL 到 required 列违反约束 —— 与 RESTRICT 同样整体拒绝
// (此前级联直写 null 绕过 validateRowrequired 列被静默置空)
if (hasDependents && colDef.onUpdate === 'SET NULL' && colDef.required) {
throw new DatabaseError(
`Cannot update "${tableName}" key "${oldPk}": foreign key "${colName}" in "${refTableName}" is required (SET NULL violates constraint)`,
'FOREIGN_KEY_VIOLATION',
);
}
}
}
}
/**
* v0.4.2-fix: ON UPDATE 外键级联 — 被引用表主键变更时处理引用表:
* RESTRICT 抛错 / CASCADE 更新 FK 值 / SET NULL 置空。
* v0.7.3-perf: 删除冗余的阶段 1 RESTRICT 扫描 —— checkUpdateRestrict 已在
* 两阶段 update 预检(阶段 1b)覆盖 RESTRICT 与 SET NULL+required
* 此处任何修改前重复全表扫描纯属浪费。直接执行 CASCADE / SET NULL。
*/
private async applyUpdateCascade(tableName: string, oldPk: string, newPk: string): Promise<void> {
for (const [refTableName, refSchema] of this.schemas) {
if (refTableName === tableName) continue;
for (const [colName, colDef] of Object.entries(refSchema.columns)) {
if (!colDef.references || !colDef.onUpdate) continue;
const [refTable] = colDef.references.split('.');
if (refTable !== tableName) continue;
const refTableData = this.tables.get(refTableName);
if (!refTableData) continue;
if (colDef.onUpdate !== 'CASCADE' && colDef.onUpdate !== 'SET NULL') continue;
// v0.8.0(B-1):级联写入也必须过统一校验。
//
// 此前这里**直接赋值**绕过校验:`onUpdate: 'CASCADE'` 把新主键写入引用列时,
// 若该列有 maxLength / min / max 约束(新主键更长或超出范围),
// 约束被静默绕过 —— 与 A12 是同一类"校验只在部分写入路径生效"的问题。
// SET NULL 到 required/非空列的检查由 checkUpdateRestrict 在任何修改前完成,
// 此处再校验可同时覆盖 maxLength/min/max 这类"具体值相关"的约束。
const validator = this.rowValidator(refSchema);
for (const [refPk, refRow] of refTableData) {
if (String(refRow[colName]) !== oldPk) continue;
const nextValue = colDef.onUpdate === 'CASCADE' ? newPk : null;
const { values } = validator.validatePartial({ [colName]: nextValue });
this.removeIndexEntries(refTableName, refRow, refPk);
refRow[colName] = values[colName];
this.updateIndexes(refTableName, refRow, refPk);
}
}
}
}
async delete(tableName: string, query: QueryPlan): Promise<number> {
this.ensureTable(tableName);
// v0.7.4: 防御 —— QueryBuilder 直通引擎不经 Executor 子查询解析,
// 未解析的 $subquery/$col/$exists 在 matchWhere 中恒 false → 静默 0 行
if (containsUnresolvedSubqueries(query.where)) {
throw new DatabaseError(
'Unresolved subqueries/column references in DELETE WHERE (use db.query() to execute subqueries)',
'NOT_SUPPORTED',
);
}
const table = this.tables.get(tableName)!;
const toDelete: { pk: string; row: Record<string, unknown> }[] = [];
for (const [pk, row] of table) {
if (!query.where || Object.keys(query.where).length === 0 || matchWhere(row, query.where)) {
toDelete.push({ pk, row });
}
}
// v0.6.3-fix: 级联两阶段 —— 先对全部待删行做 RESTRICT 预检(沿 CASCADE 链递归),
// 任何一行违规则整体拒绝。此前逐行执行:第 N 行 RESTRICT 抛错时,前 N-1 行的
// 级联子行已被删除、父行未删 → 无事务下部分级联(数据不一致)
//
// v0.7.3-fix: 索引清理移到预检之后 —— 此前 removeIndexEntries 在收集阶段执行,
// RESTRICT 预检抛错时行未删但索引条目已删 → 唯一约束失效、索引查询丢行
const restrictVisited = new Set<string>();
for (const { pk } of toDelete) {
this.checkCascadeRestrict(tableName, pk, restrictVisited);
}
// 预检通过:清理索引 + 级联删除(此阶段不再抛校验类错误)
let cascadeCount = 0;
for (const { pk, row } of toDelete) {
// v0.3.3: 删除行前清理其索引条目(修复删除后索引残留)
this.removeIndexEntries(tableName, row, pk);
cascadeCount += await this.cascadeDelete(tableName, pk, row);
}
for (const { pk } of toDelete) table.delete(pk);
return toDelete.length + cascadeCount;
}
/**
* v0.6.3: RESTRICT 预检(delete 级联两阶段之一)。
* 递归沿 CASCADE 链检查引用表:RESTRICT 引用存在依赖行则抛 FOREIGN_KEY_VIOLATION。
*/
private checkCascadeRestrict(tableName: string, pkValue: string, visited: Set<string>): void {
const visitKey = `${tableName}:${pkValue}`;
if (visited.has(visitKey)) return;
visited.add(visitKey);
for (const [refTableName, refSchema] of this.schemas) {
if (refTableName === tableName) continue;
for (const [colName, colDef] of Object.entries(refSchema.columns)) {
if (!colDef.references || !colDef.onDelete) continue;
const [refTable] = colDef.references.split('.');
if (refTable !== tableName) continue;
const refTableData = this.tables.get(refTableName);
if (!refTableData) continue;
const refPks: string[] = [];
for (const [refPk, refRow] of refTableData) {
if (String(refRow[colName]) === pkValue) refPks.push(refPk);
}
if (colDef.onDelete === 'RESTRICT' && refPks.length > 0) {
throw new DatabaseError(
`Cannot delete from "${tableName}": foreign key "${colName}" in "${refTableName}" has dependent rows`,
'FOREIGN_KEY_VIOLATION',
);
}
// v0.7.2: SET NULL 到 required 列违反约束 —— 预检阶段整体拒绝
if (colDef.onDelete === 'SET NULL' && colDef.required && refPks.length > 0) {
throw new DatabaseError(
`Cannot delete from "${tableName}": foreign key "${colName}" in "${refTableName}" is required (SET NULL violates constraint)`,
'FOREIGN_KEY_VIOLATION',
);
}
if (colDef.onDelete === 'CASCADE') {
for (const refPk of refPks) {
this.checkCascadeRestrict(refTableName, refPk, visited);
}
}
}
}
}
async count(tableName: string, query?: QueryPlan): Promise<number> {
this.ensureTable(tableName);
const table = this.tables.get(tableName)!;
if (!query?.where || Object.keys(query.where).length === 0) return table.size;
let result = 0;
for (const row of table.values()) { if (matchWhere(row, query.where)) result++; }
return result;
}
async clear(tableName: string): Promise<void> {
this.ensureTable(tableName);
this.tables.get(tableName)!.clear();
const tableIndexes = this.indexes.get(tableName);
if (tableIndexes) for (const colIndex of tableIndexes.values()) colIndex.clear();
}
// ---- 动态索引(v0.3.0 ----
async createIndex(tableName: string, column: string, unique?: boolean): Promise<void> {
// v0.7.2: 事务内修改列级标志(colDef.index/unique)会写入共享列对象,
// 事务快照无法回滚 → 与 alterTable 同样显式拒绝
if (this.snapshot) {
throw new DatabaseError(
`CREATE INDEX is not supported inside a transaction (MemoryEngine DDL is not transactional)`,
'NOT_SUPPORTED',
);
}
this.ensureTable(tableName);
const schema = this.schemas.get(tableName)!;
const colDef = schema.columns[column];
if (!colDef) throw new DatabaseError(`Column "${column}" does not exist in table "${tableName}"`, 'COLUMN_NOT_FOUND');
if (colDef.index || colDef.unique) return; // 已存在
const tableIndexes = this.indexes.get(tableName)!;
if (!tableIndexes.has(column)) tableIndexes.set(column, new Map());
const colIndex = tableIndexes.get(column)!;
const table = this.tables.get(tableName)!;
try {
for (const [pk, row] of table) {
const value = row[column];
if (value !== undefined && value !== null) {
// v0.7.3: UNIQUE 索引回填校验存量唯一性 —— 此前重复数据静默建索引
// (SQLite 语义应报错),且此后该列唯一约束永远无法满足
if (unique && colIndex.has(value)) {
throw new DatabaseError(
`Unique index on column "${column}" in table "${tableName}" cannot be created: duplicate value "${String(value)}"`,
'UNIQUE_VIOLATION',
);
}
if (!colIndex.has(value)) colIndex.set(value, new Set());
colIndex.get(value)!.add(pk);
}
}
} catch (error) {
// 回填失败(唯一冲突):清理半初始化索引,标志未落,保持原子语义
tableIndexes.delete(column);
throw error;
}
colDef.index = true;
if (unique) {
colDef.unique = true;
// v0.7.4: 记录唯一约束来源(DROP INDEX 时可解除;建表约束不可)
this.uniqueIndexCols.add(`${tableName}:${column}`);
}
}
async dropIndex(tableName: string, column: string, _indexName?: string): Promise<void> {
// v0.7.2: 同 createIndex —— 列级标志修改无法通过事务快照回滚,显式拒绝
if (this.snapshot) {
throw new DatabaseError(
`DROP INDEX is not supported inside a transaction (MemoryEngine DDL is not transactional)`,
'NOT_SUPPORTED',
);
}
this.ensureTable(tableName);
const schema = this.schemas.get(tableName)!;
const colDef = schema.columns[column];
if (!colDef) throw new DatabaseError(`Column "${column}" does not exist in table "${tableName}"`, 'COLUMN_NOT_FOUND');
// v0.4.1: DROP 不存在的索引应报错(此前静默成功)
if (!colDef.index && !colDef.unique) {
throw new DatabaseError(`Index on column "${column}" does not exist in table "${tableName}"`, 'INDEX_NOT_FOUND');
}
// v0.7.4: 建表 UNIQUE 约束不可通过 DROP INDEX 解除 —— 此前 colDef.unique = false
// 静默解除约束(后续唯一性检查失效、重复数据入库)。对齐 SQLite 语义:
// 约束随建表存在,解除需重建表;仅 CREATE UNIQUE INDEX 添加的约束可随索引删除。
const uniqueKey = `${tableName}:${column}`;
if (colDef.unique && !this.uniqueIndexCols.has(uniqueKey)) {
throw new DatabaseError(
`Cannot drop index on column "${column}" in table "${tableName}": ` +
'UNIQUE constraint defined at table creation must be removed by recreating the table',
'NOT_SUPPORTED',
);
}
colDef.index = false;
colDef.unique = false;
this.uniqueIndexCols.delete(uniqueKey);
const tableIndexes = this.indexes.get(tableName);
if (tableIndexes) tableIndexes.delete(column);
}
// ---- 事务 ----
async beginTransaction(): Promise<void> {
if (this.snapshot) throw new DatabaseError('Transaction already in progress', 'TX_ACTIVE');
this.snapshot = {
tables: this.deepCloneMapMap(this.tables),
schemas: new Map(this.schemas),
indexes: this.deepCloneIndexes(this.indexes),
};
}
async commitTransaction(): Promise<void> {
if (!this.snapshot) throw new DatabaseError('No active transaction', 'TX_NONE');
this.snapshot = null;
}
async rollbackTransaction(): Promise<void> {
if (!this.snapshot) throw new DatabaseError('No active transaction', 'TX_NONE');
this.tables = this.snapshot.tables;
this.schemas = this.snapshot.schemas;
this.indexes = this.snapshot.indexes;
this.snapshot = null;
}
// ---- 事务快照辅助 ----
private deepCloneMapMap(source: Map<string, Map<string, Record<string, unknown>>>): Map<string, Map<string, Record<string, unknown>>> {
const clone = new Map<string, Map<string, Record<string, unknown>>>();
for (const [k, v] of source) {
const innerClone = new Map<string, Record<string, unknown>>();
for (const [ik, iv] of v) innerClone.set(ik, { ...iv });
clone.set(k, innerClone);
}
return clone;
}
private deepCloneIndexes(source: Map<string, Map<string, Map<unknown, Set<string>>>>): Map<string, Map<string, Map<unknown, Set<string>>>> {
const clone = new Map<string, Map<string, Map<unknown, Set<string>>>>();
for (const [tableName, tableIndexes] of source) {
const tableClone = new Map<string, Map<unknown, Set<string>>>();
for (const [col, colIndex] of tableIndexes) {
const colClone = new Map<unknown, Set<string>>();
for (const [val, pkSet] of colIndex) colClone.set(val, new Set(pkSet));
tableClone.set(col, colClone);
}
clone.set(tableName, tableClone);
}
return clone;
}
// ---- 内部辅助 ----
private ensureTable(tableName: string): void {
if (!this.tables.has(tableName)) throw new DatabaseError(`Table "${tableName}" does not exist`, 'TABLE_NOT_FOUND');
}
private getPrimaryKey(schema: TableSchema): string {
for (const [name, col] of Object.entries(schema.columns)) { if (col.primaryKey) return name; }
return Object.keys(schema.columns)[0];
}
private validateRow(schema: TableSchema, row: Record<string, unknown>): Record<string, unknown> {
// v0.8.0B-1):委托给**唯一**的行校验实现(table/validation.ts)。
//
// 此前这里是第三份独立实现:只做类型检查,**没有** maxLength / min / max
// 约束(Aria 有)—— 于是同一份 schema、同一条 INSERT 是否报错取决于引擎
//(缺陷 A12)。同时它对未知列静默丢弃(A17)。
return this.rowValidator(schema).validateRow(row);
}
/**
* 取该 schema 的行校验器(每次调用重新编译)。
*
* 不缓存在引擎字段上:`alterTable` 会原地修改 schema 对象,
* 长期缓存会继续用过期列定义("加了列却仍被当未知列"这类难查问题)。
* 编译本身只是 `Object.entries` + Set 构造,相对一次 INSERT 的索引维护可忽略。
*/
private rowValidator(schema: TableSchema): RowValidator {
return compileValidator(schema);
}
/**
* v0.8.0B-1):写入前置校验(见 `IStorageEngine.validatePayload` 契约)。
*
* 引擎在 `insert` / `update` 内部**同样**会校验 —— 本方法只是让 Executor 与
* QueryBuilder 能在"开始写入之前"拿到同一套判定结果,从而:
* - 多行 INSERT 的预检发生在任何副作用之前(错误信息带列名清单);
* - 直通路径与 SQL 路径不可能给出不同结论(同一个 `compileValidator`)。
*/
async validatePayload(
tableName: string,
rows: Record<string, unknown>[],
mode: 'insert' | 'update' = 'insert',
): Promise<void> {
this.ensureTable(tableName);
const schema = this.schemas.get(tableName)!;
const validator = this.rowValidator(schema);
for (const row of rows) {
if (mode === 'update') validator.validatePartial(stripUndefinedUpdates(row));
else validator.validateRow(row);
}
}
/** 索引查找 */
private tryIndexLookup(
tableName: string, table: Map<string, Record<string, unknown>>, query: QueryPlan,
): Record<string, unknown>[] {
const tableIndexes = this.indexes.get(tableName);
if (!tableIndexes || !query.where) return Array.from(table.values());
// v0.7.3: 递归展开 $and 中的等值条件 —— 此前仅顶层键,
// `WHERE a AND b`(解析为顶层 $and)永远全表扫描,索引形同虚设。
// $or/$not 语义不适用单索引下推,保守跳过。命中索引后 find 仍以
// 全条件 matchWhere 过滤(子集语义安全)。
const flat: [string, unknown][] = [];
const collect = (w: WhereCondition): void => {
for (const [k, v] of Object.entries(w)) {
if (k === '$and') {
for (const sub of (v as WhereCondition[])) collect(sub);
continue;
}
if (k === '$or' || k === '$not') continue;
flat.push([k, v]);
}
};
collect(query.where);
for (const [col, condition] of flat) {
// v0.4.1: 支持 { $eq: value } 形式(SQL 解析器生成的等值条件)走索引
let targetValue: unknown;
if (typeof condition !== 'object' || condition === null) {
targetValue = condition;
} else if ('$eq' in (condition as Record<string, unknown>) && Object.keys(condition as Record<string, unknown>).length === 1) {
targetValue = (condition as Record<string, unknown>).$eq;
} else {
continue;
}
// v0.7.3: null/undefined 条件不走索引 —— 索引不含 null 条目,
// colIndex.get(null) 恒 undefined → return [] 短路全表扫描 → 索引列
// IS NULL 恒空(对齐 AriaEngine v0.6.2 修复)
if (targetValue === null || targetValue === undefined) continue;
// v0.8.0 根治(与 AriaEngine 同步):**非原始值**(对象/数组)一律不走索引。
//
// 索引键只存原始值,因此 colIndex.get({...}) 恒 undefined → 下面 `return []`
// 会短路全表扫描 → 结果静默为空。真实触发场景正是"未解析的操作数":
// { $eq: { $col: 'y' } } ← 列对列比较(t.x = t.y)
// { $in: { $subquery: ... } } ← 关联 IN 子查询
// 它们本该由 executor 逐行求值,却先在这里被索引路径吞成空集。
if (typeof targetValue === 'object') continue;
const colIndex = tableIndexes.get(col);
if (colIndex) {
const pks = colIndex.get(targetValue);
if (pks) {
const result: Record<string, unknown>[] = [];
for (const pk of pks) { const r = table.get(pk); if (r) result.push(r); }
return result;
}
return [];
}
}
return Array.from(table.values());
}
/** 更新索引 */
private updateIndexes(tableName: string, row: Record<string, unknown>, pk: string): void {
const tableIndexes = this.indexes.get(tableName);
if (!tableIndexes) return;
for (const [colName, colIndex] of tableIndexes) {
const value = row[colName];
if (value !== undefined && value !== null) {
if (!colIndex.has(value)) colIndex.set(value, new Set());
colIndex.get(value)!.add(pk);
}
}
}
/** v0.3.3: 从所有索引中移除一行的条目(update/delete 前调用,修复索引过期/残留) */
private removeIndexEntries(tableName: string, row: Record<string, unknown>, pk: string): void {
const tableIndexes = this.indexes.get(tableName);
if (!tableIndexes) return;
for (const [colName, colIndex] of tableIndexes) {
const value = row[colName];
if (value !== undefined && value !== null) {
const pks = colIndex.get(value);
if (pks) {
pks.delete(pk);
if (pks.size === 0) colIndex.delete(value);
}
}
}
}
// ---- 外键级联 ----
/**
* 级联删除:查找引用 tableName.pkValue 的所有表的行并删除。
* v0.6.1-fix: 环路保护(A→B→A 级联环不再无限递归栈溢出,AriaEngine 同语义)。
* @returns 级联删除的行数
*/
private async cascadeDelete(
tableName: string,
pkValue: string,
_row: Record<string, unknown>,
visited: Set<string> = new Set(),
): Promise<number> {
const visitKey = `${tableName}:${pkValue}`;
if (visited.has(visitKey)) return 0;
visited.add(visitKey);
let totalCascade = 0;
for (const [refTableName, refSchema] of this.schemas) {
if (refTableName === tableName) continue;
for (const [colName, colDef] of Object.entries(refSchema.columns)) {
if (!colDef.references || !colDef.onDelete) continue;
const [refTable] = colDef.references.split('.');
if (refTable !== tableName) continue;
const refTableData = this.tables.get(refTableName);
if (!refTableData) continue;
// 查找所有引用此主键的行
const toDelete: string[] = [];
for (const [refPk, refRow] of refTableData) {
if (String(refRow[colName]) === pkValue) {
toDelete.push(refPk);
}
}
// RESTRICT: 存在引用行时禁止删除
if (colDef.onDelete === 'RESTRICT' && toDelete.length > 0) {
throw new DatabaseError(
`Cannot delete from "${tableName}": foreign key "${colName}" in "${refTableName}" has dependent rows`,
'FOREIGN_KEY_VIOLATION',
);
}
if (colDef.onDelete === 'CASCADE') {
// 递归级联
for (const refPk of toDelete) {
const refRow = refTableData.get(refPk);
if (refRow) {
// v0.3.3: 级联删除前清理索引条目
this.removeIndexEntries(refTableName, refRow, refPk);
totalCascade += await this.cascadeDelete(refTableName, refPk, refRow, visited);
}
refTableData.delete(refPk);
totalCascade++;
}
} else if (colDef.onDelete === 'SET NULL') {
for (const refPk of toDelete) {
const refRow = refTableData.get(refPk);
if (refRow) {
// v0.6.3-fix: 复用 removeIndexEntries 清理旧值索引 —— 此前手动
// `pks.get(v)?.delete(pk)` 后遗留空 Set → checkUniqueness 对旧值
// 永久误报 UNIQUE_VIOLATION(外键列带 unique 约束时)
this.removeIndexEntries(refTableName, refRow, refPk);
refRow[colName] = null;
this.updateIndexes(refTableName, refRow, refPk);
}
}
}
}
}
return totalCascade;
}
}