MetonaSqlark/tests/engine/aria-compression.test.ts

/**
 * v0.8.0 回归套件 —— A38 页面化压缩 + A39 compressLZ4 复杂度
 * ============================================================================
 * A38 **`compression` 在页面化路径上被静默忽略**
 *   `AriaEngine` 的默认配置是 `pageStorage` 自动（OPFS 后端下为 true），
 *   而压缩只写在"整 value 存一个 backend value"的旧分支里，页面化分支
 *   **提前 return** —— 于是 `compression: true` 在默认配置下完全没有效果，
 *   且没有任何提示。本套件的断言方式是**结构性**的：
 *   开启压缩后 `SSTableMeta.totalSize`（实际落盘字节数）必须显著变小 ——
 *   它与实现细节无关，只反映"数据真的被压缩了"。
 *
 *   * 为什么之前没被发现：既有测试只断言"压缩后能读回来"（往返正确），
 *     而"根本没压缩"同样能正确读回 —— 断言太弱，测不出"功能是否生效"。
 *
 * A39 **`compressLZ4` 匹配搜索是 O(n²)**
 *   旧实现逐字节向前扫描最多 65535 个候选位置，每个位置再逐字节比较。
 *   在低压缩率数据（伪随机）上退化为二次复杂度：实测 60KB 输入耗时 **2.3 秒**
 *   （新实现 6ms，约 390×）。SSTable 页/日志段正好是几百 KB 到几 MB，
 *   因此这是普通写入路径上的真实卡顿，不是极端场景。
 *
 * 本套件同时锁定"输出格式未变"：新实现与保留的线性参考实现在同一输入上
 * 必须产出**可互相解压**的流（压缩率可略有差异，因为新实现能找到不同但
 * 同样合法的匹配）。
 */
import { AriaEngine } from '../../src/engine/aria/index';
import { compressLZ4, compressLZ4LinearReference, decompressLZ4 } from '../../src/engine/aria/compression/lz4';
import { resetOPFSMock } from '../helpers/storage-harness';

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

const SCHEMA = () => ({
  name: 't',
  columns: {
    id: { type: 'string' as const, primaryKey: true },
    blob: { type: 'string' as const },
  },
});

/** 高度可压缩的载荷 */
const COMPRESSIBLE = 'x'.repeat(500) + 'y'.repeat(500);

/** 写入 n 行并返回 SSTable 元数据统计 */
async function writeAndMeasure(opts: {
  dbName: string;
  pageStorage: boolean;
  compression: boolean;
  rows?: number;
}): Promise<{ storedBytes: number; sstableCount: number; rowCount: number; roundTripOk: boolean }> {
  const rows = opts.rows ?? 60;
  const engine = new AriaEngine({
    storageBackend: 'opfs',
    pageStorage: opts.pageStorage,
    compression: opts.compression,
    memtableSizeThreshold: 2048,
    checkpointInterval: 100_000_000,
  } as never);
  await engine.open(opts.dbName, 1);
  await engine.createTable(SCHEMA() as never);
  for (let i = 0; i < rows; i++) await engine.insert('t', [{ id: `k${i}`, blob: COMPRESSIBLE }]);

  const lsm = (engine as unknown as {
    lsm: { flush(): Promise<void>; sstableStore: { listMeta(): Promise<Array<{ totalSize: number }>> } };
  }).lsm;
  // v0.8.0（B-6）：**测量前显式 flush**。
  //
  // 此前这里直接读 meta 求和，于是"落盘字节数"取决于测量瞬间有多少数据恰好在
  // SSTable 里 —— 而 manifest 提交（单一提交点）让每次 flush 多一次原子提交，
  // 后台 flush 的进度随之变化，两次实验的"已落盘比例"不再相同，比值就变成在
  // 测时序而不是测压缩（实测：未 flush 时 off=19040/on=5130，flush 后
  // off=58570/on=10096 —— 后者才是同一份数据的真实压缩率）。
  await lsm.flush();
  const metas = await lsm.sstableStore.listMeta();
  const storedBytes = metas.reduce((sum, m) => sum + (m.totalSize ?? 0), 0);

  const read = await engine.find('t', { table: 't' });
  const roundTripOk = read.length === rows && read.every((r) => r.blob === COMPRESSIBLE);
  await engine.close();
  return { storedBytes, sstableCount: metas.length, rowCount: read.length, roundTripOk };
}

