fix: v0.7.2 语句级原子性 + 事务 DDL 拒绝 + 约束/绑定硬化 — 6 项修复 + 43 回归 + CI 重型套件串行
- UPDATE 语句级部分提交（P1，四引擎）：两阶段全量预检后执行，批内唯一互查， 任何一行失败整句不执行（aria 场景 WAL 与内存不再错位） - 事务内 ALTER/CREATE INDEX/DROP INDEX 残留（P1）：Memory/KVStore 显式拒绝 （对齐 Aria），createTable/dropTable 保持可回滚 - SET NULL 级联绕过 required 约束（P1）：预检阶段整体拒绝 FOREIGN_KEY_VIOLATION - bindParameters 注释误判（P2）：行注释/块注释中的 ? 与引号不再参与绑定 - 未闭合字符串静默接受 → lexer 抛 PARSE_ERROR；未知 where 操作符抛 QUERY_ERROR - UPDATE undefined 覆盖列值 → 语义化为不更新（null 仍置空） - Hybrid 写穿透非原子（P1）：磁盘失败自动重载内存对齐磁盘再抛原错误 - CI：Run tests 拆常规并行 + 重型串行（runInBand），重型测试超时余量提升， 性能护栏 kv 120→240s / opfs 150→300s（仍拦截悬崖回归） - 测试 1155 → 1198（74 套件），覆盖率 89.82% 保持
This commit is contained in:
thzxx
+141
-57
@@ -9,7 +9,7 @@ import type { IStorageEngine } from '../interface';
|
||||
import type { QueryPlan, TableSchema, ColumnDef } from '../../constants';
|
||||
import { DatabaseError } from '../../constants';
|
||||
import { matchWhere, applyOrderBy, projectColumns } from '../../query/where-matcher';
|
||||
import { checkFieldType } from '../../table/schema';
|
||||
import { checkFieldType, stripUndefinedUpdates } from '../../table/schema';
|
||||
|
||||
import type { AriaEngineConfig, SSTableMeta } from './types';
|
||||
import { DEFAULT_ARIA_CONFIG } from './types';
|
||||
@@ -695,16 +695,18 @@ export class AriaEngine implements IStorageEngine {
|
||||
const walRecords: Omit<import('./types').WALRecord, 'lsn' | 'checksum'>[] = [];
|
||||
// v0.4.2-fix: ON UPDATE 级联环路保护
|
||||
const visited = new Set<string>();
|
||||
// v0.7.2: undefined 值视为"不更新该列"（保留旧值），null 显式置空
|
||||
const cleanUpdates = stripUndefinedUpdates(updates);
|
||||
|
||||
// v0.6.2: 唯一约束 — 批量预加载本批更新涉及的唯一列索引范围（一次 drainChain）
|
||||
const uniqueCols = this.uniqueColumns(tableName, schema);
|
||||
for (const colName of uniqueCols) {
|
||||
const idxLsm = this.secondaryIndexes.get(`${tableName}:idx:${colName}`)!;
|
||||
const ranges: [string, string][] = [];
|
||||
if (updates[colName] !== undefined && updates[colName] !== null) {
|
||||
const p = `${String(updates[colName])}:`;
|
||||
if (cleanUpdates[colName] !== undefined && cleanUpdates[colName] !== null) {
|
||||
const p = `${String(cleanUpdates[colName])}:`;
|
||||
ranges.push([p, `${p}\uffff`]);
|
||||
} else if (!(colName in updates)) {
|
||||
} else if (!(colName in cleanUpdates)) {
|
||||
for (const row of rows) {
|
||||
const val = row[colName];
|
||||
if (val === undefined || val === null) continue;
|
||||
@@ -715,77 +717,95 @@ export class AriaEngine implements IStorageEngine {
|
||||
await idxLsm.prefetchPrefixRanges(ranges);
|
||||
}
|
||||
|
||||
// v0.7.2: 语句级原子性 — 两阶段（先全量预检，后执行）。
|
||||
// 此前逐行"校验+写入"：第 N 行唯一冲突/校验失败抛错时，前 N-1 行已写入
|
||||
// 且其 WAL 记录随 appendBatch 一起丢失 → 内存已改、WAL 无记录、调用方已收到错误
|
||||
// （无事务下语句级部分提交 + 崩溃后进一步不一致）。
|
||||
const planned: { row: Record<string, unknown>; pk: string; key: string; updated: Record<string, unknown>; newPk: string; pkChanged: boolean }[] = [];
|
||||
const batchUnique: Map<string, Set<unknown>> = new Map();
|
||||
|
||||
// 阶段 1：全量预检（任何一行失败 → 整条语句不执行）
|
||||
for (const row of rows) {
|
||||
const pkCol = this.tablePKs.get(tableName)!;
|
||||
const key = `${tableName}:${row[pkCol]}`;
|
||||
if (query.where && Object.keys(query.where).length > 0 && !matchWhere(row, query.where)) continue;
|
||||
|
||||
if (!query.where || Object.keys(query.where).length === 0 || matchWhere(row, query.where)) {
|
||||
const updated = { ...row, ...updates };
|
||||
this.validateRow(schema, updated);
|
||||
const updated = { ...row, ...cleanUpdates };
|
||||
this.validateRow(schema, updated);
|
||||
|
||||
// v0.6.2: 唯一约束检查（排除自身旧索引条目：主键变更时旧条目仍以旧键存在）
|
||||
this.checkUniqueSync(tableName, uniqueCols, updated, String(row[pkCol]));
|
||||
// 批内唯一互查（索引尚未更新，两行同时改到同一新值需要互查兜底）
|
||||
this.checkBatchUnique(tableName, uniqueCols, updated, batchUnique);
|
||||
// v0.6.2: 唯一约束检查（排除自身旧索引条目：主键变更时旧条目仍以旧键存在）
|
||||
this.checkUniqueSync(tableName, uniqueCols, updated, String(row[pkCol]));
|
||||
|
||||
// v0.4.2-fix: 支持更新主键 — 删除旧键 + 落新键 + WAL 两条记录
|
||||
const newPk = String(updated[pkCol]);
|
||||
const pkChanged = newPk !== String(row[pkCol]);
|
||||
// v0.4.2-fix: 支持更新主键 — 删除旧键 + 落新键 + WAL 两条记录
|
||||
const newPk = String(updated[pkCol]);
|
||||
const pkChanged = newPk !== String(row[pkCol]);
|
||||
|
||||
// v0.6.2-fix(P0): 主键变更撞已有主键 → 抛 DUPLICATE_KEY
|
||||
// （此前静默覆盖另一行丢数据；与 MemoryEngine 对齐）
|
||||
if (pkChanged) {
|
||||
const newKey = `${tableName}:${newPk}`;
|
||||
const existing = this.currentTxnId
|
||||
? (this.txnSnapshot?.get(newKey) ?? this.lsm.get(newKey))
|
||||
: this.lsm.get(newKey);
|
||||
if (existing && !(existing as unknown as Record<string, unknown>).__txn_deleted) {
|
||||
throw new DatabaseError(
|
||||
`Duplicate primary key "${newPk}" in table "${tableName}" (cannot update key to existing value)`,
|
||||
'DUPLICATE_KEY',
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
if (pkChanged) {
|
||||
// ON UPDATE 外键级联（RESTRICT 抛错 / CASCADE / SET NULL）
|
||||
await this.applyForeignKeyUpdateRules(
|
||||
tableName, String(row[pkCol]), newPk, walRecords, visited,
|
||||
// v0.6.2-fix(P0): 主键变更撞已有主键 → 抛 DUPLICATE_KEY
|
||||
// （此前静默覆盖另一行丢数据；与 MemoryEngine 对齐）
|
||||
if (pkChanged) {
|
||||
const newKey = `${tableName}:${newPk}`;
|
||||
const existing = this.currentTxnId
|
||||
? (this.txnSnapshot?.get(newKey) ?? this.lsm.get(newKey))
|
||||
: this.lsm.get(newKey);
|
||||
if (existing && !(existing as unknown as Record<string, unknown>).__txn_deleted) {
|
||||
throw new DatabaseError(
|
||||
`Duplicate primary key "${newPk}" in table "${tableName}" (cannot update key to existing value)`,
|
||||
'DUPLICATE_KEY',
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
if (this.currentTxnId && this.txnSnapshot) {
|
||||
if (pkChanged) {
|
||||
this.txnSnapshot.set(key, { __txn_deleted: true } as unknown as Record<string, unknown>);
|
||||
this.mvcc.deleteVersion(tableName, String(row[pkCol]), this.currentTxnId);
|
||||
}
|
||||
this.txnSnapshot.set(`${tableName}:${newPk}`, updated);
|
||||
this.mvcc.writeVersion(tableName, newPk, updated, this.currentTxnId);
|
||||
} else {
|
||||
if (pkChanged) this.lsm.delete(key);
|
||||
this.lsm.put(`${tableName}:${newPk}`, updated);
|
||||
}
|
||||
count++;
|
||||
planned.push({ row, pk: String(row[pkCol]), key, updated, newPk, pkChanged });
|
||||
}
|
||||
// 阶段 1b：主键变更 RESTRICT / SET NULL+required 预检（任何修改前）
|
||||
for (const p of planned) {
|
||||
if (p.pkChanged) {
|
||||
await this.checkForeignKeyUpdateRestrict(tableName, p.pk, p.newPk);
|
||||
}
|
||||
}
|
||||
|
||||
// 阶段 2：执行（预检已通过，此阶段不再抛校验类错误）
|
||||
for (const { row, pk, key, updated, newPk, pkChanged } of planned) {
|
||||
if (pkChanged) {
|
||||
// ON UPDATE 外键级联（RESTRICT 抛错 / CASCADE / SET NULL）
|
||||
await this.applyForeignKeyUpdateRules(tableName, pk, newPk, walRecords, visited);
|
||||
}
|
||||
|
||||
if (this.currentTxnId && this.txnSnapshot) {
|
||||
if (pkChanged) {
|
||||
walRecords.push({
|
||||
type: WALRecordType.DELETE,
|
||||
txnId: this.currentTxnId ?? 0,
|
||||
tableName,
|
||||
key: String(row[pkCol]),
|
||||
});
|
||||
this.txnSnapshot.set(key, { __txn_deleted: true } as unknown as Record<string, unknown>);
|
||||
this.mvcc.deleteVersion(tableName, pk, this.currentTxnId);
|
||||
}
|
||||
this.txnSnapshot.set(`${tableName}:${newPk}`, updated);
|
||||
this.mvcc.writeVersion(tableName, newPk, updated, this.currentTxnId);
|
||||
} else {
|
||||
if (pkChanged) this.lsm.delete(key);
|
||||
this.lsm.put(`${tableName}:${newPk}`, updated);
|
||||
}
|
||||
count++;
|
||||
|
||||
if (pkChanged) {
|
||||
walRecords.push({
|
||||
type: WALRecordType.UPDATE,
|
||||
type: WALRecordType.DELETE,
|
||||
txnId: this.currentTxnId ?? 0,
|
||||
tableName,
|
||||
key: newPk,
|
||||
data: updated,
|
||||
key: pk,
|
||||
});
|
||||
|
||||
// 更新二级索引（主键变更时旧索引条目一并清理）
|
||||
// v0.6.2-fix: 此前非主键更新不传旧行 → 旧索引条目残留
|
||||
// （唯一性检查误报 / 索引存储膨胀）；现在统一传旧行清理旧值
|
||||
this.updateSecondaryIndexes(tableName, newPk, updated, row);
|
||||
}
|
||||
walRecords.push({
|
||||
type: WALRecordType.UPDATE,
|
||||
txnId: this.currentTxnId ?? 0,
|
||||
tableName,
|
||||
key: newPk,
|
||||
data: updated,
|
||||
});
|
||||
|
||||
// 更新二级索引（主键变更时旧索引条目一并清理）
|
||||
// v0.6.2-fix: 此前非主键更新不传旧行 → 旧索引条目残留
|
||||
// （唯一性检查误报 / 索引存储膨胀）；现在统一传旧行清理旧值
|
||||
this.updateSecondaryIndexes(tableName, newPk, updated, row);
|
||||
}
|
||||
|
||||
await this.wal.appendBatch(walRecords);
|
||||
@@ -796,6 +816,63 @@ export class AriaEngine implements IStorageEngine {
|
||||
return count;
|
||||
}
|
||||
|
||||
/**
|
||||
* v0.7.2: 批内唯一互查 — 两条行在同一语句中更新到同一唯一值时的兜底检查
|
||||
* （阶段 1 中索引尚未反映本语句的变更）。
|
||||
*/
|
||||
private checkBatchUnique(
|
||||
tableName: string,
|
||||
uniqueCols: string[],
|
||||
updated: Record<string, unknown>,
|
||||
batchUnique: Map<string, Set<unknown>>,
|
||||
): void {
|
||||
for (const colName of uniqueCols) {
|
||||
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 "${tableName}"`,
|
||||
'UNIQUE_VIOLATION',
|
||||
);
|
||||
}
|
||||
seen.add(value);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* v0.7.2: ON UPDATE 外键预检 — 从 applyForeignKeyUpdateRules 提取（两阶段 update 用）：
|
||||
* RESTRICT 存在依赖行抛错；SET NULL 撞 required 列同样整体拒绝。
|
||||
*/
|
||||
private async checkForeignKeyUpdateRestrict(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;
|
||||
if (colDef.onUpdate === 'RESTRICT' || (colDef.onUpdate === 'SET NULL' && colDef.required)) {
|
||||
const refRows = await this.getAllRows(refTableName);
|
||||
for (const refRow of refRows) {
|
||||
if (String(refRow[colName]) === oldPk) {
|
||||
const reason = colDef.onUpdate === 'RESTRICT'
|
||||
? `foreign key "${colName}" in "${refTableName}" has dependent rows`
|
||||
: `foreign key "${colName}" in "${refTableName}" is required (SET NULL violates constraint)`;
|
||||
throw new DatabaseError(
|
||||
`Cannot update "${tableName}" key "${oldPk}": ${reason}`,
|
||||
'FOREIGN_KEY_VIOLATION',
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* v0.4.2-fix: ON UPDATE 外键级联 — 主键 oldPk → newPk 时处理引用表。
|
||||
* RESTRICT 抛错 / CASCADE 更新 FK / SET NULL 置空（含索引与 WAL 记录）。
|
||||
@@ -947,6 +1024,13 @@ export class AriaEngine implements IStorageEngine {
|
||||
'FOREIGN_KEY_VIOLATION',
|
||||
);
|
||||
}
|
||||
// v0.7.2: SET NULL 到 required 列违反约束 —— 预检阶段整体拒绝
|
||||
if (colDef.onDelete === 'SET NULL' && colDef.required && matched.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') {
|
||||
const refPkCol = this.tablePKs.get(refTableName)!;
|
||||
for (const refRow of matched) {
|
||||
|
||||
Reference in New Issue
Block a user