/**
 * AriaEngine — 自研页面式存储引擎主类
 * @module engine/aria/index
 *
 * v0.4.1: 外键级联 + ALTER TABLE 重写 + clearAll 重置 + 崩溃恢复加固
 */

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 type { AriaEngineConfig, SSTableMeta } from './types';
import { DEFAULT_ARIA_CONFIG } from './types';

import { LSM } from './index/lsm';
import type { SSTableStore } from './index/lsm';
import { WAL } from './wal/log';
import { SegmentedWALStore } from './wal/segmented_store';
import { DatabaseLock } from './locks';
import { WALRecordType, type WALRecord } from './types';
import { CheckpointManager } from './wal/checkpoint';
import { MemoryBackend, type IStorageBackend } from './store/backend';
import { OPFSBackend } from './store/opfs_backend';
import { EncryptedBackend } from './store/encrypted_backend';
import { PageSSTableStore } from './store/page_sstable_store';
import { FileManager } from './store/file_manager';
import { MVCCManager } from './transaction/mvcc';
import { BufferPool } from './buffer/pool';
import { compressLZ4, decompressLZ4 } from './compression/lz4';
import type { WALRecord as _WALRecord } from './types';

// ---------------------------------------------------------------------------
//  AriaEngine
// ---------------------------------------------------------------------------

export class AriaEngine implements IStorageEngine {
  readonly name = 'aria';

  private config!: Required<Omit<AriaEngineConfig, 'encryption' | 'pageStorage'>> &
    Pick<AriaEngineConfig, 'encryption' | 'pageStorage'>;
  private lsm!: LSM; // 主键索引 LSM
  private wal!: WAL;
  private checkpointManager!: CheckpointManager;
  private backend!: IStorageBackend;
  private opened = false;
  private dbName = '';

  // v0.4.5: 页面化物理存储（FileManager + BufferPool 提升为引擎字段，close/repair 时落盘/清理）
  private fileManager!: FileManager;
  private bufferPool!: BufferPool;

  // v0.4.5: 多标签页独占锁（Web Locks API，OPFS 等无事务后端防并发写）
  private dbLock: DatabaseLock | null = null;

  // 表结构
  private schemas: Map<string, TableSchema> = new Map();
  private tablePKs: Map<string, string> = new Map();
  private opCounter = 0;

  // 二级索引：table.colKey → LSM
  private secondaryIndexes: Map<string, LSM> = new Map();

  // MVCC 事务
  private mvcc: MVCCManager = new MVCCManager();
  private currentTxnId: number | null = null;
  private txnSnapshot: Map<string, Record<string, unknown>> | null = null;
  private gcCounter = 0;

  constructor(config: AriaEngineConfig = {}) {
    this.config = { ...DEFAULT_ARIA_CONFIG, ...config };
  }

  // =======================================================================
  //  生命周期
  // =======================================================================

  async open(dbName: string, _version: number): Promise<void> {
    if (this.opened) return;
    // v0.4.2-fix: 引擎内部错误统一包装为 DatabaseError（ARIA_OPEN_ERROR），
    // 应用层可拿到 code 分类处理，不再抛出原生 RangeError/TypeError
    try {
      await this.openInternal(dbName);
    } catch (error) {
      // 打开失败：释放已获取的锁（避免锁泄漏阻塞其他标签页）
      if (this.dbLock) {
        try { await this.dbLock.release(); } catch { /* ignore */ }
        this.dbLock = null;
      }
      if (error instanceof DatabaseError) throw error;
      throw new DatabaseError(
        `Failed to open AriaEngine database "${dbName}"`,
        'ARIA_OPEN_ERROR',
        error,
      );
    }
  }

  /** open 内部实现（错误包装在 open 外层） */
  private async openInternal(dbName: string): Promise<void> {
    this.dbName = dbName;

    // v0.4.5: 多标签页独占锁（Web Locks）— 不支持的环境降级为无锁（文档注明）
    const lock = new DatabaseLock();
    this.dbLock = lock;
    const lockAcquired = await lock.acquire(dbName);
    if (!lockAcquired) {
      // eslint-disable-next-line no-console
      console.warn(
        `[AriaEngine] Web Locks API unavailable: no multi-tab protection for "${dbName}" ` +
        '(open the same database in multiple tabs may corrupt data)',
      );
    }

    // 1. 存储后端（可选全库加密包装）
    let baseBackend: IStorageBackend;
    if (this.config.storageBackend === 'opfs') {
      baseBackend = new OPFSBackend();
    } else {
      baseBackend = new MemoryBackend();
    }
    await baseBackend.open(dbName);

    // v0.4.5: encryption 配置 → 透明加密封装（密码错误/数据损坏在 open 或首次读取时暴露）
    if (this.config.encryption?.password) {
      this.backend = new EncryptedBackend(baseBackend, this.config.encryption.password);
    } else {
      this.backend = baseBackend;
      // 反向检测：库中存在密钥元数据但未提供密码 → 拒绝打开（避免密文被当明文解析成空库）
      if (await baseBackend.exists('__aria_keymeta')) {
        await baseBackend.close();
        throw new DatabaseError(
          'Database is encrypted: provide encryption.password to open it',
          'ARIA_ENCRYPT_REQUIRED',
        );
      }
    }
    await this.backend.open(dbName);

    // 2a. FileManager (PageIO 实现) + Buffer Pool
    this.fileManager = new FileManager(this.backend);
    await this.fileManager.init(dbName);
    this.bufferPool = new BufferPool(this.fileManager, this.config.bufferPoolPages);

    // 2. 构建 SSTableStore
    const sstableStore = this.createSSTableStore('main');

    // 3. 初始化主 LSM（PK 索引）
    this.lsm = new LSM({
      memtableSizeThreshold: this.config.memtableSizeThreshold,
      levelSizeMultiplier: this.config.levelSizeMultiplier,
      blockSize: this.config.pageSize,
      bloomBitsPerKey: this.config.bloomFilterBitsPerKey,
      // SSTable 缓存上限 = BufferPool 页数 × 页面大小（默认 256 页 ≈ 1MB 可控内存）
      cacheLimitBytes: this.config.bufferPoolPages * this.config.pageSize,
      sstableStore,
    });

    // 4. 初始化 WAL
    // v0.4.5: 分片式 WAL 存储（__wal_%06d.bin），序号内嵌记录字节流无需 count 键，
    // append 单文件原子写；空洞检测截断；兼容旧格式 __wal_N + __wal_count
    this.wal = new WAL(
      new SegmentedWALStore(this.backend),
      this.config.walEnabled,
      this.config.walSyncMode,
    );

    // 5. 恢复 Schema
    await this.loadSchemas();

    // v0.4.2-fix: 为 schema 中带 index/unique 标记的列重建二级索引 LSM。
    // 此前重开只恢复 schema 不恢复索引 LSM → 索引查询静默回退全表、
    // createIndex 因 colDef 已有标记直接 return → 索引永久缺失。
    // 索引数据已持久化在独立命名空间（sst_idx_* / meta），init() 直接加载。
    for (const [tableName, schema] of this.schemas) {
      const pkCol = this.tablePKs.get(tableName)!;
      for (const [colName, colDef] of Object.entries(schema.columns)) {
        if ((colDef.index || colDef.unique) && colName !== pkCol) {
          const idxKey = `${tableName}:idx:${colName}`;
          if (!this.secondaryIndexes.has(idxKey)) {
            const idxLsm = new LSM({
              memtableSizeThreshold: this.config.memtableSizeThreshold,
              levelSizeMultiplier: this.config.levelSizeMultiplier,
              blockSize: this.config.pageSize,
              bloomBitsPerKey: this.config.bloomFilterBitsPerKey,
              cacheLimitBytes: this.config.bufferPoolPages * this.config.pageSize,
              sstableStore: this.createSSTableStore(`idx_${tableName}_${colName}`),
            });
            await idxLsm.init();
            this.secondaryIndexes.set(idxKey, idxLsm);
          }
        }
      }
    }

    // 6. 初始化 LSM（加载 SSTable 元数据）
    await this.lsm.init();

    // 7. WAL 恢复（两阶段：先扫描事务边界，仅回放已提交事务）
    const committedTxns = new Set<number>();
    const allRecords: WALRecord[] = [];
    await this.wal.recover((r) => allRecords.push(r));
    // 第一遍：确定已提交事务
    for (const r of allRecords) {
      if (r.type === WALRecordType.COMMIT) committedTxns.add(r.txnId);
      if (r.type === WALRecordType.ROLLBACK) committedTxns.delete(r.txnId);
    }
    // 第二遍：仅应用 txnId==0（非事务）或已提交事务的数据
    for (const r of allRecords) {
      if (r.txnId === 0 || committedTxns.has(r.txnId)) {
        if (r.type === WALRecordType.DROP_TABLE) {
          // v0.3.3: DROP_TABLE 回放（异步：需预加载 SSTable 后清除残留数据）
          await this.applyDropTableRecovery(r.tableName);
        } else {
          this.applyWALRecord(r);
        }
      }
    }

    // v0.3.3: 恢复完成后将回放数据落盘并截断 WAL，
    // 避免每次重启重复回放 + WAL 无限膨胀
    if (allRecords.length > 0) {
      await this.lsm.flush();
      await this.wal.checkpoint();
      // v0.4.2-fix: WAL 回放只更新主 LSM，二级索引 LSM 未同步 →
      // 崩溃前最后一批写入的索引缺失，重开时索引查询丢行。
      // 恢复后全量重建所有表的二级索引（幂等）。
      for (const tableName of this.schemas.keys()) {
        await this.reindexTableInternal(tableName);
      }
    }

    // 8. Checkpoint Manager（接入 WAL 大小阈值）
    // v0.4.2-fix: 事务活跃时 checkpoint 不得截断 WAL —
    // 否则 BEGIN/INSERT 记录被截断，COMMIT 后崩溃恢复丢失整个事务数据
    this.checkpointManager = new CheckpointManager(
      this.lsm,
      {
        checkpoint: async () => {
          if (this.currentTxnId) return;
          await this.wal.checkpoint();
        },
        flush: async () => {
          if (this.currentTxnId) return;
          await this.wal.flush();
        },
        getBufferedBytes: () => this.wal.getBufferedBytes(),
        getBufferedCount: () => this.wal.getBufferedCount(),
      } as unknown as WAL,
      { flushAll: async () => { await this.lsm.flush(); } } as any,
      this.config.checkpointInterval,
      this.config.walSizeThreshold,
    );

    this.opened = true;
  }

