MetonaSqlark/src/engine/aria/transaction/mvcc.ts

/**
 * AriaEngine MVCC — 多版本并发控制
 * @module engine/aria/transaction/mvcc
 *
 * v0.8.0（review 修正文档）：本模块提供**行版本链**，用途是"事务内的 undo"，
 * 提交即清理。它**不提供快照隔离** —— 引擎的隔离语义是"事务串行"（同一实例
 * 同时只允许一个事务，并发 `beginTransaction` 抛 `TX_ACTIVE`），读取走
 * `txnSnapshot` 未提交快照。旧注释声称 "实现快照隔离 (Snapshot Isolation)"
 * 与实现不符（全量审查发现）。
 */

import type { RowVersion, TxnEntry } from '../types';
import { TransactionState } from '../types';

// ---------------------------------------------------------------------------
//  MVCCManager
// ---------------------------------------------------------------------------

export class MVCCManager {
  /** 所有行版本的存储：tableName.key → 版本链 */
  private versionStore: Map<string, RowVersion[]> = new Map();

  /** 活跃事务表：txnId → TxnEntry */
  private activeTxns: Map<number, TxnEntry> = new Map();

  /** 事务 ID 计数器 */
  private nextTxnId = 1;

  /** 全局提交序列号（用于可见性判断） */
  private globalCommitLsn = 0;

  /**
   * v0.4.2-fix: 每个事务写入的 tableKey 集合 —
   * commit/rollback 只遍历本事务写过的 key，避免全库版本链扫描（大表事务 O(N) → O(写入数)）
   */
  private txnWriteKeys: Map<number, Set<string>> = new Map();

  // =======================================================================
  //  事务管理
  // =======================================================================

  /** 开始一个事务，返回事务 ID */
  beginTransaction(): number {
    const txnId = this.nextTxnId++;
    this.activeTxns.set(txnId, {
      txnId,
      state: TransactionState.ACTIVE,
      snapshotLsn: this.globalCommitLsn,
      startTime: Date.now(),
    });
    this.txnWriteKeys.set(txnId, new Set());
    return txnId;
  }

  /** 提交事务 */
  commitTransaction(txnId: number): void {
    const txn = this.activeTxns.get(txnId);
    if (!txn) throw new Error(`Transaction ${txnId} not found`);

    txn.state = TransactionState.COMMITTED;
    this.globalCommitLsn++;

    // v0.6.3-fix: 已提交版本直接清理 —— 快照读取已移除（v0.5.1），版本链仅作
    // 事务内 undo 记录（rollback/savepoint 用），提交后 LSM 持有权威数据。
    // 此前 commit 仅标记 committed → versionStore 随写入量无限增长（行数据双份常驻）。
    const writeKeys = this.txnWriteKeys.get(txnId);
    if (writeKeys) {
      for (const tableKey of writeKeys) {
        const versions = this.versionStore.get(tableKey);
        if (!versions) continue;
        const filtered = versions.filter((v) => v.txnId !== txnId);
        if (filtered.length === 0) {
          this.versionStore.delete(tableKey);
        } else {
          this.versionStore.set(tableKey, filtered);
        }
      }
    }

    // 清理已提交事务的记录
    this.activeTxns.delete(txnId);
    this.txnWriteKeys.delete(txnId);
  }

  /** 回滚事务 */
  rollbackTransaction(txnId: number): void {
    const txn = this.activeTxns.get(txnId);
    if (!txn) throw new Error(`Transaction ${txnId} not found`);

    txn.state = TransactionState.ABORTED;

    // v0.4.2-fix: 仅移除本事务写入的版本（此前遍历全库 versionStore）
    const writeKeys = this.txnWriteKeys.get(txnId);
    if (writeKeys) {
      for (const tableKey of writeKeys) {
        const versions = this.versionStore.get(tableKey);
        if (!versions) continue;
        const filtered = versions.filter((v) => v.txnId !== txnId);
        if (filtered.length === 0) {
          this.versionStore.delete(tableKey);
        } else {
          this.versionStore.set(tableKey, filtered);
        }
      }
    }

    this.activeTxns.delete(txnId);
    this.txnWriteKeys.delete(txnId);
  }

  // =======================================================================
  //  版本读写
  // =======================================================================

  /**
   * 写入一行（创建新版本）。
   */
  writeVersion(
    tableName: string,
    key: string,
    data: Record<string, unknown>,
    txnId: number,
  ): void {
    const tableKey = `${tableName}.${key}`;
    const versions = this.versionStore.get(tableKey) ?? [];

    const newVersion: RowVersion = {
      txnId,
      data,
      prevVersion: versions.length > 0 ? versions[versions.length - 1] : null,
      committed: false,
    };

    versions.push(newVersion);
    this.versionStore.set(tableKey, versions);
    // v0.4.2-fix: 记录本事务写过的 key（commit/rollback 精准清理）
    this.txnWriteKeys.get(txnId)?.add(tableKey);
  }

  /**
   * 删除一行（创建墓碑版本）。
   */
  deleteVersion(tableName: string, key: string, txnId: number): void {
    this.writeVersion(tableName, key, { __mvcc_tombstone: true } as unknown as Record<string, unknown>, txnId);
  }

  /**
   * v0.3.3: 丢弃指定事务的所有版本记录，但保留事务登记（Savepoint 回滚用）。
   * 快照数据由调用方（引擎 txnSnapshot）负责恢复。
   * v0.4.2-fix: 仅遍历本事务写过的 key（此前全库扫描）。
   */
  discardVersions(txnId: number): void {
    const writeKeys = this.txnWriteKeys.get(txnId);
    if (!writeKeys) return;
    for (const tableKey of writeKeys) {
      const versions = this.versionStore.get(tableKey);
      if (!versions) continue;
      const filtered = versions.filter((v) => v.txnId !== txnId);
      if (filtered.length === 0) {
        this.versionStore.delete(tableKey);
      } else {
        this.versionStore.set(tableKey, filtered);
      }
    }
  }

  /**
   * 清理过旧版本（GC）。
   * 保留每个 key 的最新 N 个已提交版本。
   */
  gc(maxVersionsPerKey: number = 100): void {
    for (const [tableKey, versions] of this.versionStore) {
      if (versions.length <= maxVersionsPerKey) continue;

      // 保留最新的 maxVersionsPerKey 个版本
      const pruned = versions.slice(versions.length - maxVersionsPerKey);
      this.versionStore.set(tableKey, pruned);
    }
  }

  /**
   * 获取全局 LSN。
   */
  getGlobalLSN(): number {
    return this.globalCommitLsn;
  }
}