331 lines
12 KiB
TypeScript
331 lines
12 KiB
TypeScript
/**
|
||
* metona-sqlark Query Executor — AST 执行器
|
||
* @module query/executor
|
||
*
|
||
* JOIN / GROUP BY / DISTINCT 逻辑在此层处理。
|
||
*/
|
||
|
||
import type { IStorageEngine } from '../engine/interface';
|
||
import type {
|
||
Statement, SelectStatement, InsertStatement, UpdateStatement,
|
||
DeleteStatement, CreateTableStatement, DropTableStatement, JoinClause,
|
||
} from './ast';
|
||
import { DatabaseError } from '../constants';
|
||
import { compileStatement } from './compiler';
|
||
import { createSchema, astColumnToColumnDef } from '../table/schema';
|
||
import { matchWhere, applyOrderBy, projectColumns } from './where-matcher';
|
||
import type { WhereCondition } from '../constants';
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Executor
|
||
// ---------------------------------------------------------------------------
|
||
|
||
export class QueryExecutor {
|
||
constructor(private engine: IStorageEngine) {}
|
||
|
||
async execute(stmt: Statement): Promise<unknown> {
|
||
switch (stmt.type) {
|
||
case 'SELECT': return this.executeSelect(stmt);
|
||
case 'INSERT': return this.executeInsert(stmt);
|
||
case 'UPDATE': return this.executeUpdate(stmt);
|
||
case 'DELETE': return this.executeDelete(stmt);
|
||
case 'CREATE_TABLE': return this.executeCreateTable(stmt);
|
||
case 'DROP_TABLE': return this.executeDropTable(stmt);
|
||
default: throw new DatabaseError('Unknown statement type', 'UNKNOWN_STATEMENT');
|
||
}
|
||
}
|
||
|
||
// ===================================================================
|
||
// SELECT
|
||
// ===================================================================
|
||
|
||
private async executeSelect(stmt: SelectStatement): Promise<Record<string, unknown>[]> {
|
||
// 先解析子查询
|
||
if (stmt.where && Object.keys(stmt.where).length > 0) {
|
||
stmt.where = await this.resolveSubqueries(stmt.where);
|
||
}
|
||
|
||
const hasGroupBy = !!(stmt.groupBy && stmt.groupBy.length > 0);
|
||
let rows: Record<string, unknown>[];
|
||
|
||
if (!stmt.joins || stmt.joins.length === 0) {
|
||
const plan = compileStatement(hasGroupBy ? { ...stmt, columns: ['*'] } : stmt);
|
||
rows = await this.engine.find(plan.table, plan);
|
||
} else {
|
||
rows = await this.executeJoinSelect(stmt);
|
||
}
|
||
|
||
if (hasGroupBy) rows = this.executeGroupBy(rows, stmt);
|
||
if (stmt.distinct) rows = this.executeDistinct(rows);
|
||
if (stmt.having && Object.keys(stmt.having).length > 0) {
|
||
rows = rows.filter((row) => matchWhere(row, stmt.having!));
|
||
}
|
||
if (stmt.orderBy && stmt.orderBy.length > 0) rows = applyOrderBy(rows, stmt.orderBy);
|
||
const offset = stmt.offset ?? 0;
|
||
const limit = stmt.limit ?? rows.length;
|
||
rows = rows.slice(offset, offset + limit);
|
||
if (!hasGroupBy && stmt.columns.length > 0 && stmt.columns[0] !== '*') {
|
||
rows = rows.map((row) => projectColumns(row, stmt.columns));
|
||
}
|
||
return rows;
|
||
}
|
||
|
||
// ---- JOIN ----
|
||
|
||
private async executeJoinSelect(stmt: SelectStatement): Promise<Record<string, unknown>[]> {
|
||
const mainAlias = stmt.alias ?? stmt.from;
|
||
const mainRows = (await this.engine.find(stmt.from, { table: stmt.from }))
|
||
.map((row) => this.prefixRow(row, mainAlias));
|
||
let resultRows = mainRows;
|
||
|
||
for (const join of stmt.joins!) {
|
||
const joinAlias = join.alias ?? join.table;
|
||
const joinRows = (await this.engine.find(join.table, { table: join.table }))
|
||
.map((row) => this.prefixRow(row, joinAlias));
|
||
resultRows = this.joinRows(resultRows, joinRows, join);
|
||
}
|
||
|
||
if (stmt.where && Object.keys(stmt.where).length > 0) {
|
||
resultRows = resultRows.filter((row) => matchWhere(row, stmt.where));
|
||
}
|
||
return resultRows;
|
||
}
|
||
|
||
private prefixRow(row: Record<string, unknown>, alias: string): Record<string, unknown> {
|
||
const prefixed: Record<string, unknown> = {};
|
||
for (const [key, value] of Object.entries(row)) prefixed[`${alias}.${key}`] = value;
|
||
return prefixed;
|
||
}
|
||
|
||
/** 嵌套循环连接（优化：避免 ON 时对象扩散） */
|
||
private joinRows(
|
||
leftRows: Record<string, unknown>[],
|
||
rightRows: Record<string, unknown>[],
|
||
join: JoinClause,
|
||
): Record<string, unknown>[] {
|
||
if (join.type === 'CROSS') {
|
||
const result: Record<string, unknown>[] = [];
|
||
for (const l of leftRows) for (const r of rightRows) result.push({ ...l, ...r });
|
||
return result;
|
||
}
|
||
|
||
const result: Record<string, unknown>[] = [];
|
||
for (const l of leftRows) {
|
||
let matched = false;
|
||
for (const r of rightRows) {
|
||
// 合并后匹配 ON（避免创建临时对象再丢弃）
|
||
const merged = { ...l, ...r };
|
||
if (matchWhere(merged, join.on, { $col: true })) {
|
||
result.push(merged);
|
||
matched = true;
|
||
}
|
||
}
|
||
if (!matched && join.type === 'LEFT') {
|
||
const nullRight: Record<string, unknown> = {};
|
||
for (const key of Object.keys(rightRows[0] ?? {})) nullRight[key] = null;
|
||
result.push({ ...l, ...nullRight });
|
||
}
|
||
}
|
||
|
||
if (join.type === 'RIGHT') {
|
||
for (const r of rightRows) {
|
||
const isMatched = leftRows.some((l) => {
|
||
const merged = { ...l, ...r };
|
||
return matchWhere(merged, join.on, { $col: true });
|
||
});
|
||
if (!isMatched) {
|
||
const nullLeft: Record<string, unknown> = {};
|
||
for (const key of Object.keys(leftRows[0] ?? {})) nullLeft[key] = null;
|
||
result.push({ ...nullLeft, ...r });
|
||
}
|
||
}
|
||
}
|
||
return result;
|
||
}
|
||
|
||
// ---- GROUP BY ----
|
||
|
||
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('|');
|
||
if (!groups.has(key)) groups.set(key, []);
|
||
groups.get(key)!.push(row);
|
||
}
|
||
const result: Record<string, unknown>[] = [];
|
||
for (const groupRows of groups.values()) {
|
||
const aggregated: Record<string, unknown> = {};
|
||
for (const col of stmt.groupBy!) aggregated[col] = groupRows[0][col];
|
||
for (const colExpr of stmt.columns) {
|
||
if (colExpr === '*') continue;
|
||
const m = colExpr.match(/^(COUNT|SUM|AVG|MIN|MAX)\((.+?)\)(?:\s+AS\s+(\w+))?$/i);
|
||
if (m) {
|
||
const [, func, arg, alias] = m;
|
||
aggregated[alias || colExpr] = this.computeAggregate(func.toUpperCase(), groupRows, arg.trim());
|
||
} else if (!stmt.groupBy!.includes(colExpr)) {
|
||
aggregated[colExpr] = groupRows[0][colExpr];
|
||
}
|
||
}
|
||
result.push(aggregated);
|
||
}
|
||
return result;
|
||
}
|
||
|
||
private computeAggregate(func: string, rows: Record<string, unknown>[], col: string): number {
|
||
const nums = rows.map((r) => r[col]).filter((v) => v !== null && v !== undefined).map(Number);
|
||
switch (func) {
|
||
case 'COUNT': return col === '*' ? rows.length : nums.length;
|
||
case 'SUM': return nums.reduce((a: number, b) => a + b, 0);
|
||
case 'AVG': return nums.length === 0 ? 0 : nums.reduce((a: number, b) => a + b, 0) / nums.length;
|
||
case 'MIN': return nums.length === 0 ? 0 : Math.min(...nums);
|
||
case 'MAX': return nums.length === 0 ? 0 : Math.max(...nums);
|
||
default: return 0;
|
||
}
|
||
}
|
||
|
||
// ---- DISTINCT（优化：列值拼接代替 JSON.stringify） ----
|
||
|
||
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');
|
||
if (seen.has(key)) return false;
|
||
seen.add(key);
|
||
return true;
|
||
});
|
||
}
|
||
|
||
// ===================================================================
|
||
// 其他语句
|
||
// ===================================================================
|
||
|
||
private async executeInsert(stmt: InsertStatement): Promise<string[]> {
|
||
const schema = await this.engine.getTableSchema(stmt.into);
|
||
if (!schema) throw new DatabaseError(`Table "${stmt.into}" does not exist`, 'TABLE_NOT_FOUND');
|
||
const colNames = stmt.columns ?? Object.keys(schema.columns);
|
||
const rows: Record<string, unknown>[] = stmt.values.map((vals: unknown[]) => {
|
||
const row: Record<string, unknown> = {};
|
||
for (let i = 0; i < colNames.length; i++) { if (i < vals.length) row[colNames[i]] = vals[i]; }
|
||
return row;
|
||
});
|
||
return this.engine.insert(stmt.into, rows);
|
||
}
|
||
|
||
private async executeUpdate(stmt: UpdateStatement): Promise<number> {
|
||
const plan = compileStatement(stmt);
|
||
return this.engine.update(plan.table, plan, stmt.sets);
|
||
}
|
||
|
||
private async executeDelete(stmt: DeleteStatement): Promise<number> {
|
||
const plan = compileStatement(stmt);
|
||
return this.engine.delete(plan.table, plan);
|
||
}
|
||
|
||
private async executeCreateTable(stmt: CreateTableStatement): Promise<void> {
|
||
const columns: Record<string, any> = {};
|
||
for (const col of stmt.columns) columns[col.name] = astColumnToColumnDef(col);
|
||
return this.engine.createTable(createSchema(stmt.name, columns));
|
||
}
|
||
|
||
private async executeDropTable(stmt: DropTableStatement): Promise<void> {
|
||
return this.engine.dropTable(stmt.name);
|
||
}
|
||
|
||
getEngine(): IStorageEngine { return this.engine; }
|
||
|
||
// ===================================================================
|
||
// 子查询解析
|
||
// ===================================================================
|
||
|
||
/**
|
||
* 递归扫描 WHERE 条件，找到 $subquery 标记并执行子查询，
|
||
* 将结果替换为具体值。
|
||
*/
|
||
private async resolveSubqueries(where: WhereCondition): Promise<WhereCondition> {
|
||
const resolved: WhereCondition = {};
|
||
|
||
for (const [key, value] of Object.entries(where)) {
|
||
// 逻辑组合操作符
|
||
if (key === '$and' && Array.isArray(value)) {
|
||
resolved.$and = await Promise.all(
|
||
(value as WhereCondition[]).map((sub) => this.resolveSubqueries(sub)),
|
||
);
|
||
continue;
|
||
}
|
||
if (key === '$or' && Array.isArray(value)) {
|
||
resolved.$or = await Promise.all(
|
||
(value as WhereCondition[]).map((sub) => this.resolveSubqueries(sub)),
|
||
);
|
||
continue;
|
||
}
|
||
if (key === '$not' && typeof value === 'object' && value !== null) {
|
||
resolved.$not = await this.resolveSubqueries(value as WhereCondition);
|
||
continue;
|
||
}
|
||
|
||
// 字段条件
|
||
if (typeof value === 'object' && value !== null) {
|
||
resolved[key] = await this.resolveOperatorSubqueries(value as Record<string, unknown>);
|
||
} else {
|
||
resolved[key] = value;
|
||
}
|
||
}
|
||
|
||
return resolved;
|
||
}
|
||
|
||
/**
|
||
* 解析操作符值中嵌套的子查询
|
||
*/
|
||
private async resolveOperatorSubqueries(ops: Record<string, unknown>): Promise<Record<string, unknown>> {
|
||
const resolved: Record<string, unknown> = {};
|
||
|
||
for (const [op, operand] of Object.entries(ops)) {
|
||
// 处理嵌套 $and/$or（在字段级条件中）
|
||
if (op === '$and' && Array.isArray(operand)) {
|
||
resolved.$and = await Promise.all(
|
||
(operand as WhereCondition[]).map((sub) => this.resolveSubqueries(sub)),
|
||
);
|
||
continue;
|
||
}
|
||
if (op === '$or' && Array.isArray(operand)) {
|
||
resolved.$or = await Promise.all(
|
||
(operand as WhereCondition[]).map((sub) => this.resolveSubqueries(sub)),
|
||
);
|
||
continue;
|
||
}
|
||
if (op === '$not') {
|
||
resolved.$not = typeof operand === 'object' && operand !== null
|
||
? await this.resolveOperatorSubqueries(operand as Record<string, unknown>)
|
||
: operand;
|
||
continue;
|
||
}
|
||
|
||
// 子查询检测
|
||
if (typeof operand === 'object' && operand !== null && '$subquery' in (operand as Record<string, unknown>)) {
|
||
const subStmt = (operand as Record<string, unknown>).$subquery as SelectStatement;
|
||
const subResult = await this.executeSelect(subStmt);
|
||
|
||
if (op === '$in' || op === '$nin') {
|
||
// IN 子查询 → 提取第一列的值列表
|
||
const colName = Object.keys(subResult[0] || {})[0];
|
||
const values = subResult.map((row) => row[colName]);
|
||
resolved[op] = values;
|
||
} else {
|
||
// 标量子查询 → 取第一行第一列
|
||
if (subResult.length === 0) {
|
||
resolved[op] = null;
|
||
} else {
|
||
const colName = Object.keys(subResult[0])[0];
|
||
resolved[op] = subResult[0][colName];
|
||
}
|
||
}
|
||
} else {
|
||
resolved[op] = operand;
|
||
}
|
||
}
|
||
|
||
return resolved;
|
||
}
|
||
}
|