  async close(): Promise<void> {
    if (!this.opened) return;
    await this.persistSchemas();
    await this.lsm.flush();
    // v0.4.2-fix: 同步落盘全部二级索引 LSM — 此前只 flush 主 LSM，
    // 优雅关闭后索引 memtable 未落盘 → 重开索引为空 → 索引查询返回空结果
    for (const idxLsm of this.secondaryIndexes.values()) {
      await idxLsm.flush();
    }
    // v0.4.5: 页面化存储 — 落盘全部脏页（save 已逐页落盘，此处兜底）
    await this.bufferPool.flushAll();
    await this.wal.flush();
    // v0.4.2-fix: close 前 checkpoint（截断 WAL）—
    // 此前只 flush 不截断，下次打开会重放全部历史 WAL 记录（含已落盘 SSTable 的数据），
    // 重复解析/重复 put 拖慢启动，并与恢复后 flush+checkpoint 竞争放大丢数据
    await this.wal.checkpoint();
    await this.backend.close();
    // v0.4.5: 释放多标签页独占锁（等待锁真正归还）
    if (this.dbLock) {
      await this.dbLock.release();
      this.dbLock = null;
    }
    // v0.4.2-fix: 清空运行期状态（此前 close 后 mvcc/txn 残留，
    // 重开时 beginTransaction 报 TX_ACTIVE 或读到陈旧快照）
    this.schemas.clear();
    this.tablePKs.clear();
    this.secondaryIndexes.clear();
    this.mvcc = new MVCCManager();
    this.currentTxnId = null;
    this.txnSnapshot = null;
    this.savepoints.clear();
    this.opCounter = 0;
    this.opened = false;
  }

  /**
   * v0.4.2-fix: 崩溃恢复/自愈 — 校验并移除损坏 SSTable、截断 WAL、重建二级索引。
   * v0.4.5 增强：清理 OPFS 残留临时文件、清理孤儿页面（meta 未引用的 pg_ 文件）。
   * 应用层检测到异常后调用，无需删库重建。
   */
  async repair(): Promise<void> {
    this.ensureOpen();
    // v0.6.0-fix: 先清页面缓存再校验 — 缓存中的"完好页面"会掩盖磁盘损坏
    await this.bufferPool.clear();
    // 1. 校验全部 SSTable，移除残缺项（打开时已做一次，此处兜底运行期损坏）
    const removed = await this.lsm.validateAll();
    // 2. 将 WAL 残留数据落盘并截断，避免无限重放（含空洞截断落地）
    await this.lsm.flush();
    await this.wal.checkpoint();
    // 3. 重建所有表的二级索引（修复索引与主数据不一致）
    for (const tableName of this.schemas.keys()) {
      await this.reindexTable(tableName);
    }
    // v0.4.5: 4. 清理 OPFS 残留临时文件（.crswap/.tmp）
    const backendAny = this.backend as unknown as { cleanupStaleFiles?: () => Promise<void> };
    if (typeof backendAny.cleanupStaleFiles === 'function') {
      try { await backendAny.cleanupStaleFiles(); } catch { /* 清理失败不阻塞 */ }
    }
    // v0.4.5: 5. 清理孤儿页面（所有 LSM 命名空间 meta 均未引用的 pg_ 文件）
    await this.cleanupOrphanPages();
    if (removed > 0) {
      // eslint-disable-next-line no-console
      console.warn(`[AriaEngine] repair: removed ${removed} corrupted SSTable(s)`);
    }
  }

  /**
   * v0.4.5: 清理孤儿页面 — 扫描全部 pg_* 文件，未被任何 LSM 命名空间 meta 引用的删除。
   * 孤儿页面来自：崩溃中断的 compaction/删除流程（旧 SSTable 页面残留）。
   */
  private async cleanupOrphanPages(): Promise<void> {
    const keys = await this.backend.listKeys();
    const pgKeys = keys.filter((k) => /^pg_\d+$/.test(k));
    if (pgKeys.length === 0) return;

    const used = new Set<number>();
    const collectMeta = async (ns: string): Promise<void> => {
      const META_KEY = ns === 'main' ? '__aria_lsm_meta' : `__aria_lsm_meta_${ns}`;
      const raw = await this.backend.read(META_KEY);
      if (!raw) return;
      try {
        const metas = JSON.parse(new TextDecoder().decode(raw)) as SSTableMeta[];
        for (const m of metas) {
          if (m.pageIds) {
            for (const pid of m.pageIds) used.add(pid);
          }
        }
      } catch { /* 损坏的 meta 忽略（validateAll 已处理） */ }
    };

    await collectMeta('main');
    // 收集全部二级索引命名空间
    for (const [tableName, schema] of this.schemas) {
      for (const [colName, colDef] of Object.entries(schema.columns)) {
        if (colDef.index || colDef.unique) {
          await collectMeta(`idx_${tableName}_${colName}`);
        }
      }
    }

    const orphanIds = pgKeys
      .map((k) => Number(k.slice('pg_'.length)))
      .filter((pid) => !used.has(pid));
    if (orphanIds.length > 0) {
      await this.backend.deleteMany(orphanIds.map((pid) => `pg_${pid}`));
    }
  }

  /**
   * v0.4.1: 重置数据库 — 清空全部数据与表结构（演示页刷新/重新初始化用）。
   * 清空存储后端、LSM、WAL、MVCC 与二级索引，后续可继续使用本实例。
   */
  async clearAll(): Promise<void> {
    this.ensureOpen();
    // 清空存储后端（页面文件 / WAL 记录 / schema 记录 / 元数据）
    await this.backend.clear();
    // v0.4.5: 清空页面缓存与页面 ID 分配状态
    await this.bufferPool.clear();
    await this.fileManager.clearAll();
    this.schemas.clear();
    this.tablePKs.clear();
    this.secondaryIndexes.clear();
    this.lsm.clear();
    this.mvcc = new MVCCManager();
    this.currentTxnId = null;
    this.txnSnapshot = null;
    this.savepoints.clear();
    this.opCounter = 0;
    // 持久化空 schema（防止旧 schema 记录残留）
    await this.persistSchemas();
    // 重置 WAL 状态（backend.clear 已清记录，同步内存计数）
    await this.wal.checkpoint();
  }

  isOpen(): boolean { return this.opened; }

  // ---- v0.4.2-fix: 库内元数据（迁移版本持久化用） ----

  async getMeta(key: string): Promise<string | null> {
    const raw = await this.backend.read(`__meta_${key}`);
    return raw ? new TextDecoder().decode(raw) : null;
  }

  async setMeta(key: string, value: string): Promise<void> {
    await this.backend.write(`__meta_${key}`, new TextEncoder().encode(value).buffer);
  }

  // =======================================================================
  //  表管理
  // =======================================================================

