MetonaSqlark/tests/sql-ext2.test.ts

/**
 * v0.3.1 功能测试
 * @module tests/sql-ext2
 *
 * 覆盖：CASE WHEN 表达式 / JOIN + 关联子查询 / WAL 批量组提交
 */
import { MetonaSqlark } from '../src/core';
import { parse } from '../src/sql/parser';
import { WAL, type WALStore } from '../src/engine/aria/wal/log';
import { WALRecordType } from '../src/engine/aria/types';

import { resetOPFSMock } from './helpers/storage-harness';

beforeEach(() => { resetOPFSMock(); });

async function createDb(mode: 'memory' | 'aria' = 'memory') {
  const db = new MetonaSqlark({ name: `sql-ext2-${mode}-${Date.now()}-${Math.random()}`, mode, diskEngine: 'opfs' });
  await db.init();
  await db.defineTable('users', {
    id: { type: 'string', primaryKey: true },
    name: { type: 'string' },
    age: { type: 'number' },
    city: { type: 'string' },
  });
  await db.defineTable('orders', {
    id: { type: 'string', primaryKey: true },
    user_id: { type: 'string' },
    amount: { type: 'number' },
  });
  await db.query(`INSERT INTO users VALUES ('u1', 'Alice', 30, 'Beijing')`);
  await db.query(`INSERT INTO users VALUES ('u2', 'Bob', 17, 'Shanghai')`);
  await db.query(`INSERT INTO users VALUES ('u3', 'Carol', 42, 'Beijing')`);
  await db.query(`INSERT INTO orders VALUES ('o1', 'u1', 100)`);
  await db.query(`INSERT INTO orders VALUES ('o2', 'u1', 200)`);
  await db.query(`INSERT INTO orders VALUES ('o3', 'u2', 50)`);
  return db;
}

// ===================================================================
//  CASE WHEN
// ===================================================================

describe('[v0.3.1] CASE WHEN', () => {
  test('基本 CASE WHEN（单条件 + ELSE）', async () => {
    const db = await createDb();
    const rows = await db.query(
      `SELECT name, CASE WHEN age >= 18 THEN 'adult' ELSE 'minor' END AS status FROM users`,
    ) as Record<string, unknown>[];
    expect(rows).toHaveLength(3);
    const byName = Object.fromEntries(rows.map((r) => [r.name, r]));
    expect(byName['Alice'].status).toBe('adult');
    expect(byName['Bob'].status).toBe('minor');
    expect(byName['Carol'].status).toBe('adult');
    await db.close();
  });

  test('多 WHEN 分支按顺序匹配', async () => {
    const db = await createDb();
    const rows = await db.query(
      `SELECT name, CASE WHEN age < 18 THEN 'teen' WHEN age < 40 THEN 'adult' ELSE 'senior' END AS age_group FROM users`,
    ) as Record<string, unknown>[];
    const byName = Object.fromEntries(rows.map((r) => [r.name, r]));
    expect(byName['Alice'].age_group).toBe('adult');
    expect(byName['Bob'].age_group).toBe('teen');
    expect(byName['Carol'].age_group).toBe('senior');
    await db.close();
  });

  test('THEN 值为列引用', async () => {
    const db = await createDb();
    const rows = await db.query(
      `SELECT CASE WHEN age >= 18 THEN city ELSE 'underage' END AS location FROM users`,
    ) as Record<string, unknown>[];
    const cities = rows.map((r) => r.location);
    expect(cities).toContain('Beijing');
    expect(cities).toContain('underage');
    await db.close();
  });

  test('无 ELSE 时返回 null', async () => {
    const db = await createDb();
    const rows = await db.query(
      `SELECT name, CASE WHEN age >= 40 THEN 'senior' END AS tag FROM users`,
    ) as Record<string, unknown>[];
    const byName = Object.fromEntries(rows.map((r) => [r.name, r]));
    expect(byName['Carol'].tag).toBe('senior');
    expect(byName['Alice'].tag).toBeNull();
    await db.close();
  });

  test('字面量：数字 / 布尔 / 字符串', async () => {
    const db = await createDb();
    const rows = await db.query(
      `SELECT name, CASE WHEN age > 20 THEN 1 ELSE 0 END AS flag, CASE WHEN city = 'Beijing' THEN true ELSE false END AS is_bj FROM users`,
    ) as Record<string, unknown>[];
    const byName = Object.fromEntries(rows.map((r) => [r.name, r]));
    expect(byName['Alice'].flag).toBe(1);
    expect(byName['Bob'].flag).toBe(0);
    expect(byName['Alice'].is_bj).toBe(true);
    expect(byName['Bob'].is_bj).toBe(false);
    await db.close();
  });

  test('多条件组合（AND/OR）', async () => {
    const db = await createDb();
    const rows = await db.query(
      `SELECT name, CASE WHEN age >= 18 AND city = 'Beijing' THEN 'local adult' ELSE 'other' END AS label FROM users`,
    ) as Record<string, unknown>[];
    const byName = Object.fromEntries(rows.map((r) => [r.name, r]));
    expect(byName['Alice'].label).toBe('local adult');
    expect(byName['Carol'].label).toBe('local adult');
    expect(byName['Bob'].label).toBe('other');
    await db.close();
  });

  test('语法解析：CASE 列被解析为原文', () => {
    const stmt = parse(`SELECT name, CASE WHEN age > 18 THEN 'x' ELSE 'y' END AS s FROM users`) as any;
    expect(stmt.columns[1]).toMatch(/^CASE WHEN age > 18 THEN 'x' ELSE 'y' END AS s$/);
  });

  test('与普通列混合投影', async () => {
    const db = await createDb();
    const rows = await db.query(
      `SELECT name, age, CASE WHEN age >= 18 THEN 'ok' ELSE 'no' END AS adult FROM users`,
    ) as Record<string, unknown>[];
    expect(rows[0].name).toBeDefined();
    expect(rows[0].age).toBeDefined();
    expect(rows[0].adult).toBeDefined();
    expect(Object.keys(rows[0])).toEqual(expect.arrayContaining(['name', 'age', 'adult']));
    await db.close();
  });

  test('Aria 引擎可用', async () => {
    const db = await createDb('aria');
    const rows = await db.query(
      `SELECT name, CASE WHEN age >= 18 THEN 'adult' ELSE 'minor' END AS status FROM users`,
    ) as Record<string, unknown>[];
    expect(rows).toHaveLength(3);
    await db.close();
  });
});

