fix(A5/A6/A7/A16/A18/A24/A28): 查询层语义根治 —— LIMIT 双重应用、未知列静默、聚合崩溃与空集语义
A6 LIMIT/OFFSET 被应用两次（丢行）
compileSelect 无条件下推 limit/offset，引擎切一次，executeSelect 末尾又切一次。
实测 4 行表：LIMIT 2 OFFSET 1 只返回 1 行；LIMIT 10 OFFSET 3 返回空。
JOIN/派生表路径因不走 plan 反而正确，同一 executor 内自相矛盾。
现引入 analyzeSelect() 统一判定 limitPushdownSafe，两条路径互斥且只应用一次。
实测 10 种查询形态（含 DISTINCT/GROUP BY/别名/深 OFFSET/LIMIT 0）全部正确。
A7 queryStream 与 query 结果不一致（四类静默分歧）
core.ts 自己重写了一套能否走引擎快路径/如何投影的规则，与 executor 各写一份：
SELECT id AS x FROM t query [{x}]，stream [{id,v}]（全列+原列名）
SELECT t.id FROM t query [{id}]，stream [{}]（空对象）
LIMIT 2 OFFSET 1 query 1 行，stream 2 行
LIMIT 0 query 0 行，stream 1 行（Aria 又是 0 行）
且流式路径不触发 beforeQuery/afterQuery、不受 maxRowsPerQuery 约束。
现由 executor.analyzeSelect() 做单一事实来源；不可流式一律回退物化路径；
列引用剥离主表别名前缀；LIMIT 0 短路；回调返回 Promise 时显式报错
（此前用 constructor.name === 'AsyncFunction' 判定，普通函数返回 Promise 会静默丢弃）。
实测 16 种查询形态 × 4 引擎 = 64 组，query 与 queryStream 逐值相等。
A5 MIN/MAX 栈溢出（崩溃）
Math.min(...arr) 展开实参：20 万行同组直接 RangeError。改为单次遍历归约 reduceNumeric。
A24 空集聚合语义
SUM/AVG/MIN/MAX 对空集返回 0（使 和为 0 与 无数据 不可区分），改为 SQL 标准的 NULL；
COUNT 仍返回 0。
A18 未知列静默产出空对象行
SELECT bogus FROM t 返回 [{},{},...]（行数对、内容空、无报错）；SELECT NAME（列名 name）
同样静默空。新增 assertProjectionColumnsExist：投影前校验列存在性，
未知列抛 COLUMN_NOT_FOUND；同一后缀命中多个表别名时抛歧义错误。空结果集不误报。
A16 INSERT 列/值个数不校验（静默丢弃/写半行）
显式列名时在解析期校验每行值个数与列数一致：
INSERT INTO t (id,name) VALUES ('4','z',9) 此前静默丢弃 9，现报错。
A28 UPDATE SET __proto__ 静默吞掉
sets['__proto__'] 触发原型 setter，sets 变空对象，既不写入也不被未知列预检看到，
表现为返回成功但什么都没发生。parser 的列名映射统一改为 Object.create(null)。
附带修复（A18 验证时发现）：
SELECT 1 AS one FROM t 返回 [{}] —— parseColumnRef 的数字分支提前 return 吞掉别名；
裸 SELECT 1 同样产出空对象 —— 投影未处理匿名常量列。现按 SQLite 语义以表达式原文为键。
This commit is contained in:
thzxx
+78
-49
@@ -257,6 +257,27 @@ export class MetonaSqlark {
|
||||
* });
|
||||
* ```
|
||||
*/
|
||||
/**
|
||||
* 流式查询：逐行回调，尽量不物化全部结果（大表友好）。
|
||||
*
|
||||
* v0.8.0 根治：**快路径与物化路径的结果必须逐值相等**。
|
||||
*
|
||||
* 此前 core.ts 自己重写了一套"能不能走引擎快路径 / 列投影怎么算"的规则，
|
||||
* 与 executor 的规则各写一份并发生漂移，实测四类静默不一致：
|
||||
* SELECT id AS x FROM t → query 返回 [{x}]，stream 返回 [{id,v}]（全列 + 原列名）
|
||||
* SELECT t.id FROM t → query 返回 [{id}]，stream 返回 [{}]（空对象）
|
||||
* ... LIMIT 2 OFFSET 1 → query 1 行，stream 2 行
|
||||
* ... LIMIT 0 → query 0 行，stream 1 行（Aria 又是 0 行，跨引擎也不同）
|
||||
* 另：流式路径既不触发 beforeQuery/afterQuery 钩子，也不受 maxRowsPerQuery 约束。
|
||||
*
|
||||
* 现在的规则：
|
||||
* 1. 是否可流式、如何投影，全部由 `executor.analyzeSelect()` 判定（单一事实来源）；
|
||||
* 2. 不可流式（以及任何不确定的情况）一律回退到 `query()` 物化后逐行回调 ——
|
||||
* 这条路径天然与 `query()` 同语义，是正确性的兜底保证；
|
||||
* 3. 快路径只覆盖"引擎层投影与 executor 投影语义等价"的简单 SELECT；
|
||||
* 4. 回调返回 Promise 时不再靠 `constructor.name` 猜（此前对普通函数返回 Promise
|
||||
* 的情况完全失效），而是直接检测返回值并显式报错，避免 Promise 被静默丢弃。
|
||||
*/
|
||||
async queryStream<T extends Record<string, unknown> = Record<string, unknown>>(
|
||||
sql: string,
|
||||
onRow: (row: T) => void,
|
||||
@@ -275,56 +296,54 @@ export class MetonaSqlark {
|
||||
}
|
||||
const select = stmt as import('./query/ast').SelectStatement;
|
||||
|
||||
// 不可流式场景：JOIN / GROUP BY / HAVING / DISTINCT / 聚合 / UNION / 关联子查询 / ORDER BY
|
||||
const aggregate = select.columns.some((c) => /^(COUNT|SUM|AVG|MIN|MAX)\(/i.test(c));
|
||||
// v0.7.3: WHERE 含子查询（$subquery / $exists / 嵌套 $col 列引用）不可流式 ——
|
||||
// 引擎层 matchWhere 的 $in/$nin 遇未解析的 $subquery 对象返回 false → 所有行
|
||||
// 被静默过滤（空结果）；$col 操作符无对应匹配分支会抛 QUERY_ERROR。
|
||||
// 递归检测后回退物化路径（resolveSubqueries 正确解析）。
|
||||
const hasSubquery = (where: import('./constants').WhereCondition | undefined): boolean => {
|
||||
if (!where) return false;
|
||||
for (const [k, v] of Object.entries(where)) {
|
||||
if (k === '$and' || k === '$or') {
|
||||
if ((v as import('./constants').WhereCondition[]).some((sub) => hasSubquery(sub))) return true;
|
||||
continue;
|
||||
}
|
||||
if (k === '$not') {
|
||||
if (hasSubquery(v as import('./constants').WhereCondition)) return true;
|
||||
continue;
|
||||
}
|
||||
if (k === '$exists') return true;
|
||||
if (typeof v === 'object' && v !== null) {
|
||||
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;
|
||||
};
|
||||
const streamable = !select.joins && !select.groupBy && !select.having && !select.distinct
|
||||
&& !aggregate && !(select.orderBy && select.orderBy.length > 0)
|
||||
&& !hasSubquery(select.where);
|
||||
// 执行形态由 executor 统一判定（与 query() 路径共用同一规则）
|
||||
const shape = this.executor.analyzeSelect(select);
|
||||
|
||||
if (streamable && typeof this.engine.findStream === 'function') {
|
||||
// 用户回调为 async（返回 Promise）时引擎同步扫描无法 await → 回退物化
|
||||
const isAsync = (onRow as { constructor?: { name?: string } }).constructor?.name === 'AsyncFunction';
|
||||
if (!isAsync) {
|
||||
const where = this.normalizeWhereForStream(select);
|
||||
const plainCols = select.columns.filter((c) => !/\s+AS\s+\w+$/i.test(c));
|
||||
return this.engine.findStream(select.from, {
|
||||
table: select.from,
|
||||
columns: plainCols.length > 0 && plainCols[0] !== '*' ? plainCols : ['*'],
|
||||
where: where && Object.keys(where).length > 0 ? where : undefined,
|
||||
limit: select.limit,
|
||||
offset: select.offset,
|
||||
}, onRow as (row: Record<string, unknown>) => void);
|
||||
}
|
||||
// LIMIT 0 语义：任何引擎都必须返回 0 行。
|
||||
// 引擎对 `limit: 0` 的解释并不一致（Aria 返回 0 行，Memory/KVStore/Hybrid 把 0 当
|
||||
// "无限制"返回全部行 —— 实测 LIMIT 0 在四种引擎下分别为 0/1/1/1 行）。
|
||||
// 流式路径直接短路，避免依赖各引擎对 0 的解释。
|
||||
if (select.limit === 0) return 0;
|
||||
|
||||
if (shape.streamable && typeof this.engine.findStream === 'function') {
|
||||
const where = this.normalizeWhereForStream(select);
|
||||
// 与 executor 的非 JOIN 路径一致：剥离主表别名前缀后再交给引擎
|
||||
// （executor 对 `SELECT t.id FROM t` 会发 columns=['id']；此前流式路径把
|
||||
// 't.id' 原样传给引擎，引擎按 't.id' 建键 → 行里取不到 → 回调收到 {}）。
|
||||
const mainAliases = [select.alias ?? select.from].filter(Boolean);
|
||||
const columns = select.columns.length > 0
|
||||
? select.columns.map((c) => this.stripAliasPrefix(c, mainAliases))
|
||||
: ['*'];
|
||||
const maxRows = this.maxRowsPerQuery;
|
||||
let emitted = 0;
|
||||
let streamingError: unknown = null;
|
||||
|
||||
const count = await this.engine.findStream(select.from, {
|
||||
table: select.from,
|
||||
columns,
|
||||
where: where && Object.keys(where).length > 0 ? where : undefined,
|
||||
limit: select.limit,
|
||||
offset: select.offset,
|
||||
}, (row: Record<string, unknown>) => {
|
||||
// maxRowsPerQuery 必须与物化路径一致地生效（此前流式路径完全不受约束）
|
||||
if (maxRows > 0 && emitted >= maxRows) return;
|
||||
emitted++;
|
||||
const ret = (onRow as (r: Record<string, unknown>) => unknown)(row);
|
||||
if (ret && typeof (ret as PromiseLike<unknown>).then === 'function') {
|
||||
// 同步扫描无法 await 用户回调 —— 显式报错而不是静默丢弃 Promise
|
||||
streamingError = new DatabaseError(
|
||||
'queryStream callback returned a Promise; use await db.query() for async row handlers',
|
||||
'NOT_SUPPORTED',
|
||||
);
|
||||
}
|
||||
});
|
||||
|
||||
if (streamingError) throw streamingError;
|
||||
// 引擎返回的行数在 maxRowsPerQuery 截断时需与回调次数一致
|
||||
return maxRows > 0 ? Math.min(count, maxRows) : count;
|
||||
}
|
||||
|
||||
// 回退：物化后逐行回调
|
||||
// 回退：物化后逐行回调（与 query() 完全同语义，含钩子与 maxRowsPerQuery）
|
||||
const result = await this.query(sql);
|
||||
if (Array.isArray(result)) {
|
||||
for (const row of result as T[]) {
|
||||
@@ -335,9 +354,19 @@ export class MetonaSqlark {
|
||||
return 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* v0.8.0: 剥离列引用上的主表别名前缀（`t.id` → `id`）。
|
||||
* 与 executor 非 JOIN 路径的 `stripAlias` 语义保持一致。
|
||||
*/
|
||||
private stripAliasPrefix(col: string, aliases: string[]): string {
|
||||
for (const a of aliases) {
|
||||
if (a && col.startsWith(`${a}.`)) return col.slice(a.length + 1);
|
||||
}
|
||||
return col;
|
||||
}
|
||||
|
||||
/** 流式查询用：剥离主表别名前缀（复用 query 路径的规范化逻辑） */
|
||||
private normalizeWhereForStream(select: import('./query/ast').SelectStatement): import('./constants').WhereCondition | undefined {
|
||||
const aliases = [select.alias ?? select.from].filter(Boolean);
|
||||
private normalizeWhereForStream(select: import('./query/ast').SelectStatement): import('./constants').WhereCondition | undefined { const aliases = [select.alias ?? select.from].filter(Boolean);
|
||||
const strip = (col: string): string => {
|
||||
for (const a of aliases) {
|
||||
if (col.startsWith(`${a}.`)) return col.slice(a.length + 1);
|
||||
|
||||
Reference in New Issue
Block a user