fix: v0.7.4 写语句子查询 / 约束硬化 / 真惰性流式 — UPDATE-DELETE WHERE 子查询静默 0 行(四引擎,关联引用显式 NOT_SUPPORTED + EXPLAIN 同步)/ 主键 NULL-undefined 强制拒绝 / DROP INDEX 保留建表 UNIQUE(仅索引来源可解除)/ GROUP BY-DISTINCT-UNION 键类型安全编码 / UPDATE 未知列报错 / queryStream 多语句拒绝 / KVStore 后台错误跨 reopen 清理 + Hybrid begin 补偿 / RB-Tree 删除双黑修复 + LSM 死代码清理 / findStream 迭代器化真惰性(limit 早停 O(1) 内存)/ REINDEX 单次扫描 + 48 回归
This commit is contained in:
+1
-1
@@ -214,4 +214,4 @@ export class DatabaseError extends Error {
|
||||
// 版本
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export const VERSION = '0.7.3';
|
||||
export const VERSION = '0.7.4';
|
||||
|
||||
+8
-1
@@ -262,7 +262,14 @@ export class MetonaSqlark {
|
||||
onRow: (row: T) => void,
|
||||
): Promise<number> {
|
||||
this.ensureReady();
|
||||
const stmt = parseAll(sql)[0];
|
||||
// v0.7.4: 多语句显式拒绝 —— 此前 parseAll(sql)[0] 静默忽略后续语句:
|
||||
// 可流式时后续语句(如 DELETE)不执行,不可流式时回退 query() 却会执行
|
||||
// 全部语句 → 同一条 SQL 两种语义。流式 API 要求单条 SELECT。
|
||||
const statements = parseAll(sql);
|
||||
if (statements.length !== 1) {
|
||||
throw new DatabaseError('queryStream requires exactly one SELECT statement', 'PARSE_ERROR');
|
||||
}
|
||||
const stmt = statements[0];
|
||||
if (!stmt || stmt.type !== 'SELECT') {
|
||||
throw new DatabaseError('queryStream only supports SELECT statements', 'NOT_SUPPORTED');
|
||||
}
|
||||
|
||||
+81
-10
@@ -8,7 +8,7 @@
|
||||
import type { IStorageEngine } from '../interface';
|
||||
import type { QueryPlan, TableSchema, ColumnDef, WhereCondition } from '../../constants';
|
||||
import { DatabaseError } from '../../constants';
|
||||
import { matchWhere, applyOrderBy, projectColumns } from '../../query/where-matcher';
|
||||
import { matchWhere, applyOrderBy, projectColumns, containsUnresolvedSubqueries } from '../../query/where-matcher';
|
||||
import { checkFieldType, stripUndefinedUpdates } from '../../table/schema';
|
||||
|
||||
import type { AriaEngineConfig, SSTableMeta } from './types';
|
||||
@@ -63,6 +63,14 @@ export class AriaEngine implements IStorageEngine {
|
||||
// 二级索引:table.colKey → LSM
|
||||
private secondaryIndexes: Map<string, LSM> = new Map();
|
||||
|
||||
/**
|
||||
* v0.7.4: 由 CREATE UNIQUE INDEX 添加的 unique 列(table:col)。
|
||||
* 与建表 UNIQUE 约束区分:DROP INDEX 只允许解除索引来源的 unique,
|
||||
* 建表约束需重建表(对齐 SQLite 语义,此前静默解除且重启后永久消失)。
|
||||
* 注:重启后无法区分历史来源,schema 中的 unique 一律按建表约束保护(保守)。
|
||||
*/
|
||||
private uniqueIndexCols: Set<string> = new Set();
|
||||
|
||||
// MVCC 事务
|
||||
private mvcc: MVCCManager = new MVCCManager();
|
||||
private currentTxnId: number | null = null;
|
||||
@@ -297,6 +305,7 @@ export class AriaEngine implements IStorageEngine {
|
||||
this.schemas.clear();
|
||||
this.tablePKs.clear();
|
||||
this.secondaryIndexes.clear();
|
||||
this.uniqueIndexCols.clear();
|
||||
this.mvcc = new MVCCManager();
|
||||
this.currentTxnId = null;
|
||||
this.txnSnapshot = null;
|
||||
@@ -392,6 +401,7 @@ export class AriaEngine implements IStorageEngine {
|
||||
this.schemas.clear();
|
||||
this.tablePKs.clear();
|
||||
this.secondaryIndexes.clear();
|
||||
this.uniqueIndexCols.clear();
|
||||
this.lsm.clear();
|
||||
this.mvcc = new MVCCManager();
|
||||
this.currentTxnId = null;
|
||||
@@ -479,6 +489,11 @@ export class AriaEngine implements IStorageEngine {
|
||||
// v0.4.2-fix: 清理该表的全部二级索引 LSM 与持久化文件 —
|
||||
// 此前残留孤儿索引,重建同名表后旧索引数据污染新表(索引查询返回错误行)
|
||||
await this.cleanupTableIndexes(tableName);
|
||||
// v0.7.4: 清理该表的 unique 索引来源标记
|
||||
const uPrefix = `${tableName}:`;
|
||||
for (const k of this.uniqueIndexCols) {
|
||||
if (k.startsWith(uPrefix)) this.uniqueIndexCols.delete(k);
|
||||
}
|
||||
|
||||
this.schemas.delete(tableName);
|
||||
this.tablePKs.delete(tableName);
|
||||
@@ -701,6 +716,14 @@ export class AriaEngine implements IStorageEngine {
|
||||
): Promise<number> {
|
||||
this.ensureOpen();
|
||||
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 UPDATE WHERE (use db.query() to execute subqueries)',
|
||||
'NOT_SUPPORTED',
|
||||
);
|
||||
}
|
||||
const schema = this.schemas.get(tableName)!;
|
||||
const rows = await this.getAllRows(tableName);
|
||||
let count = 0;
|
||||
@@ -711,6 +734,14 @@ export class AriaEngine implements IStorageEngine {
|
||||
// v0.7.2: undefined 值视为"不更新该列"(保留旧值),null 显式置空
|
||||
const cleanUpdates = stripUndefinedUpdates(updates);
|
||||
|
||||
// v0.7.4: 未知列显式报错 —— 此前 SET nonexistent = ... 被静默写入存储行
|
||||
// (validateRow 只遍历 schema 列,脏列残留在行内并随 SSTable 持久化)
|
||||
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.6.2: 唯一约束 — 批量预加载本批更新涉及的唯一列索引范围(一次 drainChain)
|
||||
const uniqueCols = this.uniqueColumns(tableName, schema);
|
||||
for (const colName of uniqueCols) {
|
||||
@@ -941,6 +972,14 @@ export class AriaEngine implements IStorageEngine {
|
||||
async delete(tableName: string, query: QueryPlan): Promise<number> {
|
||||
this.ensureOpen();
|
||||
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 rows = await this.getAllRows(tableName);
|
||||
let count = 0;
|
||||
@@ -1175,13 +1214,14 @@ export class AriaEngine implements IStorageEngine {
|
||||
return count;
|
||||
}
|
||||
|
||||
// 全表惰性扫描(含 WHERE 过滤,不物化)
|
||||
// 全表惰性扫描(含 WHERE 过滤,不物化;v0.7.4: callback 返回 false 提前终止,
|
||||
// 未消费的 SSTable 块 / 子树不再解析 —— 真流式,大表 limit 内存 O(1))
|
||||
await this.lsm.prefetchRange(prefix, `${prefix}\uffff`);
|
||||
this.lsm.rangeScanLazy(prefix, `${prefix}\uffff`, (key, value) => {
|
||||
if (count >= limit) return;
|
||||
if (count >= limit) return false;
|
||||
const row = { ...value };
|
||||
row[pkCol] = key.slice(prefix.length);
|
||||
emit(row);
|
||||
return emit(row);
|
||||
});
|
||||
return count;
|
||||
}
|
||||
@@ -1350,7 +1390,11 @@ export class AriaEngine implements IStorageEngine {
|
||||
throw error;
|
||||
}
|
||||
colDef.index = true;
|
||||
if (unique) colDef.unique = true;
|
||||
if (unique) {
|
||||
colDef.unique = true;
|
||||
// v0.7.4: 记录唯一约束来源(DROP INDEX 时可解除;建表约束不可)
|
||||
this.uniqueIndexCols.add(`${tableName}:${column}`);
|
||||
}
|
||||
await this.persistSchemas();
|
||||
}
|
||||
|
||||
@@ -1369,8 +1413,20 @@ export class AriaEngine implements IStorageEngine {
|
||||
if (!colDef.index && !colDef.unique && !this.secondaryIndexes.has(`${tableName}:idx:${column}`)) {
|
||||
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
|
||||
// 静默解除约束(重启后 persistSchemas 使约束永久消失)。对齐 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 idxKey = `${tableName}:idx:${column}`;
|
||||
const idxLsm = this.secondaryIndexes.get(idxKey);
|
||||
@@ -1596,6 +1652,14 @@ export class AriaEngine implements IStorageEngine {
|
||||
if (colDef.required && (value === undefined || value === null)) {
|
||||
throw new DatabaseError(`Column "${colName}" is required in table "${schema.name}"`, 'VALIDATION_ERROR');
|
||||
}
|
||||
// v0.7.4: 主键列强制非空(SQL 语义 PK 隐含 NOT NULL)——
|
||||
// 此前 null/undefined 主键被 String() 化为 "null"/"undefined" 静默入库
|
||||
if (colDef.primaryKey && (value === undefined || value === null)) {
|
||||
throw new DatabaseError(
|
||||
`Primary key column "${colName}" in table "${schema.name}" cannot be null or undefined`,
|
||||
'VALIDATION_ERROR',
|
||||
);
|
||||
}
|
||||
if (value !== undefined && value !== null) {
|
||||
this.checkType(colName, colDef.type, value, colDef);
|
||||
}
|
||||
@@ -2147,9 +2211,17 @@ export class AriaEngine implements IStorageEngine {
|
||||
if (!schema) return 0;
|
||||
let rebuiltCount = 0;
|
||||
|
||||
for (const [colName, colDef] of Object.entries(schema.columns)) {
|
||||
// v0.3.3: 主键列不建冗余二级索引(主 LSM 即 PK 索引)
|
||||
if (!colDef.index && !colDef.unique) continue;
|
||||
// v0.7.4-perf: 单次全表扫描重建全部索引列 —— 此前每个索引列各做一次
|
||||
// getAllRows(N 列 × M 行全表扫描 + 每次 prefetchRange drainChain),
|
||||
// 多索引大表 REINDEX/崩溃恢复按索引列数线性放大
|
||||
const pkCol = this.tablePKs.get(tableName)!;
|
||||
const idxCols = Object.entries(schema.columns).filter(
|
||||
([, colDef]) => colDef.index || colDef.unique,
|
||||
);
|
||||
if (idxCols.length === 0) return 0;
|
||||
|
||||
const rows = await this.getAllRows(tableName);
|
||||
for (const [colName] of idxCols) {
|
||||
const idxKey = `${tableName}:idx:${colName}`;
|
||||
const idxLsm = this.secondaryIndexes.get(idxKey);
|
||||
if (!idxLsm) continue;
|
||||
@@ -2159,11 +2231,10 @@ export class AriaEngine implements IStorageEngine {
|
||||
rebuiltCount++;
|
||||
|
||||
// 从主 LSM 重建索引
|
||||
const rows = await this.getAllRows(tableName);
|
||||
for (const row of rows) {
|
||||
const val = row[colName];
|
||||
if (val !== undefined && val !== null) {
|
||||
idxLsm.put(`${String(val)}:${row[this.tablePKs.get(tableName)!]}`, { pk: row[this.tablePKs.get(tableName)!] });
|
||||
idxLsm.put(`${String(val)}:${row[pkCol]}`, { pk: row[pkCol] });
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -11,7 +11,7 @@
|
||||
import { MemTable } from './memtable';
|
||||
import { SSTableBuilder } from './sstable_builder';
|
||||
import { SSTableReader } from './sstable';
|
||||
import { MergeIterator, ArrayEntrySource } from './merge_iterator';
|
||||
import { MergeIterator, ArrayEntrySource, GeneratorEntrySource } from './merge_iterator';
|
||||
import { DatabaseError } from '../../../constants';
|
||||
import type { SSTableMeta } from '../types';
|
||||
import {
|
||||
@@ -427,26 +427,31 @@ export class LSM {
|
||||
|
||||
rangeScan(startKey: string, endKey: string): [string, Record<string, unknown>][] {
|
||||
const result: [string, Record<string, unknown>][] = [];
|
||||
this.rangeScanLazy(startKey, endKey, (k, v) => result.push([k, v]));
|
||||
this.rangeScanLazy(startKey, endKey, (k, v) => { result.push([k, v]); });
|
||||
return result;
|
||||
}
|
||||
|
||||
/** 惰性范围扫描:通过回调逐条返回,不一次性物化所有源 */
|
||||
/**
|
||||
* 惰性范围扫描:通过回调逐条返回,不一次性物化。
|
||||
* v0.7.4: 真惰性 —— 各源(MemTable/frozen/SSTable)以生成器接入 MergeIterator,
|
||||
* 逐条拉取;回调返回 false 时提前终止(未消费部分不再解析/物化)。
|
||||
* 此前实现内部 mergeIter.drain() 全量物化,与"流式不物化"宣称不符。
|
||||
*/
|
||||
rangeScanLazy(
|
||||
startKey: string,
|
||||
endKey: string,
|
||||
callback: (key: string, value: Record<string, unknown>) => void,
|
||||
callback: (key: string, value: Record<string, unknown>) => boolean | void,
|
||||
): void {
|
||||
const mergeIter = new MergeIterator();
|
||||
|
||||
mergeIter.addSource(new ArrayEntrySource(
|
||||
this.memtable.rangeScan(startKey, endKey),
|
||||
mergeIter.addSource(new GeneratorEntrySource(
|
||||
this.memtable.scanLazy(startKey, endKey),
|
||||
));
|
||||
|
||||
// pending frozen memtables(从新到旧,新数据 sourceIndex 更小)
|
||||
for (let i = this.frozenMemtables.length - 1; i >= 0; i--) {
|
||||
mergeIter.addSource(new ArrayEntrySource(
|
||||
this.frozenMemtables[i].rangeScan(startKey, endKey),
|
||||
mergeIter.addSource(new GeneratorEntrySource(
|
||||
this.frozenMemtables[i].scanLazy(startKey, endKey),
|
||||
));
|
||||
}
|
||||
|
||||
@@ -455,49 +460,24 @@ export class LSM {
|
||||
if (endKey < meta.minKey || startKey > meta.maxKey) continue;
|
||||
const reader = this.loadSSTableReader(meta);
|
||||
if (!reader) continue;
|
||||
reader.rangeScan(startKey, endKey, (k, v) => {
|
||||
mergeIter.addSource(new ArrayEntrySource([[k, v]]));
|
||||
});
|
||||
mergeIter.addSource(new GeneratorEntrySource(
|
||||
reader.scanLazy(startKey, endKey),
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
const merged = mergeIter.drain();
|
||||
for (const [k, v] of merged) {
|
||||
let entry = mergeIter.next();
|
||||
while (entry) {
|
||||
const [k, v] = entry;
|
||||
if (!(v as unknown as Record<string, unknown>).__tombstone) {
|
||||
callback(k, v);
|
||||
const cont = callback(k, v);
|
||||
// v0.7.4: 提前终止(流式 limit 达成)
|
||||
if (cont === false) return;
|
||||
}
|
||||
entry = mergeIter.next();
|
||||
}
|
||||
}
|
||||
|
||||
getAllEntries(): [string, Record<string, unknown>][] {
|
||||
const result = new Map<string, Record<string, unknown>>();
|
||||
|
||||
// 从最旧层级开始聚合;同层级内从最旧到最新遍历,
|
||||
// 保证 result.set 覆盖时最终保留最新值
|
||||
for (let level = MAX_LSM_LEVELS - 1; level >= 0; level--) {
|
||||
for (let i = this.levels[level].length - 1; i >= 0; i--) {
|
||||
const meta = this.levels[level][i];
|
||||
const reader = this.loadSSTableReader(meta);
|
||||
if (!reader) continue;
|
||||
reader.scanAll((k, v) => result.set(k, v));
|
||||
}
|
||||
}
|
||||
|
||||
// MemTable 覆盖(最新)
|
||||
for (const [k, v] of this.memtable.getAllEntries()) {
|
||||
result.set(k, v);
|
||||
}
|
||||
for (let i = this.frozenMemtables.length - 1; i >= 0; i--) {
|
||||
for (const [k, v] of this.frozenMemtables[i].getAllEntries()) {
|
||||
result.set(k, v);
|
||||
}
|
||||
}
|
||||
|
||||
return Array.from(result.entries()).filter(
|
||||
([, v]) => !(v as unknown as Record<string, unknown>).__tombstone,
|
||||
);
|
||||
}
|
||||
|
||||
// =======================================================================
|
||||
// Compaction
|
||||
// =======================================================================
|
||||
|
||||
@@ -114,6 +114,35 @@ class RedBlackTree<K, V> {
|
||||
this._rangeScan(this.root, startKey, endKey, callback);
|
||||
}
|
||||
|
||||
/**
|
||||
* v0.7.4: 惰性范围遍历(显式栈中序迭代 + 边界剪枝)。
|
||||
* 真流式扫描:生成器按需产出,提前终止(limit 达成)时剩余子树不再遍历。
|
||||
*/
|
||||
*scanLazy(startKey: K, endKey: K): Generator<[K, V]> {
|
||||
const stack: RBNode<K, V>[] = [];
|
||||
// 定位到 >= startKey 的最左节点(沿路入栈)
|
||||
let cur: RBNode<K, V> | null = this.root;
|
||||
while (cur) {
|
||||
if (cur.key >= startKey) {
|
||||
stack.push(cur);
|
||||
cur = cur.left;
|
||||
} else {
|
||||
cur = cur.right;
|
||||
}
|
||||
}
|
||||
while (stack.length > 0) {
|
||||
const node = stack.pop()!;
|
||||
// 中序递增:越过 endKey 后所有剩余节点均越界
|
||||
if (node.key > endKey) break;
|
||||
if (node.key >= startKey) yield [node.key, node.value];
|
||||
cur = node.right;
|
||||
while (cur) {
|
||||
stack.push(cur);
|
||||
cur = cur.left;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** 获取所有条目 */
|
||||
getAllEntries(): [K, V][] {
|
||||
const entries: [K, V][] = [];
|
||||
@@ -157,8 +186,14 @@ class RedBlackTree<K, V> {
|
||||
if (node.color === Color.BLACK) this.fixDelete(node.left, node.left!.parent);
|
||||
} else {
|
||||
const successor = this.minimum(node.right);
|
||||
if (successor!.parent !== node) {
|
||||
this.transplant(successor!, successor!.right);
|
||||
// v0.7.4: 在 transplant 重连前捕获 successor 原右子与父 ——
|
||||
// x(双黑修复起点)= successor 原右子(占位在 successor 原位置)。
|
||||
// 此前 `successor.right?.parent ?? null` 在重赋值后取值:x 指向 node 右子树,
|
||||
// 且 null 时 parent 为 null → fixDelete 直接跳过修复(删除黑色节点后失衡)。
|
||||
const successorRight = successor!.right;
|
||||
const successorParent = successor!.parent;
|
||||
if (successorParent !== node) {
|
||||
this.transplant(successor!, successorRight);
|
||||
successor!.right = node.right;
|
||||
successor!.right!.parent = successor;
|
||||
}
|
||||
@@ -167,7 +202,13 @@ class RedBlackTree<K, V> {
|
||||
successor!.left!.parent = successor;
|
||||
const origColor = successor!.color;
|
||||
successor!.color = node.color;
|
||||
if (origColor === Color.BLACK) this.fixDelete(successor!.right, successor!.right?.parent ?? null);
|
||||
if (origColor === Color.BLACK) {
|
||||
const x = successorRight;
|
||||
// x 为 null 占位:直接右子时其父为 successor(已移到 node 位置),
|
||||
// 间接右子时其父为 successor 原父(transplant 已把 x 接到其下)
|
||||
const xParent = x ? x.parent : (successorParent === node ? successor! : successorParent);
|
||||
this.fixDelete(x, xParent);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -425,6 +466,14 @@ export class MemTable {
|
||||
return entries;
|
||||
}
|
||||
|
||||
/** v0.7.4: 惰性范围扫描(真流式,逐条产出) */
|
||||
scanLazy(
|
||||
startKey: string,
|
||||
endKey: string,
|
||||
): Generator<[string, Record<string, unknown>]> {
|
||||
return this.tree.scanLazy(startKey, endKey);
|
||||
}
|
||||
|
||||
/** 条目数 */
|
||||
getEntryCount(): number {
|
||||
return this.tree.size;
|
||||
|
||||
@@ -35,6 +35,28 @@ export class ArrayEntrySource implements EntrySource {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* v0.7.4: 生成器数据源 —— 惰性迭代(真流式扫描)。
|
||||
* MergeIterator 的 next() 逐条拉取,生成器按需产出(findStream 提前终止时
|
||||
* 未消费部分不再物化,大表流式内存 O(1))。
|
||||
*/
|
||||
export class GeneratorEntrySource implements EntrySource {
|
||||
private iter: Generator<[string, Record<string, unknown>]>;
|
||||
|
||||
constructor(iter: Generator<[string, Record<string, unknown>]>) {
|
||||
this.iter = iter;
|
||||
}
|
||||
|
||||
next(): [string, Record<string, unknown>] | null {
|
||||
const r = this.iter.next();
|
||||
return r.done ? null : r.value;
|
||||
}
|
||||
|
||||
reset(): void {
|
||||
// 生成器不可重置;MergeIterator 无 reset 消费方,接口保留
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Heap 节点(用于多路归并)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
@@ -108,6 +108,20 @@ export class SSTableReader {
|
||||
endKey: string,
|
||||
callback: (key: string, value: Record<string, unknown>) => void,
|
||||
): void {
|
||||
// v0.7.4: 包装惰性生成器(行为一致,消除双份解析循环)
|
||||
for (const [key, value] of this.scanLazy(startKey, endKey)) {
|
||||
callback(key, value);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* v0.7.4: 惰性范围扫描 —— 生成器逐块逐条产出(真流式)。
|
||||
* 提前终止时未消费的块不再解析,大表流式内存 O(1)。
|
||||
*/
|
||||
*scanLazy(
|
||||
startKey: string,
|
||||
endKey: string,
|
||||
): Generator<[string, Record<string, unknown>]> {
|
||||
if (this.indexEntries.length === 0) return;
|
||||
const startBlockIdx = Math.max(0, this.locateBlockGE(startKey));
|
||||
// v0.6.1-fix(P0): 索引键是"块内最后一个 key"(builder 约定),
|
||||
@@ -121,7 +135,7 @@ export class SSTableReader {
|
||||
for (let bi = startBlockIdx; bi <= endBlockIdx && bi >= 0; bi++) {
|
||||
const entry = this.indexEntries[bi];
|
||||
const blockData = this.getBlockData(entry);
|
||||
// v0.4.1-fix: 残缺块跳过(rangeScan 继续后续块,不抛异常)
|
||||
// v0.4.1-fix: 残缺块跳过(继续后续块,不抛异常)
|
||||
if (!blockData) continue;
|
||||
const blockView = new DataView(blockData.buffer, blockData.byteOffset, blockData.byteLength);
|
||||
|
||||
@@ -144,7 +158,7 @@ export class SSTableReader {
|
||||
if (key >= startKey && key <= endKey) {
|
||||
try {
|
||||
const value = JSON.parse(new TextDecoder().decode(valBytes));
|
||||
callback(key, value);
|
||||
yield [key, value];
|
||||
} catch {
|
||||
// skip corrupted entry
|
||||
}
|
||||
|
||||
@@ -85,6 +85,8 @@ export class KVStore {
|
||||
async open(dbName: string): Promise<void> {
|
||||
if (this.opened) return;
|
||||
this.dbName = dbName;
|
||||
// v0.7.4: 打开时同样清理后台错误状态(防 close/reopen 残留)
|
||||
this.lastBackgroundError = null;
|
||||
await this.medium.open(dbName);
|
||||
this.index = new Map();
|
||||
this.seq = 0;
|
||||
@@ -162,6 +164,9 @@ export class KVStore {
|
||||
this.index.clear();
|
||||
this.seq = 0;
|
||||
this.logBytes = 0;
|
||||
// v0.7.4: 清理后台错误状态 —— 此前跨 close/reopen 残留,
|
||||
// 重开后首次 checkpoint 会抛出上一次生命周期的旧错误
|
||||
this.lastBackgroundError = null;
|
||||
this.opened = false;
|
||||
}
|
||||
|
||||
|
||||
+62
-2
@@ -6,7 +6,7 @@
|
||||
import type { IStorageEngine } from './interface';
|
||||
import type { QueryPlan, TableSchema, WhereCondition } from '../constants';
|
||||
import { DatabaseError } from '../constants';
|
||||
import { matchWhere, applyOrderBy, projectColumns } from '../query/where-matcher';
|
||||
import { matchWhere, applyOrderBy, projectColumns, containsUnresolvedSubqueries } from '../query/where-matcher';
|
||||
import { stripUndefinedUpdates } from '../table/schema';
|
||||
|
||||
export class MemoryEngine implements IStorageEngine {
|
||||
@@ -18,6 +18,12 @@ export class MemoryEngine implements IStorageEngine {
|
||||
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: {
|
||||
@@ -82,6 +88,11 @@ export class MemoryEngine implements IStorageEngine {
|
||||
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); }
|
||||
@@ -229,6 +240,23 @@ export class MemoryEngine implements IStorageEngine {
|
||||
// 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 行已写入
|
||||
// → 无事务下语句级部分提交(数据半更新且调用方已收到错误)。
|
||||
@@ -421,6 +449,14 @@ export class MemoryEngine implements IStorageEngine {
|
||||
|
||||
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) {
|
||||
@@ -551,7 +587,11 @@ export class MemoryEngine implements IStorageEngine {
|
||||
throw error;
|
||||
}
|
||||
colDef.index = true;
|
||||
if (unique) colDef.unique = 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> {
|
||||
@@ -570,8 +610,20 @@ export class MemoryEngine implements IStorageEngine {
|
||||
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);
|
||||
}
|
||||
@@ -644,6 +696,14 @@ export class MemoryEngine implements IStorageEngine {
|
||||
if (colDef.required && (value === undefined || value === null)) {
|
||||
throw new DatabaseError(`Column "${colName}" is required in table "${schema.name}"`, 'VALIDATION_ERROR');
|
||||
}
|
||||
// v0.7.4: 主键列强制非空(SQL 语义 PK 隐含 NOT NULL)——
|
||||
// 此前 null/undefined 主键被 String() 化为 "null"/"undefined" 静默入库
|
||||
if (colDef.primaryKey && (value === undefined || value === null)) {
|
||||
throw new DatabaseError(
|
||||
`Primary key column "${colName}" in table "${schema.name}" cannot be null or undefined`,
|
||||
'VALIDATION_ERROR',
|
||||
);
|
||||
}
|
||||
if (value !== undefined && value !== null) this.checkType(colName, colDef.type, value);
|
||||
if (value !== undefined) validated[colName] = value;
|
||||
}
|
||||
|
||||
+8
-1
@@ -292,7 +292,14 @@ export class HybridEngine implements IStorageEngine {
|
||||
|
||||
async beginTransaction(): Promise<void> {
|
||||
await this.memoryEngine.beginTransaction();
|
||||
await this.diskEngine.beginTransaction();
|
||||
// v0.7.4: 磁盘 begin 失败时补偿回滚内存快照 —— 此前内存已 begin、
|
||||
// 磁盘抛错 → 内存快照泄漏(后续所有事务报 TX_ACTIVE)
|
||||
try {
|
||||
await this.diskEngine.beginTransaction();
|
||||
} catch (error) {
|
||||
await this.memoryEngine.rollbackTransaction();
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async commitTransaction(): Promise<void> {
|
||||
|
||||
+57
-4
@@ -18,6 +18,26 @@ import { matchWhere, applyOrderBy, projectColumns } from './where-matcher';
|
||||
import { parseWhereCondition } from '../sql/parser';
|
||||
import type { WhereCondition } from '../constants';
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// 分组 / 去重键编码(v0.7.4)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* v0.7.4: 分组/去重键的类型安全编码 —— 此前 `String(v ?? 'null')` 使
|
||||
* null 与字符串 'null' 合并为一组(GROUP BY 静默少组),`String(v ?? '\0')`
|
||||
* 使 null/undefined/'\0' 在 DISTINCT/UNION 中互相吞并。类型前缀编码后
|
||||
* 各类型独立,仅同类型同值合并(与 where-matcher 的 === 语义一致)。
|
||||
*/
|
||||
function encodeGroupKey(v: unknown): string {
|
||||
if (v === null) return 'n';
|
||||
if (v === undefined) return 'u';
|
||||
if (typeof v === 'string') return `s${v}`;
|
||||
if (typeof v === 'number') return `d${v}`;
|
||||
if (typeof v === 'boolean') return `b${v}`;
|
||||
if (typeof v === 'object') return `o${JSON.stringify(v)}`;
|
||||
return `x${String(v)}`;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// CASE WHEN 表达式(v0.3.1)
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -147,7 +167,7 @@ export class QueryExecutor {
|
||||
const seen = new Set<string>();
|
||||
const result: Record<string, unknown>[] = [];
|
||||
for (const row of normalized) {
|
||||
const key = Object.values(row).map((v) => String(v ?? '\0')).join('\x1f');
|
||||
const key = Object.values(row).map(encodeGroupKey).join('\x1f');
|
||||
if (!seen.has(key)) {
|
||||
seen.add(key);
|
||||
result.push(row);
|
||||
@@ -155,7 +175,7 @@ export class QueryExecutor {
|
||||
}
|
||||
for (const row of rightRows) {
|
||||
const projected = this.projectUnionRow(row, leftCols);
|
||||
const key = Object.values(projected).map((v) => String(v ?? '\0')).join('\x1f');
|
||||
const key = Object.values(projected).map(encodeGroupKey).join('\x1f');
|
||||
if (!seen.has(key)) {
|
||||
seen.add(key);
|
||||
result.push(projected);
|
||||
@@ -195,7 +215,10 @@ export class QueryExecutor {
|
||||
rows = Array.isArray(result) ? result.length : 0;
|
||||
} else if (stmt.query.type === 'UPDATE' || stmt.query.type === 'DELETE') {
|
||||
try {
|
||||
// v0.7.4: 子查询解析后估算 —— 此前 $subquery 未解析使 count 恒 0
|
||||
await this.resolveWriteWhere(stmt.query);
|
||||
const plan = compileStatement(stmt.query);
|
||||
plan.where = stmt.query.where;
|
||||
rows = await this.engine.count(plan.table, plan);
|
||||
} catch { rows = 0; }
|
||||
}
|
||||
@@ -593,7 +616,9 @@ export class QueryExecutor {
|
||||
private executeGroupBy(rows: Record<string, unknown>[], stmt: SelectStatement): Record<string, unknown>[] {
|
||||
const groups = new Map<string, Record<string, unknown>[]>();
|
||||
for (const row of rows) {
|
||||
const key = stmt.groupBy!.map((col) => String(row[col] ?? 'null')).join('|');
|
||||
// v0.7.4: 类型安全键编码 —— 此前 String(row[col] ?? 'null') 使
|
||||
// null 与字符串 'null' 合并为一组(GROUP BY 静默少组)
|
||||
const key = stmt.groupBy!.map((col) => encodeGroupKey(row[col])).join('\x1f');
|
||||
if (!groups.has(key)) groups.set(key, []);
|
||||
groups.get(key)!.push(row);
|
||||
}
|
||||
@@ -660,7 +685,8 @@ export class QueryExecutor {
|
||||
private executeDistinct(rows: Record<string, unknown>[]): Record<string, unknown>[] {
|
||||
const seen = new Set<string>();
|
||||
return rows.filter((row) => {
|
||||
const key = Object.values(row).map((v) => String(v ?? '\0')).join('\x1f');
|
||||
// v0.7.4: 类型安全键编码(null 与 'null' 字符串、'\0' 分离)
|
||||
const key = Object.values(row).map(encodeGroupKey).join('\x1f');
|
||||
if (seen.has(key)) return false;
|
||||
seen.add(key);
|
||||
return true;
|
||||
@@ -714,15 +740,42 @@ export class QueryExecutor {
|
||||
}
|
||||
|
||||
private async executeUpdate(stmt: UpdateStatement): Promise<number> {
|
||||
// v0.7.4: 先解析 WHERE 子查询 —— 此前直接 compileStatement 调引擎:
|
||||
// 引擎层 matchWhere 的 $in/$nin 遇未解析的 $subquery 对象恒 false →
|
||||
// 所有行不匹配,UPDATE 静默影响 0 行(与 queryStream v0.7.3 修复同类)。
|
||||
await this.resolveWriteWhere(stmt);
|
||||
const plan = compileStatement(stmt);
|
||||
plan.where = stmt.where;
|
||||
return this.engine.update(plan.table, plan, stmt.sets);
|
||||
}
|
||||
|
||||
private async executeDelete(stmt: DeleteStatement): Promise<number> {
|
||||
// v0.7.4: 同 executeUpdate —— DELETE 子查询 WHERE 此前静默删除 0 行
|
||||
await this.resolveWriteWhere(stmt);
|
||||
const plan = compileStatement(stmt);
|
||||
plan.where = stmt.where;
|
||||
return this.engine.delete(plan.table, plan);
|
||||
}
|
||||
|
||||
/**
|
||||
* v0.7.4: 写语句(UPDATE/DELETE)WHERE 的子查询解析。
|
||||
* 非关联子查询($subquery)解析为具体值列表/标量;
|
||||
* 关联引用($col / 关联 EXISTS)在写语句中无法逐行绑定外层上下文
|
||||
* (引擎层 matchWhere 无 $col 绑定选项)→ 显式 NOT_SUPPORTED 而非静默 0 行。
|
||||
*/
|
||||
private async resolveWriteWhere(stmt: UpdateStatement | DeleteStatement): Promise<WhereCondition> {
|
||||
const where = stmt.where;
|
||||
if (!where || Object.keys(where).length === 0) return where ?? {};
|
||||
if (this.hasCorrelatedRefs(where)) {
|
||||
throw new DatabaseError(
|
||||
'Correlated subqueries and column references are not supported in UPDATE/DELETE WHERE clauses',
|
||||
'NOT_SUPPORTED',
|
||||
);
|
||||
}
|
||||
stmt.where = await this.resolveSubqueries(where);
|
||||
return stmt.where;
|
||||
}
|
||||
|
||||
private async executeCreateTable(stmt: CreateTableStatement): Promise<void> {
|
||||
// IF NOT EXISTS: 表已存在时静默返回
|
||||
if (stmt.ifNotExists) {
|
||||
|
||||
@@ -31,6 +31,36 @@ function compileLikeRegex(pattern: string): RegExp {
|
||||
// WHERE 匹配(顶层入口)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* v0.7.4: 检测 WHERE 中未解析的子查询/列引用标记($subquery / $col / $exists)。
|
||||
* QueryBuilder 等直通引擎的写路径(update/delete)不经 Executor 解析子查询,
|
||||
* 引擎层 matchWhere 对未解析标记恒 false → 静默影响 0 行。
|
||||
* 写路径预检阶段显式拒绝;SELECT 路径由 Executor 解析(不调用本函数)。
|
||||
*/
|
||||
export function containsUnresolvedSubqueries(where: WhereCondition | undefined): boolean {
|
||||
if (!where) return false;
|
||||
for (const [k, v] of Object.entries(where)) {
|
||||
if (k === '$and' || k === '$or') {
|
||||
if ((v as WhereCondition[]).some((sub) => containsUnresolvedSubqueries(sub))) return true;
|
||||
continue;
|
||||
}
|
||||
if (k === '$not') {
|
||||
if (containsUnresolvedSubqueries(v as WhereCondition)) return true;
|
||||
continue;
|
||||
}
|
||||
if (k === '$exists') return true;
|
||||
if (typeof v === 'object' && v !== null && !Array.isArray(v)) {
|
||||
for (const [, operand] of Object.entries(v as Record<string, unknown>)) {
|
||||
if (typeof operand === 'object' && operand !== null) {
|
||||
const ops = operand as Record<string, unknown>;
|
||||
if ('$subquery' in ops || '$col' in ops) return true;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* 匹配完整 WHERE 条件
|
||||
* @param row 当前数据行
|
||||
@@ -41,8 +71,7 @@ export function matchWhere(
|
||||
row: Record<string, unknown>,
|
||||
where: WhereCondition,
|
||||
options: { $col?: boolean } = {},
|
||||
): boolean {
|
||||
for (const [field, condition] of Object.entries(where)) {
|
||||
): boolean { for (const [field, condition] of Object.entries(where)) {
|
||||
// 顶层 $caseResult(v0.3.2):由 Executor 对 CASE WHEN 表达式逐行求值后产生
|
||||
if (field === '$caseResult') {
|
||||
if (condition !== true) return false;
|
||||
|
||||
@@ -103,6 +103,15 @@ export function validateRow(schema: TableSchema, row: Record<string, unknown>):
|
||||
);
|
||||
}
|
||||
|
||||
// v0.7.4: 主键列强制非空(SQL 语义 PK 隐含 NOT NULL)——
|
||||
// 此前 null/undefined 主键被 String() 化为 "null"/"undefined" 静默入库
|
||||
if (colDef.primaryKey && (value === undefined || value === null)) {
|
||||
throw new DatabaseError(
|
||||
`Primary key column "${colName}" in table "${schema.name}" cannot be null or undefined`,
|
||||
'VALIDATION_ERROR',
|
||||
);
|
||||
}
|
||||
|
||||
// 类型检查
|
||||
if (value !== undefined && value !== null) {
|
||||
checkFieldType(schema.name, colName, colDef.type, value, colDef);
|
||||
|
||||
Reference in New Issue
Block a user