// ===================================================================
//  JOIN + 关联子查询
// ===================================================================

describe('[v0.3.1] JOIN + 关联子查询', () => {
  test('JOIN 结果上执行 EXISTS 关联过滤', async () => {
    const db = await createDb();
    // 有高额订单（>150）的用户
    const rows = await db.query(
      `SELECT u.name FROM users u JOIN orders o ON u.id = o.user_id WHERE EXISTS (SELECT 1 FROM orders o2 WHERE o2.user_id = u.id AND o2.amount > 150)`,
    ) as Record<string, unknown>[];
    // 只有 u1 有 200 元订单
    expect(rows).toHaveLength(2); // JOIN 展开 2 行（u1 有 2 个订单）
    expect(rows.every((r) => r['u.name'] === 'Alice')).toBe(true);
    await db.close();
  });

  test('JOIN + NOT EXISTS 排除关联行', async () => {
    const db = await createDb();
    const rows = await db.query(
      `SELECT DISTINCT u.name FROM users u LEFT JOIN orders o ON u.id = o.user_id WHERE NOT EXISTS (SELECT 1 FROM orders o2 WHERE o2.user_id = u.id)`,
    ) as Record<string, unknown>[];
    // 无订单用户：u3 (Carol)（JOIN 列键带表名前缀）
    expect(rows.map((r) => r['u.name'])).toEqual(['Carol']);
    await db.close();
  });

  test('JOIN + 关联子查询 + 普通条件组合', async () => {
    const db = await createDb();
    const rows = await db.query(
      `SELECT u.name FROM users u JOIN orders o ON u.id = o.user_id WHERE EXISTS (SELECT 1 FROM orders o2 WHERE o2.user_id = u.id) AND o.amount > 50`,
    ) as Record<string, unknown>[];
    // 有订单且订单 >50：u1 的 o1(100), o2(200) + u2 的 o3(50 不满足)
    expect(rows).toHaveLength(2);
    expect(rows.every((r) => r['u.name'] === 'Alice')).toBe(true);
    await db.close();
  });

  test('JOIN 子查询中的 $col 引用绑定外层行', async () => {
    const db = await createDb();
    const rows = await db.query(
      `SELECT o.id, o.amount FROM orders o JOIN users u ON u.id = o.user_id WHERE EXISTS (SELECT 1 FROM orders o2 WHERE o2.amount > o.amount)`,
    ) as Record<string, unknown>[];
    // 存在比自身金额更大的订单：o1(100) < o2(200) → o1 满足；o2 无更大；o3(50) < o1/o2 → 满足
    const ids = rows.map((r) => r['o.id']).sort();
    expect(ids).toEqual(['o1', 'o3']);
    await db.close();
  });

  test('Aria 引擎 JOIN + EXISTS 可用', async () => {
    const db = await createDb('aria');
    const rows = await db.query(
      `SELECT u.name FROM users u JOIN orders o ON u.id = o.user_id WHERE EXISTS (SELECT 1 FROM orders o2 WHERE o2.user_id = u.id AND o2.amount > 150)`,
    ) as Record<string, unknown>[];
    expect(rows).toHaveLength(2);
    await db.close();
  });
});

