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

/**
 * AriaEngine MVCC — 多版本并发控制
 * @module engine/aria/transaction/mvcc
 *
 * 实现快照隔离 (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;

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

  /** 开始一个事务，返回事务 ID */
  beginTransaction(): number {
    const txnId = this.nextTxnId++;
    this.activeTxns.set(txnId, {
      txnId,
      state: TransactionState.ACTIVE,
      snapshotLsn: this.globalCommitLsn,
      startTime: Date.now(),
    });
    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++;

    // 标记此事务写入的所有版本为已提交
    for (const [, versions] of this.versionStore) {
      for (const version of versions) {
        if (version.txnId === txnId) {
          version.committed = true;
        }
      }
    }

    // 清理已提交事务的记录
    this.activeTxns.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;

    // 移除此事务写入的所有版本
    for (const [tableKey, versions] of this.versionStore) {
      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);
  }

  /** 检查事务是否活跃 */
  isActive(txnId: number): boolean {
    const txn = this.activeTxns.get(txnId);
    return txn !== undefined && txn.state === TransactionState.ACTIVE;
  }

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

  /**
   * 写入一行（创建新版本）。
   */
  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);
  }

  /**
   * 读取一行（对指定事务可见的最新版本）。
   */
  readVersion(
    tableName: string,
    key: string,
    txnId: number,
  ): Record<string, unknown> | null {
    const txn = this.activeTxns.get(txnId);
    if (!txn) return null;

    const tableKey = `${tableName}.${key}`;
    const versions = this.versionStore.get(tableKey);
    if (!versions || versions.length === 0) return null;

    // 从最新版本向前遍历
    for (let i = versions.length - 1; i >= 0; i--) {
      const version = versions[i];

      // 1. 如果是当前事务写入的（未提交），可见
      if (version.txnId === txnId) {
        return version.data;
      }

      // 2. 如果是已提交的版本，且在快照 LSN 之前提交，可见
      if (version.committed) {
        // 简化：所有已提交版本都可见
        return version.data;
      }

      // 3. 其他事务的未提交版本，不可见，继续找更早的版本
    }

    return null;
  }

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

  /**
   * 获取所有行的最新已提交版本（用于非事务读取）。
   */
  getLatestCommittedVersions(
    tableName: string,
  ): Record<string, Record<string, unknown>> {
    const result: Record<string, Record<string, unknown>> = {};

    for (const [tableKey, versions] of this.versionStore) {
      if (!tableKey.startsWith(`${tableName}.`)) continue;
      const key = tableKey.slice(tableName.length + 1);

      for (let i = versions.length - 1; i >= 0; i--) {
        const version = versions[i];
        if (version.committed && !(version.data as unknown as Record<string, unknown>).__mvcc_tombstone) {
          result[key] = version.data;
          break;
        }
      }
    }

    return result;
  }

  /**
   * v0.3.3: 丢弃指定事务的所有版本记录，但保留事务登记（Savepoint 回滚用）。
   * 快照数据由调用方（引擎 txnSnapshot）负责恢复。
   */
  discardVersions(txnId: number): void {
    for (const [tableKey, versions] of this.versionStore) {
      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);
    }
  }

  /**
   * 获取所有未提交事务中的 key 列表。
   */
  getActiveWriteKeys(tableName: string, txnId: number): Set<string> {
    const keys = new Set<string>();
    const prefix = `${tableName}.`;

    for (const [tableKey, versions] of this.versionStore) {
      if (!tableKey.startsWith(prefix)) continue;
      const latestVersion = versions[versions.length - 1];
      if (latestVersion.txnId === txnId && !latestVersion.committed) {
        keys.add(tableKey.slice(prefix.length));
      }
    }

    return keys;
  }

  /**
   * 清理指定表的所有版本。
   */
  clearTable(tableName: string): void {
    const prefix = `${tableName}.`;
    for (const [tableKey] of this.versionStore) {
      if (tableKey.startsWith(prefix)) {
        this.versionStore.delete(tableKey);
      }
    }
  }

  /**
   * 获取活跃事务数。
   */
  getActiveTxnCount(): number {
    return this.activeTxns.size;
  }

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