  async createTable(schema: TableSchema): Promise<void> {
    this.ensureOpen();
    // v0.4.2-fix: Aria 事务中 DDL 显式拒绝（事务快照只覆盖行数据，
    // 结构变更无法回滚；Memory/IndexedDB 引擎快照可回滚，行为不一致 → 明确报错而非静默）
    this.ensureNoDDLInTransaction('CREATE TABLE');
    if (this.schemas.has(schema.name)) {
      throw new DatabaseError(`Table "${schema.name}" already exists`, 'TABLE_EXISTS');
    }

    this.schemas.set(schema.name, schema);
    this.tablePKs.set(schema.name, this.getPK(schema));

    // 为索引列创建二级索引 LSM（每个索引使用独立命名空间的 SSTableStore，避免 id/meta 冲突）
    // v0.3.3: 主键列不建冗余二级索引（主 LSM 本身就是 PK 索引，范围查询走前缀扫描）
    for (const [colName, colDef] of Object.entries(schema.columns)) {
      if (colDef.index || colDef.unique) {
        const idxKey = `${schema.name}:idx:${colName}`;
        if (!this.secondaryIndexes.has(idxKey)) {
          const idxLsm = new LSM({
            memtableSizeThreshold: this.config.memtableSizeThreshold,
            levelSizeMultiplier: this.config.levelSizeMultiplier,
            blockSize: this.config.pageSize,
            bloomBitsPerKey: this.config.bloomFilterBitsPerKey,
            cacheLimitBytes: this.config.bufferPoolPages * this.config.pageSize,
            sstableStore: this.createSSTableStore(`idx_${schema.name}_${colName}`),
          });
          await idxLsm.init();
          this.secondaryIndexes.set(idxKey, idxLsm);
        }
      }
    }

    await this.persistSchemas();

    await this.wal.append({
      type: WALRecordType.CREATE_TABLE,
      txnId: 0,
      tableName: schema.name,
      key: '',
      data: { schema: JSON.stringify(schema) } as unknown as Record<string, unknown>,
    });
  }

  async dropTable(tableName: string): Promise<void> {
    this.ensureOpen();
    this.ensureNoDDLInTransaction('DROP TABLE');
    this.ensureTable(tableName);

    // 删除表中所有行
    const rows = await this.getAllRows(tableName);
    for (const row of rows) {
      const pkCol = this.tablePKs.get(tableName)!;
      this.lsm.delete(`${tableName}:${row[pkCol]}`);
    }

    // v0.4.2-fix: 清理该表的全部二级索引 LSM 与持久化文件 —
    // 此前残留孤儿索引，重建同名表后旧索引数据污染新表（索引查询返回错误行）
    await this.cleanupTableIndexes(tableName);

    this.schemas.delete(tableName);
    this.tablePKs.delete(tableName);
    await this.persistSchemas();

    await this.wal.append({
      type: WALRecordType.DROP_TABLE,
      txnId: 0,
      tableName,
      key: '',
    });
  }