describe('[v0.8.0] A38 compression 必须真的生效（含页面化路径）', () => {
  it('页面化路径：开启压缩后落盘字节数显著下降', async () => {
    const off = await writeAndMeasure({ dbName: 'a38-pages-off', pageStorage: true, compression: false });
    const on = await writeAndMeasure({ dbName: 'a38-pages-on', pageStorage: true, compression: true });

    // 先确认两次实验都写出了 SSTable（否则下面的比值没有意义）
    expect(off.sstableCount).toBeGreaterThan(0);
    expect(on.sstableCount).toBeGreaterThan(0);
    // 数据完整性不受压缩影响
    expect(off.roundTripOk).toBe(true);
    expect(on.roundTripOk).toBe(true);

    // 核心断言：修复前 compression 在页面化路径上被忽略 → 两者字节数相同
    expect(on.storedBytes).toBeLessThan(off.storedBytes);
    // 高可压缩内容应有的量级（10 倍是保守下界，实测远高于此）
    expect(off.storedBytes).toBeGreaterThan(on.storedBytes * 5);
  });

  it('整 value 路径：压缩同样生效（两条路径行为一致）', async () => {
    const off = await writeAndMeasure({ dbName: 'a38-whole-off', pageStorage: false, compression: false });
    const on = await writeAndMeasure({ dbName: 'a38-whole-on', pageStorage: false, compression: true });
    expect(off.roundTripOk).toBe(true);
    expect(on.roundTripOk).toBe(true);
    expect(off.storedBytes).toBeGreaterThan(on.storedBytes * 5);
  });

  it('页面化 + 压缩：重开后数据与索引完整', async () => {
    const engine = new AriaEngine({
      storageBackend: 'opfs', pageStorage: true, compression: true,
      memtableSizeThreshold: 2048, checkpointInterval: 100_000_000,
    } as never);
    await engine.open('a38-reopen', 1);
    await engine.createTable(SCHEMA() as never);
    for (let i = 0; i < 40; i++) await engine.insert('t', [{ id: `k${i}`, blob: COMPRESSIBLE }]);
    // 不 close（模拟崩溃后重开，覆盖压缩数据的恢复路径）
    const engine2 = new AriaEngine({
      storageBackend: 'opfs', pageStorage: true, compression: true,
      memtableSizeThreshold: 2048, checkpointInterval: 100_000_000,
    } as never);
    await engine2.open('a38-reopen', 1);
    const rows = await engine2.find('t', { table: 't' });
    expect(rows).toHaveLength(40);
    expect(rows.every((r) => r.blob === COMPRESSIBLE)).toBe(true);
    // 单行精确读取（走 SSTable 数据块解析，而不只是全表扫描）
    const one = await engine2.find('t', { table: 't', where: { id: 'k7' } });
    expect(one).toHaveLength(1);
    expect(one[0].blob).toBe(COMPRESSIBLE);
    await engine.close().catch(() => { /* 已崩溃语义，忽略 */ });
    await engine2.close();
  });

  it('压缩关闭时数据同样完整（回归护栏）', async () => {
    const result = await writeAndMeasure({ dbName: 'a38-nocomp', pageStorage: true, compression: false });
    expect(result.roundTripOk).toBe(true);
    expect(result.rowCount).toBe(60);
  });
});

// ---------------------------------------------------------------------------
//  A39：compressLZ4 复杂度与格式兼容
// ---------------------------------------------------------------------------

/** 确定性伪随机（避免测试本身依赖随机源） */
function makeRandom(length: number, seed = 42): Uint8Array {
  let s = seed >>> 0;
  const out = new Uint8Array(length);
  for (let i = 0; i < length; i++) {
    s = (Math.imul(s, 1664525) + 1013904223) >>> 0;
    out[i] = (s >>> 24) & 0xFF;
  }
  return out;
}

describe('[v0.8.0] A39 compressLZ4：复杂度与格式兼容', () => {
  const cases: Array<[string, Uint8Array]> = [
    ['空输入', new Uint8Array(0)],
    ['单字节', new Uint8Array([42])],
    ['全同字节', new Uint8Array(5000).fill(7)],
    ['周期序列', new Uint8Array(Array.from({ length: 40_000 }, (_, i) => i % 7))],
    ['伪随机 8KB', makeRandom(8_000)],
    ['伪随机 60KB', makeRandom(60_000)],
  ];

  it.each(cases)('%s：往返一致且与线性参考实现可互解', (_label, data) => {
    const fast = compressLZ4(data);
    const slow = compressLZ4LinearReference(data);

    // 新实现自身往返
    const fromFast = decompressLZ4(fast);
    expect(fromFast.length).toBe(data.length);
    expect(Array.from(fromFast)).toEqual(Array.from(data));

    // 旧实现的输出，新解压器必须能读（格式兼容：既有落盘数据不需要迁移）
    const fromSlow = decompressLZ4(slow);
    expect(Array.from(fromSlow)).toEqual(Array.from(data));

    // 新实现的输出，旧解压器（同一份代码，此处仅作对称性验证）也必须能读
    expect(Array.from(decompressLZ4(fast))).toEqual(Array.from(data));
  });

  it('低压缩率数据不再退化（60KB 伪随机在 1 秒内完成）', () => {
    const data = makeRandom(60_000);
    const start = Date.now();
    const compressed = compressLZ4(data);
    const elapsed = Date.now() - start;

    // 修复前该输入耗时约 2.3 秒（逐位置线性扫描 → 二次复杂度）。
    // 阈值取 1 秒：即使 CI 慢 10 倍也仍能通过，而回退到旧实现必然失败。
    expect(elapsed).toBeLessThan(1000);
    expect(decompressLZ4(compressed).length).toBe(data.length);
  });

  it('大输入（1MB 可压缩内容）在合理时间内完成', () => {
    const data = new Uint8Array(1_000_000);
    for (let i = 0; i < data.length; i++) data[i] = i % 251;
    const start = Date.now();
    const compressed = compressLZ4(data);
    const elapsed = Date.now() - start;
    expect(elapsed).toBeLessThan(3000);
    // 周期性内容应被显著压缩
    expect(compressed.length).toBeLessThan(data.length);
    expect(decompressLZ4(compressed).length).toBe(data.length);
  });
});