// ===================================================================
//  WAL 批量组提交
// ===================================================================

describe('[v0.3.1] WAL 批量组提交', () => {
  class MockWALStore implements WALStore {
    chunks: Uint8Array[] = [];
    appendCount = 0;
    async append(data: Uint8Array) { this.chunks.push(data); this.appendCount++; }
    async readAll(): Promise<Uint8Array> {
      const total = this.chunks.reduce((s, c) => s + c.byteLength, 0);
      const combined = new Uint8Array(total);
      let off = 0;
      for (const c of this.chunks) { combined.set(c, off); off += c.byteLength; }
      return combined;
    }
    async truncate() { this.chunks = []; this.appendCount = 0; }
    async exists() { return this.chunks.length > 0; }
  }

  test('appendBatch 合并为一次底层写入', async () => {
    const store = new MockWALStore();
    const wal = new WAL(store, true, 'full');

    await wal.appendBatch([
      { type: WALRecordType.INSERT, txnId: 0, tableName: 't', key: '1', data: { v: 1 } },
      { type: WALRecordType.INSERT, txnId: 0, tableName: 't', key: '2', data: { v: 2 } },
      { type: WALRecordType.INSERT, txnId: 0, tableName: 't', key: '3', data: { v: 3 } },
    ]);

    expect(store.appendCount).toBe(1); // 3 条记录 1 次写入
  });

  test('appendBatch 记录可恢复', async () => {
    const store = new MockWALStore();
    const wal = new WAL(store, true, 'full');

    await wal.appendBatch([
      { type: WALRecordType.INSERT, txnId: 0, tableName: 'users', key: 'a', data: { n: 1 } },
      { type: WALRecordType.INSERT, txnId: 0, tableName: 'users', key: 'b', data: { n: 2 } },
    ]);

    const records: { tableName: string; key: string }[] = [];
    await wal.recover((r) => records.push(r));
    expect(records).toHaveLength(2);
    expect(records[0].key).toBe('a');
    expect(records[1].key).toBe('b');
  });

  test('batch 模式 appendBatch 缓冲后 flush', async () => {
    const store = new MockWALStore();
    const wal = new WAL(store, true, 'batch');

    await wal.appendBatch([
      { type: WALRecordType.INSERT, txnId: 0, tableName: 't', key: '1', data: {} },
      { type: WALRecordType.INSERT, txnId: 0, tableName: 't', key: '2', data: {} },
    ]);
    expect(store.appendCount).toBe(0); // 缓冲未落盘

    await wal.flush();
    expect(store.appendCount).toBe(1);
  });

  test('append 与 appendBatch 共存', async () => {
    const store = new MockWALStore();
    const wal = new WAL(store, true, 'full');

    await wal.append({ type: WALRecordType.BEGIN, txnId: 7, tableName: '', key: '' });
    await wal.appendBatch([
      { type: WALRecordType.INSERT, txnId: 7, tableName: 't', key: '1', data: {} },
      { type: WALRecordType.INSERT, txnId: 7, tableName: 't', key: '2', data: {} },
    ]);
    await wal.append({ type: WALRecordType.COMMIT, txnId: 7, tableName: '', key: '' });

    expect(store.appendCount).toBe(3); // BEGIN + 批量(1) + COMMIT
    const records: number[] = [];
    await wal.recover((r) => records.push(r.type));
    expect(records).toEqual([
      WALRecordType.BEGIN,
      WALRecordType.INSERT,
      WALRecordType.INSERT,
      WALRecordType.COMMIT,
    ]);
  });

  test('引擎 insert 批量写入只触发一次 WAL 落盘', async () => {
    const db = new MetonaSqlark({ name: `wal-batch-${Date.now()}`, mode: 'aria', diskEngine: 'memory' });
    await db.init();
    await db.defineTable('users', { id: { type: 'string', primaryKey: true }, name: { type: 'string' } });

    const engine = db.getEngine() as any;
    // v0.4.5: WAL 改用分片存储（SegmentedWALStore，无 append 接口的后端回退单 key write）；
    // 组提交语义保持：批量记录仍合并为 1 次底层落盘
    const origWrite = engine.backend.write.bind(engine.backend);
    let walWrites = 0;
    engine.backend.write = async (key: string, data: ArrayBuffer) => {
      if (key.startsWith('__wal_')) walWrites++;
      return origWrite(key, data);
    };

    await db.table('users').insertMany([
      { id: '1', name: 'A' },
      { id: '2', name: 'B' },
      { id: '3', name: 'C' },
      { id: '4', name: 'D' },
    ]);

    expect(walWrites).toBe(1); // 4 行合并为 1 次 WAL 落盘
    const rows = await db.query('SELECT * FROM users') as Record<string, unknown>[];
    expect(rows).toHaveLength(4);
    await db.close();
  });
});