  /**
   * v0.4.2-fix: 清理指定表的全部二级索引 LSM（内存 + 存储文件 + meta）。
   * dropTable / DROP_TABLE 恢复 / alterTable DROP 索引列 共用。
   */
  private async cleanupTableIndexes(tableName: string): Promise<void> {
    const prefix = `${tableName}:idx:`;
    const toDelete: string[] = [];
    for (const [idxKey, idxLsm] of this.secondaryIndexes) {
      if (!idxKey.startsWith(prefix)) continue;
      toDelete.push(idxKey);
      try {
        await idxLsm.clear();
      } catch { /* 清理失败不阻塞 */ }
    }
    for (const idxKey of toDelete) {
      this.secondaryIndexes.delete(idxKey);
    }
  }

  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;
  }

  // =======================================================================
  //  CRUD
  // =======================================================================

  async insert(tableName: string, rows: Record<string, unknown>[]): Promise<string[]> {
    this.ensureOpen();
    this.ensureTable(tableName);

    const schema = this.schemas.get(tableName)!;
    const pkCol = this.tablePKs.get(tableName)!;
    const pks: string[] = [];
    // v0.3.1: 批量 WAL 写入（组提交），一次 insert 合并为一次落盘
    const walRecords: Omit<WALRecord, 'lsn' | 'checksum'>[] = [];

    for (const row of rows) {
      const validated = this.validateRow(schema, row);
      const pkValue = String(validated[pkCol]);
      const key = `${tableName}:${pkValue}`;

      // Check duplicate in LSM + transaction snapshot
      await this.lsm.prefetchKeys([key]);
      const existing = this.currentTxnId
        ? (this.txnSnapshot?.get(key) ?? this.lsm.get(key))
        : this.lsm.get(key);
      if (existing && !(existing as unknown as Record<string, unknown>).__txn_deleted) {
        throw new DatabaseError(
          `Duplicate primary key "${pkValue}" in table "${tableName}"`,
          'DUPLICATE_KEY',
        );
      }

      if (this.currentTxnId && this.txnSnapshot) {
        // Within transaction: buffer to snapshot + MVCC version chain
        this.txnSnapshot.set(key, validated);
        this.mvcc.writeVersion(tableName, pkValue, validated, this.currentTxnId);
      } else {
        // Direct write to LSM (PK index)
        this.lsm.put(key, validated);
      }

      // 更新二级索引
      this.updateSecondaryIndexes(tableName, pkValue, validated, null);

      pks.push(pkValue);

      walRecords.push({
        type: WALRecordType.INSERT,
        txnId: this.currentTxnId ?? 0,
        tableName,
        key: pkValue,
        data: validated,
      });
    }

    await this.wal.appendBatch(walRecords);

    this.opCounter += rows.length;
    this.checkMemoryBudget();
    await this.checkpointManager.tick();
    this.tryGC();
    return pks;
  }

  async find(tableName: string, query: QueryPlan): Promise<Record<string, unknown>[]> {
    this.ensureOpen();
    this.ensureTable(tableName);

    let rows: Record<string, unknown>[];

    // Try index lookup
    const fastPath = await this.tryIndexLookup(tableName, query);
    if (fastPath !== null) {
      rows = fastPath;
    } else {
      rows = await this.getAllRows(tableName);
    }

    // v0.3.3: 事务内合并未提交快照（统一在 mergeTxnSnapshot 处理）
    rows = this.mergeTxnSnapshot(tableName, rows);

    // WHERE filter
    if (query.where && Object.keys(query.where).length > 0) {
      rows = rows.filter((row) => matchWhere(row, query.where!));
    }

    // ORDER
    if (query.orderBy && query.orderBy.length > 0) {
      rows = applyOrderBy(rows, query.orderBy);
    }

    // LIMIT/OFFSET
    const offset = query.offset ?? 0;
    const limit = query.limit ?? rows.length;
    rows = rows.slice(offset, offset + limit);

    // Column projection
    if (query.columns && query.columns.length > 0 && query.columns[0] !== '*') {
      rows = rows.map((row) => projectColumns(row, query.columns!));
    }

    // 查询完成，回收查询期间的临时缓存超限
    this.trimAllCaches();

    return rows;
  }

  async update(
    tableName: string,
    query: QueryPlan,
    updates: Record<string, unknown>,
  ): Promise<number> {
    this.ensureOpen();
    this.ensureTable(tableName);
    const schema = this.schemas.get(tableName)!;
    const rows = await this.getAllRows(tableName);
    let count = 0;
    // v0.3.1: 批量 WAL 写入（组提交）
    const walRecords: Omit<import('./types').WALRecord, 'lsn' | 'checksum'>[] = [];
    // v0.4.2-fix: ON UPDATE 级联环路保护
    const visited = new Set<string>();

    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)) {
        const updated = { ...row, ...updates };
        this.validateRow(schema, updated);

        // v0.4.2-fix: 支持更新主键 — 删除旧键 + 落新键 + WAL 两条记录
        const newPk = String(updated[pkCol]);
        const pkChanged = newPk !== String(row[pkCol]);

        if (pkChanged) {
          // ON UPDATE 外键级联（RESTRICT 抛错 / CASCADE / SET NULL）
          await this.applyForeignKeyUpdateRules(
            tableName, String(row[pkCol]), newPk, walRecords, visited,
          );
        }

        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++;

        if (pkChanged) {
          walRecords.push({
            type: WALRecordType.DELETE,
            txnId: this.currentTxnId ?? 0,
            tableName,
            key: String(row[pkCol]),
          });
        }
        walRecords.push({
          type: WALRecordType.UPDATE,
          txnId: this.currentTxnId ?? 0,
          tableName,
          key: newPk,
          data: updated,
        });

        // 更新二级索引（主键变更时旧索引条目一并清理）
        this.updateSecondaryIndexes(tableName, newPk, updated, pkChanged ? row : null);
      }
    }

    await this.wal.appendBatch(walRecords);

    this.opCounter += count;
    await this.checkpointManager.tick();
    this.trimAllCaches();
    return count;
  }

  /**
   * v0.4.2-fix: ON UPDATE 外键级联 — 主键 oldPk → newPk 时处理引用表。
   * RESTRICT 抛错 / CASCADE 更新 FK / SET NULL 置空（含索引与 WAL 记录）。
   * 两阶段：先全量 RESTRICT 检查，再执行级联。
   */
  private async applyForeignKeyUpdateRules(
    tableName: string,
    oldPk: string,
    newPk: string,
    walRecords: Omit<import('./types').WALRecord, 'lsn' | 'checksum'>[],
    visited: Set<string>,
  ): Promise<void> {
    const visitKey = `${tableName}:${oldPk}`;
    if (visited.has(visitKey)) return;
    visited.add(visitKey);

    // 阶段 1: RESTRICT 检查
    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') continue;
        const refRows = await this.getAllRows(refTableName);
        if (refRows.some((r) => String(r[colName]) === oldPk)) {
          throw new DatabaseError(
            `Cannot update "${tableName}" key "${oldPk}": foreign key "${colName}" in "${refTableName}" has dependent rows`,
            'FOREIGN_KEY_VIOLATION',
          );
        }
      }
    }

    // 阶段 2: CASCADE / SET NULL
    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 !== 'CASCADE' && colDef.onUpdate !== 'SET NULL') continue;
        const refRows = await this.getAllRows(refTableName);
        for (const refRow of refRows) {
          if (String(refRow[colName]) !== oldPk) continue;
          const refPkCol = this.tablePKs.get(refTableName)!;
          const refPk = String(refRow[refPkCol]);
          const updatedRef = { ...refRow, [colName]: colDef.onUpdate === 'CASCADE' ? newPk : null };
          const refKey = `${refTableName}:${refPk}`;
          if (this.currentTxnId && this.txnSnapshot) {
            this.txnSnapshot.set(refKey, updatedRef);
            this.mvcc.writeVersion(refTableName, refPk, updatedRef, this.currentTxnId);
          } else {
            this.lsm.put(refKey, updatedRef);
          }
          this.updateSecondaryIndexes(refTableName, refPk, updatedRef, refRow);
          walRecords.push({
            type: WALRecordType.UPDATE,
            txnId: this.currentTxnId ?? 0,
            tableName: refTableName,
            key: refPk,
            data: updatedRef,
          });
        }
      }
    }
  }

  async delete(tableName: string, query: QueryPlan): Promise<number> {
    this.ensureOpen();
    this.ensureTable(tableName);

    const rows = await this.getAllRows(tableName);
    let count = 0;
    // v0.3.1: 批量 WAL 写入（组提交）
    const walRecords: Omit<import('./types').WALRecord, 'lsn' | 'checksum'>[] = [];
    // v0.4.1: 外键级联（环路保护）
    const visited = new Set<string>();

    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)) {
        // v0.4.1: 外键规则（RESTRICT 抛错 / CASCADE 递归删 / SET NULL 置空）
        count += await this.applyForeignKeyRules(tableName, String(row[pkCol]), walRecords, visited);
        if (this.currentTxnId && this.txnSnapshot) {
          // Buffer delete in snapshot + MVCC tombstone
          this.txnSnapshot.set(key, { __txn_deleted: true } as unknown as Record<string, unknown>);
          this.mvcc.deleteVersion(tableName, String(row[pkCol]), this.currentTxnId);
        } else {
          this.lsm.delete(key);
        }
        count++;

        walRecords.push({
          type: WALRecordType.DELETE,
          txnId: this.currentTxnId ?? 0,
          tableName,
          key: String(row[pkCol]),
        });

        // 移除二级索引
        this.updateSecondaryIndexes(tableName, String(row[pkCol]), null, row);
      }
    }

    await this.wal.appendBatch(walRecords);

    this.opCounter += count;
    await this.checkpointManager.tick();
    this.trimAllCaches();
    return count;
  }

  /**
   * v0.4.1: 外键级联规则 — 对齐 MemoryEngine.cascadeDelete 行为。
   * 删除 tableName 主键为 pkValue 的行前，检查引用它的所有表：
   * - RESTRICT: 存在引用行 → 抛 FOREIGN_KEY_VIOLATION
   * - CASCADE: 递归删除引用行（含索引/WAL）
   * - SET NULL: 引用行外键列置 null（含索引/WAL）
   * @returns 级联影响的行数（CASCADE 删除行数 + SET NULL 更新行数）
   */
  private async applyForeignKeyRules(
    tableName: string,
    pkValue: string,
    walRecords: Omit<import('./types').WALRecord, 'lsn' | 'checksum'>[],
    visited: Set<string>,
  ): Promise<number> {
    let total = 0;
    const visitKey = `${tableName}:${pkValue}`;
    if (visited.has(visitKey)) return 0;
    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 refRows = await this.getAllRows(refTableName);
        const matched = refRows.filter((r) => String(r[colName]) === pkValue);

        if (colDef.onDelete === 'RESTRICT' && matched.length > 0) {
          throw new DatabaseError(
            `Cannot delete from "${tableName}": foreign key "${colName}" in "${refTableName}" has dependent rows`,
            'FOREIGN_KEY_VIOLATION',
          );
        }

        if (colDef.onDelete === 'CASCADE') {
          const refPkCol = this.tablePKs.get(refTableName)!;
          for (const refRow of matched) {
            const refPk = String(refRow[refPkCol]);
            // 递归级联（先处理更深层引用）
            total += await this.applyForeignKeyRules(refTableName, refPk, walRecords, visited);
            // 删除引用行
            const refKey = `${refTableName}:${refPk}`;
            if (this.currentTxnId && this.txnSnapshot) {
              this.txnSnapshot.set(refKey, { __txn_deleted: true } as unknown as Record<string, unknown>);
              this.mvcc.deleteVersion(refTableName, refPk, this.currentTxnId);
            } else {
              this.lsm.delete(refKey);
            }
            this.updateSecondaryIndexes(refTableName, refPk, null, refRow);
            walRecords.push({
              type: WALRecordType.DELETE,
              txnId: this.currentTxnId ?? 0,
              tableName: refTableName,
              key: refPk,
            });
            total++;
          }
        } else if (colDef.onDelete === 'SET NULL') {
          const refPkCol = this.tablePKs.get(refTableName)!;
          for (const refRow of matched) {
            const refPk = String(refRow[refPkCol]);
            const updated = { ...refRow, [colName]: null };
            const refKey = `${refTableName}:${refPk}`;
            if (this.currentTxnId && this.txnSnapshot) {
              this.txnSnapshot.set(refKey, updated);
              this.mvcc.writeVersion(refTableName, refPk, updated, this.currentTxnId);
            } else {
              this.lsm.put(refKey, updated);
            }
            this.updateSecondaryIndexes(refTableName, refPk, updated, refRow);
            walRecords.push({
              type: WALRecordType.UPDATE,
              txnId: this.currentTxnId ?? 0,
              tableName: refTableName,
              key: refPk,
              data: updated,
            });
            // 对齐 Memory 语义：SET NULL 不影响返回的删除行数
          }
        }
      }
    }
    return total;
  }

  /**
   * v0.4.0: 流式查询 — 逐行回调，不物化结果数组。
   * 全表路径走 LSM rangeScanLazy 惰性扫描；索引等值/范围路径复用 tryIndexLookup。
   * 事务中回退物化（快照合并需要全量行集）。
   */
  async findStream(
    tableName: string,
    query: QueryPlan,
    onRow: (row: Record<string, unknown>) => void,
  ): Promise<number> {
    this.ensureOpen();
    this.ensureTable(tableName);

    const hasWhere = !!(query.where && Object.keys(query.where).length > 0);
    const project = query.columns && query.columns.length > 0 && query.columns[0] !== '*'
      ? (row: Record<string, unknown>) => projectColumns(row, query.columns!)
      : null;
    const limit = query.limit ?? Infinity;
    const offset = query.offset ?? 0;
    const pkCol = this.tablePKs.get(tableName)!;
    const prefix = `${tableName}:`;
    let count = 0;
    let skipped = 0;

    const emit = (row: Record<string, unknown>): boolean => {
      if (hasWhere && !matchWhere(row, query.where!)) return true;
      if (skipped < offset) { skipped++; return true; }
      onRow(project ? project(row) : row);
      count++;
      return count < limit;
    };

    if (this.currentTxnId && this.txnSnapshot) {
      // 事务中：物化后逐行回调（快照合并需要全量行集）
      const rows = await this.find(tableName, { ...query, orderBy: undefined, limit: undefined, offset: undefined });
      for (const row of rows) {
        onRow(project ? project(row) : row);
      }
      return rows.length;
    }

    // 索引路径：等值/范围查找（结果行已过滤，直接回调）
    const fastPath = await this.tryIndexLookup(tableName, query);
    if (fastPath !== null) {
      for (const row of fastPath) {
        if (!emit(row)) break;
      }
      return count;
    }

    // 全表惰性扫描（含 WHERE 过滤，不物化）
    await this.lsm.prefetchRange(prefix, `${prefix}\uffff`);
    this.lsm.rangeScanLazy(prefix, `${prefix}\uffff`, (key, value) => {
      if (count >= limit) return;
      const row = { ...value };
      row[pkCol] = key.slice(prefix.length);
      emit(row);
    });
    return count;
  }

  async count(tableName: string, query?: QueryPlan): Promise<number> {
    this.ensureOpen();
    this.ensureTable(tableName);
    const rows = await this.getAllRows(tableName);
    this.trimAllCaches();
    if (!query?.where || Object.keys(query.where).length === 0) return rows.length;
    return rows.filter((row) => matchWhere(row, query.where!)).length;
  }

  async clear(tableName: string): Promise<void> {
    this.ensureOpen();
    this.ensureTable(tableName);
    const rows = await this.getAllRows(tableName);
    // v0.3.3: 事务内清空走快照（删除标记），提交时生效；并写入 WAL
    const walRecords: Omit<import('./types').WALRecord, 'lsn' | 'checksum'>[] = [];
    for (const row of rows) {
      const pkCol = this.tablePKs.get(tableName)!;
      const key = `${tableName}:${row[pkCol]}`;
      if (this.currentTxnId && this.txnSnapshot) {
        this.txnSnapshot.set(key, { __txn_deleted: true } as unknown as Record<string, unknown>);
        this.mvcc.deleteVersion(tableName, String(row[pkCol]), this.currentTxnId);
      } else {
        this.lsm.delete(key);
      }
      walRecords.push({
        type: WALRecordType.DELETE,
        txnId: this.currentTxnId ?? 0,
        tableName,
        key: String(row[pkCol]),
      });
      // 移除二级索引
      this.updateSecondaryIndexes(tableName, String(row[pkCol]), null, row);
    }
    await this.wal.appendBatch(walRecords);
    this.opCounter += rows.length;
    await this.checkpointManager.tick();
    this.tryGC();
  }

  // ---- ALTER TABLE（v0.4.1） ----

  /**
   * v0.4.1: ALTER TABLE — 结构变更真正生效于存储：
   * - ADD: 持久化 schema（persistSchemas），行无需修改
   * - DROP: 持久化 schema + 遍历主 LSM 重写所有行（移除该列键）+ WAL UPDATE 记录
   * （通用路径 getTableSchema 返回副本，Executor 的引用修改对 Aria 无效）
   */
  async alterTable(
    tableName: string,
    action: 'ADD' | 'DROP',
    column: import('../../constants').ColumnDef & { name: string },
  ): Promise<void> {
    this.ensureOpen();
    this.ensureNoDDLInTransaction('ALTER TABLE');
    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;
      await this.persistSchemas();
      return;
    }

    // DROP
    if (!schema.columns[column.name]) {
      throw new DatabaseError(`Column "${column.name}" does not exist in table "${tableName}"`, 'COLUMN_NOT_FOUND');
    }
    // v0.4.2-fix: 被删列是索引列 → 先清理索引 LSM（残留会导致后续同名列索引脏数据）
    if (schema.columns[column.name].index || schema.columns[column.name].unique) {
      const idxKey = `${tableName}:idx:${column.name}`;
      const idxLsm = this.secondaryIndexes.get(idxKey);
      if (idxLsm) {
        try {
          await idxLsm.clear();
        } catch { /* 清理失败不阻塞 */ }
        this.secondaryIndexes.delete(idxKey);
      }
    }
    delete schema.columns[column.name];
    await this.persistSchemas();

    // 重写主 LSM：移除所有行的该列键（find 副本无法就地删除，必须重写存储）
    const prefix = `${tableName}:`;
    const endKey = `${prefix}\uffff`;
    await this.lsm.prefetchRange(prefix, endKey);
    const entries = this.lsm.rangeScan(prefix, endKey);
    const walRecords: Omit<import('./types').WALRecord, 'lsn' | 'checksum'>[] = [];
    for (const [key, value] of entries) {
      if (!(column.name in value)) continue;
      const updated = { ...value };
      delete updated[column.name];
      this.lsm.put(key, updated);
      // 二级索引列被删时同步清理索引
      const pk = key.slice(prefix.length);
      this.updateSecondaryIndexes(tableName, pk, updated, value);
      walRecords.push({
        type: WALRecordType.UPDATE,
        txnId: this.currentTxnId ?? 0,
        tableName,
        key: pk,
        data: updated,
      });
    }
    await this.wal.appendBatch(walRecords);
    this.opCounter += walRecords.length;
    await this.checkpointManager.tick();
    this.trimAllCaches();
  }

  async createIndex(tableName: string, column: string, unique?: boolean): Promise<void> {
    this.ensureOpen();
    this.ensureNoDDLInTransaction('CREATE INDEX');
    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');
    const idxKey = `${tableName}:idx:${column}`;
    // v0.4.2-fix: 以索引 LSM 是否已建为准（schema 标记可能因重启恢复而存在，
    // 但索引 LSM 未恢复 → 此前静默 return 导致索引永久缺失）
    if (this.secondaryIndexes.has(idxKey)) return;
    colDef.index = true;
    if (unique) colDef.unique = true;

    const idxLsm = new LSM({
      memtableSizeThreshold: this.config.memtableSizeThreshold,
      levelSizeMultiplier: this.config.levelSizeMultiplier,
      blockSize: this.config.pageSize,
      bloomBitsPerKey: this.config.bloomFilterBitsPerKey,
      cacheLimitBytes: this.config.bufferPoolPages * this.config.pageSize,
      sstableStore: this.createSSTableStore(`idx_${tableName}_${column}`),
    });
    await idxLsm.init();
    this.secondaryIndexes.set(idxKey, idxLsm);

    // 从主 LSM 重建索引数据
    const pkCol = this.tablePKs.get(tableName)!;
    const rows = await this.getAllRows(tableName);
    for (const row of rows) {
      const value = row[column];
      if (value !== undefined && value !== null) {
        idxLsm.put(`${String(value)}:${row[pkCol]}`, { pk: row[pkCol] });
      }
    }
    await idxLsm.flush();
    await this.persistSchemas();
  }

  async dropIndex(tableName: string, column: string, _indexName?: string): Promise<void> {
    this.ensureOpen();
    this.ensureNoDDLInTransaction('DROP INDEX');
    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');
    // 主键索引不可删除（PK 查找依赖主 LSM）
    if (colDef.primaryKey) {
      throw new DatabaseError(`Cannot drop primary key index on column "${column}"`, 'NOT_SUPPORTED');
    }
    // v0.4.1: DROP 不存在的索引应报错（此前静默成功）
    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');
    }
    colDef.index = false;
    colDef.unique = false;

    const idxKey = `${tableName}:idx:${column}`;
    const idxLsm = this.secondaryIndexes.get(idxKey);
    if (idxLsm) {
      await idxLsm.clear();
      this.secondaryIndexes.delete(idxKey);
    }
    await this.persistSchemas();
  }

  // =======================================================================
  //  事务
  // =======================================================================

  async beginTransaction(): Promise<void> {
    if (this.currentTxnId) throw new DatabaseError('Transaction already in progress', 'TX_ACTIVE');
    this.currentTxnId = this.mvcc.beginTransaction();
    this.txnSnapshot = new Map();

    await this.wal.append({
      type: WALRecordType.BEGIN,
      txnId: this.currentTxnId,
      tableName: '',
      key: '',
    });
  }

  async commitTransaction(): Promise<void> {
    if (!this.currentTxnId) throw new DatabaseError('No active transaction', 'TX_NONE');

    // v0.4.3-fix: 先持久化 WAL COMMIT，再合并快照到 LSM —
    // 崩溃在 WAL 提交后、快照合并前：恢复时 WAL 重放数据，重启一致；
    // 崩溃在 WAL 提交前：commitTransaction 尚未返回，事务视为未提交（可回滚）
    await this.wal.append({
      type: WALRecordType.COMMIT,
      txnId: this.currentTxnId,
      tableName: '',
      key: '',
    });
    await this.wal.flush();

    if (this.txnSnapshot) {
      for (const [key, value] of this.txnSnapshot) {
        if ((value as unknown as Record<string, unknown>).__txn_deleted) {
          this.lsm.delete(key);
        } else {
          this.lsm.put(key, value);
        }
      }
    }

    this.mvcc.commitTransaction(this.currentTxnId);

    this.currentTxnId = null;
    this.txnSnapshot = null;
  }

  async rollbackTransaction(): Promise<void> {
    if (!this.currentTxnId) throw new DatabaseError('No active transaction', 'TX_NONE');

    // v0.3.3: 记录事务涉及的表（用于回滚后重建索引，消除索引残留）
    const affectedTables = new Set<string>();
    if (this.txnSnapshot) {
      for (const key of this.txnSnapshot.keys()) {
        const idx = key.indexOf(':');
        if (idx > 0) affectedTables.add(key.slice(0, idx));
      }
    }

    this.mvcc.rollbackTransaction(this.currentTxnId);
    this.txnSnapshot = null;

    await this.wal.append({
      type: WALRecordType.ROLLBACK,
      txnId: this.currentTxnId,
      tableName: '',
      key: '',
    });

    this.currentTxnId = null;

    // v0.3.3: 事务内直接写入了二级索引 LSM，回滚后全量重建受影响表的索引
    for (const tableName of affectedTables) {
      if (this.schemas.has(tableName)) {
        await this.reindexTable(tableName);
      }
    }
  }

  // ---- Savepoint 嵌套事务 ----

  private savepoints: Map<string, { txnId: number; snapshot: Map<string, Record<string, unknown>> | null }> = new Map();

  async savepoint(name: string): Promise<void> {
    if (!this.currentTxnId) throw new DatabaseError('No active transaction for savepoint', 'TX_NONE');
    if (this.savepoints.has(name)) throw new DatabaseError(`Savepoint "${name}" already exists`, 'SAVEPOINT_EXISTS');
    // 保存当前事务快照
    this.savepoints.set(name, {
      txnId: this.currentTxnId,
      snapshot: this.txnSnapshot ? new Map(this.txnSnapshot) : null,
    });
  }

  async rollbackToSavepoint(name: string): Promise<void> {
    const sp = this.savepoints.get(name);
    if (!sp) throw new DatabaseError(`Savepoint "${name}" not found`, 'SAVEPOINT_NOT_FOUND');
    // 恢复到 savepoint 时的快照
    this.txnSnapshot = sp.snapshot ? new Map(sp.snapshot) : null;
    // v0.3.3: 清理该事务在 MVCC 版本链中的全部记录（快照已含正确数据，
    // 版本链仅作 undo 记录，清空后 commit 时 LSM 写入与快照保持一致）
    this.mvcc.discardVersions(this.currentTxnId!);
    // 清除此 savepoint 之后的所有 savepoint
    let found = false;
    for (const [k] of this.savepoints) {
      if (k === name) { found = true; continue; }
      if (found) this.savepoints.delete(k);
    }
  }

  async releaseSavepoint(name: string): Promise<void> {
    if (!this.savepoints.has(name)) throw new DatabaseError(`Savepoint "${name}" not found`, 'SAVEPOINT_NOT_FOUND');
    this.savepoints.delete(name);
  }

  // ---- 在线备份 ----

  async backup(): Promise<Record<string, Record<string, unknown>[]>> {
    this.ensureOpen();
    const result: Record<string, Record<string, unknown>[]> = {};
    for (const tableName of this.schemas.keys()) {
      result[tableName] = await this.getAllRows(tableName);
    }
    return result;
  }

  // =======================================================================
  //  内部
  // =======================================================================

  private async getAllRows(tableName: string): Promise<Record<string, unknown>[]> {
    const pkCol = this.tablePKs.get(tableName)!;
    const prefix = `${tableName}:`;
    // 预加载范围内涉及的 SSTable，避免 rangeScan 时缓存未命中静默丢数据
    await this.lsm.prefetchRange(prefix, `${prefix}\uffff`);
    const entries = this.lsm.rangeScan(prefix, `${prefix}\uffff`);
    const rows = entries.map(([key, value]) => {
      const row = { ...value };
      row[pkCol] = key.slice(prefix.length);
      return row;
    });
    // v0.3.3: 事务内合并未提交快照（update/delete/count/clear 也能看到本事务的写入）
    return this.mergeTxnSnapshot(tableName, rows);
  }

  /**
   * v0.3.3: 将事务未提交快照的变更合并到行列表（新增/更新/删除标记）。
   * 幂等操作：行已是最新时不重复修改。
   */
  private mergeTxnSnapshot(tableName: string, rows: Record<string, unknown>[]): Record<string, unknown>[] {
    if (!this.currentTxnId || !this.txnSnapshot) return rows;
    const pkCol = this.tablePKs.get(tableName)!;
    const prefix = `${tableName}:`;
    for (const [key, value] of this.txnSnapshot) {
      if (!key.startsWith(prefix)) continue;
      const pk = key.slice(prefix.length);
      const del = (value as unknown as Record<string, unknown>).__txn_deleted;
      const idx = rows.findIndex((r) => r[pkCol] === pk);
      if (del) {
        if (idx >= 0) rows.splice(idx, 1);
      } else {
        const row = { ...value, [pkCol]: pk };
        if (idx >= 0) rows[idx] = row;
        else rows.push(row);
      }
    }
    return rows;
  }

  private getPK(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> {
    const validated: Record<string, unknown> = {};
    for (const [colName, colDef] of Object.entries(schema.columns)) {
      let value = row[colName];
      if (value === undefined && colDef.default !== undefined) value = colDef.default;
      if (colDef.required && (value === undefined || value === null)) {
        throw new DatabaseError(`Column "${colName}" is required in table "${schema.name}"`, 'VALIDATION_ERROR');
      }
      if (value !== undefined && value !== null) {
        this.checkType(colName, colDef.type, value, colDef);
      }
      if (value !== undefined) validated[colName] = value;
    }
    return validated;
  }

  private checkType(colName: string, type: string, value: unknown, colDef?: ColumnDef): void {
    checkFieldType('', colName, type as any, value, colDef);
  }

  // =======================================================================
  //  Schema 持久化
  // =======================================================================

  private async persistSchemas(): Promise<void> {
    const data: Record<string, Record<string, ColumnDef>> = {};
    for (const [name, schema] of this.schemas) {
      data[name] = schema.columns;
    }
    const json = JSON.stringify(data);
    const buf = new TextEncoder().encode(json).buffer;
    await this.backend.write('__aria_schemas', buf);
  }

  private async loadSchemas(): Promise<void> {
    const raw = await this.backend.read('__aria_schemas');
    if (!raw) return;

    try {
      const json = new TextDecoder().decode(raw);
      const data = JSON.parse(json) as Record<string, Record<string, ColumnDef>>;

      for (const [tableName, columns] of Object.entries(data)) {
        const schema: TableSchema = { name: tableName, columns };
        this.schemas.set(tableName, schema);
        this.tablePKs.set(tableName, this.getPK(schema));
      }
    } catch {
      // 忽略损坏的 schema 数据
    }
  }

  // =======================================================================
  //  SSTableStore 构建
  // =======================================================================

  /**
   * 创建命名空间隔离的 SSTableStore。
   *
   * 主 LSM 与每个二级索引 LSM 各持有独立实例：
   * - 文件 key 前缀隔离（sst_ / sst_idx_${table}_${col}_）
   * - 元数据 key 隔离（__aria_lsm_meta / __aria_lsm_meta_${ns}）
   * - id 序列独立（避免 v0.2.4 共享 id 空间导致的文件互相覆盖）
   */
  private createSSTableStore(ns: string): SSTableStore {
    const filePrefix = ns === 'main' ? 'sst_' : `sst_${ns}_`;
    const META_KEY = ns === 'main' ? '__aria_lsm_meta' : `__aria_lsm_meta_${ns}`;
    let seq = 0;
    let seqLoaded = false;

    // v0.4.5: 页面化物理存储（OPFS 后端默认启用）— SSTable 存为 4KB 页面，BufferPool 缓存
    const usePages = this.isPageStorage();
    const pageStore = usePages ? new PageSSTableStore(this.fileManager, this.bufferPool) : null;

    const encodeText = (text: string): ArrayBuffer => {
      return new TextEncoder().encode(text).buffer;
    };

    const readMetaList = async (): Promise<SSTableMeta[]> => {
      const raw = await this.backend.read(META_KEY);
      if (!raw) return [];
      try {
        return JSON.parse(new TextDecoder().decode(raw)) as SSTableMeta[];
      } catch {
        return [];
      }
    };

    return {
      save: async (id, data) => {
        if (pageStore) {
          // 页面化：切页写入 BufferPool 并逐页落盘（save 语义 = 已持久化）
          await pageStore.save(id, data);
          return;
        }
        let buf: ArrayBuffer = data.buffer.slice(data.byteOffset, data.byteOffset + data.byteLength) as ArrayBuffer;
        // 压缩（若启用）— 加密由 EncryptedBackend 在 backend 层透明处理（v0.4.5）
        if (this.config.compression) {
          const compressed = compressLZ4(new Uint8Array(buf));
          buf = compressed.buffer.slice(compressed.byteOffset, compressed.byteOffset + compressed.byteLength) as ArrayBuffer;
        }
        await this.backend.write(`${filePrefix}${id}`, buf);
      },
      load: async (id) => {
        // 页面化读取：meta 有 pageIds → 页面拼接；无（旧数据）→ 整 value
        if (pageStore) {
          const metas = await readMetaList();
          const meta = metas.find((m) => m.id === id);
          if (meta && meta.pageIds && meta.pageIds.length > 0) {
            return pageStore.load(id, meta.pageIds, meta.totalSize);
          }
        }
        const raw = await this.backend.read(`${filePrefix}${id}`);
        if (!raw) return null;
        let buf = new Uint8Array(raw);
        // 解压（若启用）— 解密由 EncryptedBackend 在 backend 层透明处理（v0.4.5）
        if (this.config.compression) {
          // v0.4.5: 压缩流自带原始大小头，无需外部估算
          buf = decompressLZ4(buf) as Uint8Array<ArrayBuffer>;
        }
        return buf;
      },
      delete: async (id) => {
        if (pageStore) {
          const metas = await readMetaList();
          const meta = metas.find((m) => m.id === id);
          if (meta && meta.pageIds && meta.pageIds.length > 0) {
            await pageStore.delete(id, meta.pageIds);
          }
        }
        await this.backend.delete(`${filePrefix}${id}`);
      },
      allocateId: async () => {
        // 从本命名空间的 meta 恢复 id 序列，保证单调递增且不与其他 LSM 冲突
        if (!seqLoaded) {
          const metas = await readMetaList();
          seq = metas.reduce((m, x) => Math.max(m, x.id), 0);
          seqLoaded = true;
        }
        return ++seq;
      },
      listMeta: readMetaList,
      saveMeta: async (meta) => {
        const list = await readMetaList();
        // v0.4.5: 页面化时把页面 ID 列表注入 meta（save 后、saveMeta 前由 LSM 顺序调用）
        const pageIds = pageStore?.getPageIds(meta.id);
        const metaWithPages = pageIds && pageIds.length > 0 ? { ...meta, pageIds } : meta;
        // 更新或添加
        const idx = list.findIndex((m) => m.id === meta.id);
        if (idx >= 0) list[idx] = metaWithPages;
        else list.push(metaWithPages);
        await this.backend.write(META_KEY, encodeText(JSON.stringify(list)));
      },
      deleteMeta: async (id) => {
        const list = await readMetaList();
        const filtered = list.filter((m) => m.id !== id);
        await this.backend.write(META_KEY, encodeText(JSON.stringify(filtered)));
      },
    };
  }

  /** v0.4.5: 是否启用页面化物理存储（默认 OPFS 后端启用，显式配置可覆盖） */
  private isPageStorage(): boolean {
    if (this.config.pageStorage === true) return true;
    if (this.config.pageStorage === false) return false;
    return this.config.storageBackend === 'opfs';
  }

  // =======================================================================
  //  WAL 恢复
  // =======================================================================

  private applyWALRecord(record: WALRecord): void {
    switch (record.type) {
      case WALRecordType.INSERT:
      case WALRecordType.UPDATE:
        if (record.data) {
          this.lsm.put(`${record.tableName}:${record.key}`, record.data);
        }
        break;
      case WALRecordType.DELETE:
        this.lsm.delete(`${record.tableName}:${record.key}`);
        break;
      case WALRecordType.CREATE_TABLE:
        if (record.data?.schema) {
          try {
            const s = JSON.parse(record.data.schema as string) as TableSchema;
            if (!this.schemas.has(s.name)) {
              this.schemas.set(s.name, s);
              this.tablePKs.set(s.name, this.getPK(s));
            }
          } catch { /* skip */ }
        }
        break;
      case WALRecordType.COMMIT:
      case WALRecordType.ROLLBACK:
      case WALRecordType.BEGIN:
        break;
    }
  }

  /**
   * v0.3.3: DROP_TABLE 恢复 — 删除 schema 并清除主 LSM 中该表的所有残留数据。
   *
   * 此前 DROP_TABLE 在恢复时被忽略，而 CREATE_TABLE 回放会重建 schema，
   * 导致崩溃后"已删除的表和数据复活"（实证 P0 bug）。
   */
  private async applyDropTableRecovery(tableName: string): Promise<void> {
    if (!tableName) return;
    // v0.4.2-fix: 清理该表二级索引（崩溃恢复路径同样不留孤儿索引）
    await this.cleanupTableIndexes(tableName);
    this.schemas.delete(tableName);
    this.tablePKs.delete(tableName);

    // 清除主 LSM 中该表前缀的所有数据（含 SSTable 中的旧数据）
    const prefix = `${tableName}:`;
    const endKey = `${prefix}\uffff`;
    await this.lsm.prefetchRange(prefix, endKey);
    const entries = this.lsm.rangeScan(prefix, endKey);
    for (const [key] of entries) {
      this.lsm.delete(key);
    }
  }

  // =======================================================================
  //  二级索引
  // =======================================================================

  /** 更新行的二级索引条目 */
  private updateSecondaryIndexes(
    tableName: string, pkValue: string,
    newRow: Record<string, unknown> | null,
    oldRow: Record<string, unknown> | null,
  ): void {
    const schema = this.schemas.get(tableName);
    if (!schema) return;

    for (const [colName, colDef] of Object.entries(schema.columns)) {
      // v0.3.3: 主键列不建冗余二级索引（主 LSM 即 PK 索引）
      if (!colDef.index && !colDef.unique) continue;
      const idxKey = `${tableName}:idx:${colName}`;
      const idxLsm = this.secondaryIndexes.get(idxKey);
      if (!idxLsm) continue;

      // 删除旧值
      if (oldRow) {
        const oldVal = oldRow[colName];
        if (oldVal !== undefined && oldVal !== null) {
          idxLsm.delete(`${String(oldVal)}:${pkValue}`);
        }
      }

      // 插入新值
      if (newRow) {
        const newVal = newRow[colName];
        if (newVal !== undefined && newVal !== null) {
          idxLsm.put(`${String(newVal)}:${pkValue}`, { pk: pkValue });
        }
      }
    }
  }

  /** 通过二级索引快速查找 */
  private async tryIndexLookup(
    tableName: string,
    query: QueryPlan,
  ): Promise<Record<string, unknown>[] | null> {
    if (!query.where) return null;
    const schema = this.schemas.get(tableName);
    if (!schema) return null;
    const pkCol = this.tablePKs.get(tableName)!;

    for (const [col, condition] of Object.entries(query.where)) {
      // 跳过 $and/$or/$not 逻辑组合
      if (col === '$and' || col === '$or' || col === '$not') continue;

      const colDef = schema.columns[col];
      const hasIndex = colDef && (colDef.index || colDef.unique || colDef.primaryKey);
      if (!hasIndex && col !== pkCol) continue;

      // PK 等值 → 主 LSM 精确查找
      if (col === pkCol) {
        if (typeof condition !== 'object' || condition === null) {
          const key = `${tableName}:${condition}`;
          await this.lsm.prefetchKeys([key]);
          const value = this.lsm.get(key);
          return value ? [{ ...value, [pkCol]: condition }] : [];
        }
        const cond = condition as Record<string, unknown>;
        if ('$eq' in cond) {
          const key = `${tableName}:${cond.$eq}`;
          await this.lsm.prefetchKeys([key]);
          const value = this.lsm.get(key);
          return value ? [{ ...value, [pkCol]: cond.$eq }] : [];
        }
        // v0.3.3: PK $in → 主 LSM 多次精确查找（替代冗余 PK 二级索引）
        if ('$in' in cond && Array.isArray(cond.$in)) {
          const keys = cond.$in.map((v) => `${tableName}:${v}`);
          await this.lsm.prefetchKeys(keys);
          const rows: Record<string, unknown>[] = [];
          const seen = new Set<string>(); // v0.4.1: IN 子查询可能含重复值，按 pk 去重
          for (const v of cond.$in) {
            const pk = String(v);
            if (seen.has(pk)) continue;
            const value = this.lsm.get(`${tableName}:${pk}`);
            if (value) { seen.add(pk); rows.push({ ...value, [pkCol]: pk }); }
          }
          return rows;
        }
        // v0.3.3: PK 范围查询 → 主 LSM 前缀扫描 + 条件过滤（修复字符串算术 bug）
        if ('$gt' in cond || '$gte' in cond || '$lt' in cond || '$lte' in cond) {
          const prefix = `${tableName}:`;
          await this.lsm.prefetchRange(prefix, `${prefix}\uffff`);
          const entries = this.lsm.rangeScan(prefix, `${prefix}\uffff`);
          const rows: Record<string, unknown>[] = [];
          for (const [key, value] of entries) {
            const candidate = { ...value, [pkCol]: key.slice(prefix.length) };
            if (matchWhere(candidate, { [pkCol]: condition })) rows.push(candidate);
          }
          return rows;
        }
      }

      // 二级索引查找
      const idxKey = `${tableName}:idx:${col}`;
      const idxLsm = this.secondaryIndexes.get(idxKey);
      if (!idxLsm) continue;

      // $eq → 精确查找
      if (typeof condition !== 'object' || condition === null) {
        return this.indexScanToRows(tableName, pkCol, idxLsm, String(condition), String(condition));
      }
      const c = condition as Record<string, unknown>;
      if ('$eq' in c) {
        const v = String(c.$eq);
        return this.indexScanToRows(tableName, pkCol, idxLsm, v, v);
      }
      // $in → 多次精确查找
      if ('$in' in c && Array.isArray(c.$in)) {
        const results: Record<string, unknown>[] = [];
        const seenPks = new Set<string>(); // v0.4.1: IN 值可能重复，按 pk 去重
        for (const val of c.$in) {
          const rows = await this.indexScanToRows(tableName, pkCol, idxLsm, String(val), String(val));
          for (const row of rows) {
            const pk = String(row[pkCol]);
            if (!seenPks.has(pk)) {
              seenPks.add(pk);
              results.push(row);
            }
          }
        }
        return results;
      }
      // $gt / $gte / $lt / $lte → 范围扫描
      if ('$gt' in c || '$gte' in c || '$lt' in c || '$lte' in c) {
        let startKey = '';
        let endKey = '\uffff';
        if (c.$gt !== undefined) startKey = `${String(Number(c.$gt) + 1)}:`;
        else if (c.$gte !== undefined) startKey = `${String(c.$gte)}:`;
        if (c.$lt !== undefined) endKey = `${String(Number(c.$lt) - 1)}:\uffff`;
        else if (c.$lte !== undefined) endKey = `${String(c.$lte)}:\uffff`;
        return this.indexScanToRows(tableName, pkCol, idxLsm, startKey, endKey);
      }
    }

    return null;
  }

  /** 从索引扫描结果恢复完整行 */
  private async indexScanToRows(
    tableName: string, pkCol: string, idxLsm: LSM,
    startKey: string, endKey: string,
  ): Promise<Record<string, unknown>[]> {
    // 使用前缀扫描：endKey 需要包含 \uffff 以匹配所有带后缀的 key
    const actualEndKey = endKey.includes('\uffff') ? endKey : `${endKey}\uffff`;
    // 预加载索引 LSM 与主 LSM 涉及的 SSTable
    await idxLsm.prefetchRange(startKey, actualEndKey);
    const entries = idxLsm.rangeScan(startKey, actualEndKey);
    const pks: string[] = [];
    for (const [, idxEntry] of entries) {
      const pk = (idxEntry as any).pk as string;
      if (pk) pks.push(pk);
    }
    await this.lsm.prefetchKeys(pks.map((pk) => `${tableName}:${pk}`));
    const rows: Record<string, unknown>[] = [];
    for (const pk of pks) {
      const row = this.lsm.get(`${tableName}:${pk}`);
      if (row) rows.push({ ...row, [pkCol]: pk });
    }
    return rows;
  }

  // =======================================================================
  //  辅助
  // =======================================================================

  /** 每 10 次 gc 计数器触发一次 MVCC 垃圾回收 */
  private tryGC(): void {
    this.gcCounter++;
    if (this.gcCounter >= 10) {
      this.mvcc.gc(100);
      this.gcCounter = 0;
    }
  }

  /** 回收主 LSM 与所有二级索引 LSM 的临时缓存超限 */
  private trimAllCaches(): void {
    this.lsm.trimCache();
    for (const idxLsm of this.secondaryIndexes.values()) {
      idxLsm.trimCache();
    }
  }

  /** 检查内存预算，超出时强制 flush + GC */
  private checkMemoryBudget(): void {
    const maxBytes = this.config.maxMemoryMB * 1024 * 1024;
    const used = this.lsm.getEstimatedMemory();
    if (used > maxBytes) {
      this.lsm.flush().catch(() => {});
      this.mvcc.gc(50);
    }
  }

  /** 估算 WAL 大小（字节） */
  getWALEstimatedSize(): number {
    return this.wal.getBufferedCount() * 200; // 粗略估算每条 ~200B
  }

  /**
   * ANALYZE: 收集表统计信息
   * 返回行数、平均行大小、索引深度等
   */
  async analyzeTable(tableName: string): Promise<Record<string, unknown>> {
    this.ensureOpen();
    this.ensureTable(tableName);
    const rows = await this.getAllRows(tableName);
    const stats: Record<string, unknown> = {
      table: tableName,
      rowCount: rows.length,
      avgRowSize: rows.length > 0
        ? Math.round(rows.reduce((s, r) => s + JSON.stringify(r).length, 0) / rows.length)
        : 0,
      indexDepth: this.lsm.getStats().levelCounts.filter((c: number) => c > 0).length,
      sstableCount: this.lsm.getStats().sstableCount,
      memtableSize: this.lsm.getStats().memtableSize,
      estimatedMemory: this.lsm.getEstimatedMemory(),
    };

    // 列基数统计
    const schema = this.schemas.get(tableName);
    if (schema && rows.length > 0) {
      const columnStats: Record<string, unknown> = {};
      for (const colName of Object.keys(schema.columns)) {
        const values = new Set(rows.map((r) => String(r[colName])));
        columnStats[colName] = { distinctValues: values.size };
      }
      stats.columnStats = columnStats;
    }

    return stats;
  }

  /**
   * REINDEX: 重建指定表的所有二级索引
   */
  async reindexTable(tableName: string): Promise<number> {
    this.ensureOpen();
    this.ensureTable(tableName);
    return this.reindexTableInternal(tableName);
  }

  /** v0.4.2-fix: 重建索引内部实现（不校验 opened，供 open 恢复流程调用） */
  private async reindexTableInternal(tableName: string): Promise<number> {
    const schema = this.schemas.get(tableName);
    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;
      const idxKey = `${tableName}:idx:${colName}`;
      const idxLsm = this.secondaryIndexes.get(idxKey);
      if (!idxLsm) continue;

      // 清空旧索引
      await idxLsm.clear();
      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)!] });
        }
      }
    }
    return rebuiltCount;
  }

  /**
   * VACUUM: 压缩 LSM + 清理碎片
   */
  async vacuum(): Promise<{ compactedLevels: number; gcVersions: number }> {
    this.ensureOpen();
    // 强制 flush memtable
    await this.lsm.flush();
    // 压缩各层级
    for (let level = 0; level < 6; level++) {
      if (this.lsm.getStats().levelCounts[level] >= 2) {
        await this.lsm.compactLevel(level);
      }
    }
    // GC MVCC 版本（保留最新 10 个）
    const beforeGC = this.mvcc.getGlobalLSN();
    this.mvcc.gc(10);
    return { compactedLevels: 6, gcVersions: beforeGC };
  }

  private ensureOpen(): void {
    if (!this.opened) throw new DatabaseError('AriaEngine not opened', 'DB_NOT_OPEN');
  }

  /** v0.4.2-fix: Aria 事务中 DDL 显式拒绝（结构变更无法通过行快照回滚） */
  private ensureNoDDLInTransaction(op: string): void {
    if (this.currentTxnId) {
      throw new DatabaseError(
        `${op} is not supported inside a transaction (AriaEngine DDL is not transactional)`,
        'NOT_SUPPORTED',
      );
    }
  }

  private ensureTable(tableName: string): void {
    if (!this.schemas.has(tableName)) {
      throw new DatabaseError(`Table "${tableName}" does not exist`, 'TABLE_NOT_FOUND');
    }
  }
}