diff --git a/.changeset/eval-parquet-snapshot-seed.md b/.changeset/eval-parquet-snapshot-seed.md new file mode 100644 index 0000000000..68c73e6e9e --- /dev/null +++ b/.changeset/eval-parquet-snapshot-seed.md @@ -0,0 +1,5 @@ +--- +"@hyperdx/hdx-eval": minor +--- + +Add a Parquet-snapshot fast-seed path for the MCP eval suite (Milestone 2 — kill the seed bottleneck). Full-volume seed generation is a CPU-bound JS loop that takes hours; the snapshot path generates the data once, exports each eval table to Parquet, and on later runs loads the Parquet straight into ClickHouse (I/O-bound, ~an order of magnitude faster). New CLI commands: `export-snapshot`, `load-snapshot`, and `seed-logic-hash`. The load path drops the rollup materialized views, bulk-inserts the raw Parquet (no per-block MV fan-out), then backfills the rollup tables in one shot using the same SELECTs the MVs run, and recreates the MVs — producing byte-identical rollup content much faster than letting the MVs fan out on the bulk insert. In CI the snapshot is stored in an `actions/cache` entry keyed on a hash of the seed-generation source, so it is reused across runs and regenerated only when the seeding logic changes. diff --git a/.github/workflows/evals.yml b/.github/workflows/evals.yml index 189dcd1c69..769e6da849 100644 --- a/.github/workflows/evals.yml +++ b/.github/workflows/evals.yml @@ -18,9 +18,11 @@ on: type: string default: latency-spike volume_factor: - description: 'Seed volume factor (fraction of full volume)' + description: + 'Snapshot generation volume factor (1 = full; lower = cheaper test + seed)' type: string - default: '0.01' + default: '1' runs: description: 'Runs per (scenario, MCP) cell' type: string @@ -40,13 +42,22 @@ jobs: contents: read pull-requests: write env: - # M1 defaults; workflow_dispatch inputs override on manual runs. + # M2: single scenario for the fast-seed test, FULL volume via snapshot. HDX_EVAL_SCENARIO: ${{ github.event.inputs.scenario || 'latency-spike' }} - HDX_EVAL_VOLUME_FACTOR: ${{ github.event.inputs.volume_factor || '0.01' }} HDX_EVAL_RUNS: ${{ github.event.inputs.runs || '1' }} HDX_EVAL_OUTPUT_DIR: ${{ github.workspace }}/eval-output HDX_EVALS_OUTPUT_DIR: ${{ github.workspace }}/eval-output HDX_EVALS_RUNS_DIR: ${{ github.workspace }}/eval-output/runs + # M2 Parquet snapshot fast-seed path. + HDX_EVALS_SNAPSHOT_DIR: ${{ github.workspace }}/eval-snapshot + # Path INSIDE the runner container (matches the compose mount target). + HDX_EVAL_SNAPSHOT_DIR: /work/eval-snapshot + # Generate at FULL volume on a snapshot miss (the point of M2). Override + # via workflow_dispatch input for a cheaper test. + HDX_EVAL_SNAPSHOT_VOLUME_FACTOR: + ${{ github.event.inputs.volume_factor || '1' }} + # Stable anchor so cached snapshot timestamps stay valid across runs. + HDX_EVAL_ANCHOR: '2026-06-01T00:00:00Z' steps: - name: Checkout uses: actions/checkout@v6 @@ -59,13 +70,45 @@ jobs: - name: Expose GitHub Actions cache to BuildKit uses: crazy-max/ghaction-github-runtime@v4 - - name: Prepare output dir + - name: Prepare output + snapshot dirs # World-writable so the runner container's non-root `node` user (uid - # 1000) can write runs/ + verdict.md into these bind mounts regardless - # of the host runner's uid. + # 1000) can write runs/ + verdict.md + the exported snapshot into these + # bind mounts regardless of the host runner's uid. run: | mkdir -p "${HDX_EVAL_OUTPUT_DIR}/runs" - chmod -R 777 "${HDX_EVAL_OUTPUT_DIR}" + mkdir -p "${HDX_EVALS_SNAPSHOT_DIR}" + chmod -R 777 "${HDX_EVAL_OUTPUT_DIR}" "${HDX_EVALS_SNAPSHOT_DIR}" + + # Compute the seed-logic hash — the cache key. It changes ONLY when the + # seed-generation source changes, so a cached snapshot is reused across + # runs and regenerated exactly when (and only when) seeding logic changes. + # hashFiles() here must cover the same files the CLI's seedLogicHash() + # hashes (see src/clickhouse/parquetSnapshot.ts SEED_LOGIC_GLOBS). + - name: Compute seed-logic hash + id: seedhash + run: | + HASH="${{ hashFiles( + 'packages/hdx-eval/src/generators/**', + 'packages/hdx-eval/src/scenarios/**', + 'packages/hdx-eval/src/rng/**', + 'packages/hdx-eval/src/clickhouse/insert.ts', + 'packages/hdx-eval/src/clickhouse/schema.ts', + 'packages/hdx-eval/src/clickhouse/parquetSnapshot.ts' + ) }}" + echo "hash=$HASH" >> "$GITHUB_OUTPUT" + echo "Seed-logic hash: $HASH" + + # Restore (and, at job end, save) the full-volume Parquet snapshot keyed + # on the seed-logic hash + scenario. No restore-keys: a stale snapshot is + # never partially reused across a logic change — a miss regenerates. + - name: Restore/save eval seed snapshot + uses: actions/cache@v4 + with: + path: ${{ env.HDX_EVALS_SNAPSHOT_DIR }} + key: + eval-seed-${{ env.HDX_EVAL_SCENARIO }}-v${{ + env.HDX_EVAL_SNAPSHOT_VOLUME_FACTOR }}-${{ + steps.seedhash.outputs.hash }} # Build both DISTINCT images with GHA layer caching. The HyperDX # all-in-one build is heavy (OTel collector + Next.js); caching keeps diff --git a/docker-compose.evals.yml b/docker-compose.evals.yml index 1db2e42c12..cb2336ea1c 100644 --- a/docker-compose.evals.yml +++ b/docker-compose.evals.yml @@ -107,11 +107,22 @@ services: HDX_EVAL_OUT_COMMENT: /work/eval-output/verdict.md HDX_EVAL_RUN_URL: ${HDX_EVAL_RUN_URL:-} HDX_EVAL_COMMIT_SHA: ${HDX_EVAL_COMMIT_SHA:-} + HDX_EVAL_ANCHOR: ${HDX_EVAL_ANCHOR:-2026-06-01T00:00:00Z} + # M2 Parquet snapshot fast-seed path. When HDX_EVAL_SNAPSHOT_DIR is set, + # the runner loads a cached snapshot (fast) or generates+exports one + # (miss). The workflow backs this dir with actions/cache keyed on the + # seed-logic hash. Empty by default → M1 low-volume live seed. + HDX_EVAL_SNAPSHOT_DIR: ${HDX_EVAL_SNAPSHOT_DIR:-} + HDX_EVAL_SNAPSHOT_VOLUME_FACTOR: ${HDX_EVAL_SNAPSHOT_VOLUME_FACTOR:-1} volumes: # Persist runs + rendered verdict to the host so CI can upload artifacts # and post the PR comment. - ${HDX_EVALS_OUTPUT_DIR:-./eval-output}:/work/eval-output - ${HDX_EVALS_RUNS_DIR:-./eval-output/runs}:/work/packages/hdx-eval/runs + # Parquet snapshot dir (host-side, cached by the workflow). Mounted even + # when unused so the container path is stable; the runner only touches it + # when HDX_EVAL_SNAPSHOT_DIR points inside it. + - ${HDX_EVALS_SNAPSHOT_DIR:-./eval-snapshot}:/work/eval-snapshot networks: - evals diff --git a/docker/hdx-eval-runner/run-evals.sh b/docker/hdx-eval-runner/run-evals.sh index 6846f80571..767ec89558 100755 --- a/docker/hdx-eval-runner/run-evals.sh +++ b/docker/hdx-eval-runner/run-evals.sh @@ -23,7 +23,7 @@ # ClickHouse HTTP URL as the HyperDX API sees it # (e.g. http://localhost:8123 for the all-in-one image) # -# Optional env (with M1 defaults): +# Optional env (with defaults): # HDX_EVAL_SCENARIO default: latency-spike # HDX_EVAL_VOLUME_FACTOR default: 0.01 (1% of full volume — scale up later) # HDX_EVAL_RUNS default: 1 @@ -33,6 +33,20 @@ # HDX_EVAL_OUT_COMMENT default: /work/eval-output/verdict.md # HDX_EVAL_RUN_URL workflow run URL (for the comment footer) # HDX_EVAL_COMMIT_SHA commit SHA (for the comment footer) +# HDX_EVAL_ANCHOR fixed anchor ISO for deterministic seed timestamps +# (default: 2026-06-01T00:00:00Z — must be stable so a +# cached snapshot's timestamps stay valid across runs) +# +# ── M2 Parquet snapshot (fast seed) ──────────────────────────────────────── +# HDX_EVAL_SNAPSHOT_DIR when set, enables the Parquet snapshot fast path. +# The workflow restores/saves this dir as an +# actions/cache entry keyed on the seed-logic hash. +# - dir has a manifest.json → LOAD it (fast path) +# - dir empty/no manifest → GENERATE at full volume, +# then EXPORT so the cache-save persists it +# HDX_EVAL_SNAPSHOT_VOLUME_FACTOR +# volume factor to GENERATE at on a snapshot miss. +# default: 1 (FULL volume — the whole point of M2) # # The verdict is completion-only and ADVISORY: this script exits 0 as long as # it produced a summary + verdict, even if the verdict is FAIL. Genuine @@ -48,6 +62,10 @@ MAX_TURNS="${HDX_EVAL_MAX_TURNS:-15}" TIMEOUT_MS="${HDX_EVAL_TIMEOUT_MS:-600000}" MCP="${HDX_EVAL_MCP:-hyperdx}" OUT_COMMENT="${HDX_EVAL_OUT_COMMENT:-/work/eval-output/verdict.md}" +# Fixed anchor so a cached snapshot's seeded timestamps stay valid across runs. +ANCHOR="${HDX_EVAL_ANCHOR:-2026-06-01T00:00:00Z}" +SNAPSHOT_DIR="${HDX_EVAL_SNAPSHOT_DIR:-}" +SNAPSHOT_VOLUME_FACTOR="${HDX_EVAL_SNAPSHOT_VOLUME_FACTOR:-1}" : "${ANTHROPIC_API_KEY:?ANTHROPIC_API_KEY is required}" : "${HDX_EVAL_API_URL:?HDX_EVAL_API_URL is required (e.g. http://hyperdx:8000)}" @@ -78,19 +96,46 @@ echo "::group::[1/5] Setup" --connection-ch-url "$CONNECTION_CH_URL" echo "::endgroup::" -echo "::group::[2/5] Seed (low volume)" -"${CLI[@]}" seed "$SCENARIO" --volume-factor "$VOLUME_FACTOR" +echo "::group::[2/5] Seed" +if [ -n "$SNAPSHOT_DIR" ]; then + # ── M2 Parquet snapshot fast path ── + # The workflow restores $SNAPSHOT_DIR from actions/cache (keyed on the + # seed-logic hash) BEFORE this container runs. If the cache hit, the dir has + # a manifest.json and we LOAD it (fast). If it missed, the dir is empty and + # we GENERATE at full volume then EXPORT, so the workflow's cache-save step + # persists the snapshot for the next run. + SCEN_SNAP_DIR="$SNAPSHOT_DIR/$SCENARIO" + if [ -f "$SCEN_SNAP_DIR/manifest.json" ]; then + echo "Snapshot cache HIT → loading Parquet from $SCEN_SNAP_DIR" + "${CLI[@]}" load-snapshot "$SCENARIO" --dir "$SCEN_SNAP_DIR" + else + echo "Snapshot cache MISS → generating full-volume seed (factor $SNAPSHOT_VOLUME_FACTOR) then exporting Parquet" + "${CLI[@]}" seed "$SCENARIO" \ + --volume-factor "$SNAPSHOT_VOLUME_FACTOR" \ + --now "$ANCHOR" + "${CLI[@]}" export-snapshot "$SCENARIO" \ + --dir "$SCEN_SNAP_DIR" \ + --volume-factor "$SNAPSHOT_VOLUME_FACTOR" \ + --anchor "$ANCHOR" + fi +else + # M1 path: plain low-volume live seed (no snapshot caching). + "${CLI[@]}" seed "$SCENARIO" --volume-factor "$VOLUME_FACTOR" --now "$ANCHOR" +fi echo "::endgroup::" echo "::group::[3/5+4/5] Run + Grade" # `run` auto-grades and auto-reports unless told otherwise. We let it do the -# grade + report inline so a single command drives stages 3–5. --reseed is not -# passed: the seed above already loaded data, and run reuses it. +# grade + report inline so a single command drives stages 3–5. Pin the anchor +# to the SAME value the seed used so the agent's "now" matches the seeded +# timestamps. Data already exists (seeded or snapshot-loaded above), so `run` +# skips its auto-seed — no --reseed. "${CLI[@]}" run "$SCENARIO" \ --mcp "$MCP" \ --runs "$RUNS" \ --max-turns "$MAX_TURNS" \ - --timeout "$TIMEOUT_MS" + --timeout "$TIMEOUT_MS" \ + --anchor-time "$ANCHOR" echo "::endgroup::" # Find the batch produced by the run (newest directory under runs/). diff --git a/packages/hdx-eval/src/__tests__/parquetSnapshot.test.ts b/packages/hdx-eval/src/__tests__/parquetSnapshot.test.ts new file mode 100644 index 0000000000..ce4fc348c4 --- /dev/null +++ b/packages/hdx-eval/src/__tests__/parquetSnapshot.test.ts @@ -0,0 +1,86 @@ +import { mkdtempSync, rmSync } from 'fs'; +import { tmpdir } from 'os'; +import { join } from 'path'; + +import { + manifestPath, + readManifest, + seedLogicHash, + seedLogicHashShort, + snapshotFileName, + type SnapshotManifest, + snapshotTableFields, + writeManifest, +} from '@/clickhouse/parquetSnapshot'; + +describe('seedLogicHash', () => { + it('is a stable 64-char hex sha256', () => { + const h = seedLogicHash(); + expect(h).toMatch(/^[0-9a-f]{64}$/); + }); + + it('is deterministic across calls (content-based, not time-based)', () => { + expect(seedLogicHash()).toBe(seedLogicHash()); + }); + + it('short form is the 12-char prefix', () => { + expect(seedLogicHashShort()).toBe(seedLogicHash().slice(0, 12)); + expect(seedLogicHashShort()).toHaveLength(12); + }); +}); + +describe('snapshot table fields', () => { + it('covers the raw data tables and excludes rollups', () => { + const fields = snapshotTableFields(); + expect(fields).toContain('traces'); + expect(fields).toContain('logs'); + expect(fields).toContain('metricsGauge'); + // Rollup tables must NOT be snapshotted — they repopulate via MVs on load. + expect(fields).not.toContain('tracesKvRollup'); + expect(fields).not.toContain('tracesKeyRollup'); + expect(fields).not.toContain('logsKvRollup'); + expect(fields).not.toContain('logsKeyRollup'); + }); +}); + +describe('snapshotFileName', () => { + it('maps a table name to .parquet', () => { + expect(snapshotFileName('eval_latency_spike_otel_traces')).toBe( + 'eval_latency_spike_otel_traces.parquet', + ); + }); +}); + +describe('manifest round-trip', () => { + let dir: string; + beforeEach(() => { + dir = mkdtempSync(join(tmpdir(), 'hdx-eval-snap-')); + }); + afterEach(() => { + rmSync(dir, { recursive: true, force: true }); + }); + + it('writes and reads back an identical manifest', () => { + const manifest: SnapshotManifest = { + scenarioName: 'latency-spike', + seedLogicHash: seedLogicHash(), + volumeFactor: 1, + seed: 42, + anchorMs: Date.parse('2026-06-01T00:00:00Z'), + anchorIso: '2026-06-01T00:00:00.000Z', + createdAt: new Date().toISOString(), + tables: [ + { table: 'eval_latency_spike_otel_traces', rows: 1000, bytes: 5000 }, + ], + totalRows: 1000, + totalBytes: 5000, + }; + writeManifest(dir, manifest); + expect(readManifest(dir)).toEqual(manifest); + expect(manifestPath(dir)).toBe(join(dir, 'manifest.json')); + }); + + it('returns null when no manifest is present', () => { + expect(readManifest(dir)).toBeNull(); + }); +}); diff --git a/packages/hdx-eval/src/cli.ts b/packages/hdx-eval/src/cli.ts index 845a49d2fd..c8e36c3db9 100644 --- a/packages/hdx-eval/src/cli.ts +++ b/packages/hdx-eval/src/cli.ts @@ -5,10 +5,24 @@ import { join, resolve } from 'path'; import { createEvalClient, defaultClickHouseUrl } from './clickhouse/client'; import { + type ChHttp, + exportScenarioSnapshot, + loadScenarioSnapshot, + readManifest, + seedLogicHash, + seedLogicHashShort, + writeManifest, +} from './clickhouse/parquetSnapshot'; +import { + backfillRollups, + createRollupMaterializedViews, + dropRollupMaterializedViews, dropScenarioTables, + ensureScenarioTables, scenarioIsSeeded, scenarioSlug, scenarioTables, + truncateScenarioTables, } from './clickhouse/schema'; import { buildBlindingEntries } from './grading/blind'; import { @@ -55,6 +69,13 @@ function formatCount(n: number): string { return String(n); } +function formatBytes(n: number): string { + if (n >= 1024 ** 3) return `${(n / 1024 ** 3).toFixed(2)} GB`; + if (n >= 1024 ** 2) return `${(n / 1024 ** 2).toFixed(1)} MB`; + if (n >= 1024) return `${(n / 1024).toFixed(1)} KB`; + return `${n} B`; +} + function logSeedProgress(prefix: string, startMs: number) { return (p: SeedProgress) => { const elapsed = ((Date.now() - startMs) / 1000).toFixed(0); @@ -86,6 +107,15 @@ function buildClient(opts: GlobalOpts) { }); } +/** Raw-HTTP ClickHouse connection info for Parquet snapshot import/export. */ +function buildChHttp(opts: GlobalOpts): ChHttp { + return { + url: opts.chUrl ?? defaultClickHouseUrl(), + username: opts.chUser ?? 'default', + password: opts.chPassword ?? '', + }; +} + function buildClientFromConfig( cfg: import('./hyperdx/config').EvalConfig, globals: GlobalOpts, @@ -257,6 +287,142 @@ program } }); +program + .command('seed-logic-hash') + .description( + 'Print a deterministic hash of the seed-generation source. Used as the CI ' + + 'cache key so the Parquet snapshot is reused only while seeding logic is ' + + 'unchanged.', + ) + .option('--short', 'Print the 12-char short form') + .action((cmdOpts: { short?: boolean }) => { + console.log(cmdOpts.short ? seedLogicHashShort() : seedLogicHash()); + }); + +program + .command('export-snapshot ') + .description( + 'Export the currently-seeded eval tables for a scenario to Parquet files ' + + '(M2 fast-seed snapshot). Writes one
.parquet per non-empty table ' + + 'plus a manifest.json.', + ) + .requiredOption('--dir ', 'Output directory for the Parquet snapshot') + .option( + '--volume-factor ', + 'Record the volume factor this snapshot was generated at (manifest only)', + ) + .option('--seed ', 'Record the PRNG seed used (manifest only)', '42') + .option( + '--anchor ', + 'Record the anchor time this snapshot was generated at (manifest only)', + ) + .action( + async ( + scenarioName: string, + cmdOpts: { + dir: string; + volumeFactor?: string; + seed: string; + anchor?: string; + }, + ) => { + const opts = program.opts(); + const scenario = getScenario(scenarioName); + const http = buildChHttp(opts); + const tables = scenarioTables(scenario.name); + const anchorMs = cmdOpts.anchor ? Date.parse(cmdOpts.anchor) : Date.now(); + const dir = resolve(cmdOpts.dir); + console.log(`Exporting snapshot for "${scenario.name}" → ${dir}`); + const start = Date.now(); + const result = await exportScenarioSnapshot({ + http, + scenarioName: scenario.name, + dir, + tables, + }); + const secs = ((Date.now() - start) / 1000).toFixed(1); + for (const f of result.files) { + console.log( + ` ${f.table}: ${formatCount(f.rows)} rows, ${formatBytes(f.bytes)}`, + ); + } + writeManifest(dir, { + scenarioName: scenario.name, + seedLogicHash: seedLogicHash(), + volumeFactor: cmdOpts.volumeFactor ? Number(cmdOpts.volumeFactor) : 1, + seed: Number(cmdOpts.seed), + anchorMs, + anchorIso: new Date(anchorMs).toISOString(), + createdAt: new Date().toISOString(), + tables: result.files.map(f => ({ + table: f.table, + rows: f.rows, + bytes: f.bytes, + })), + totalRows: result.totalRows, + totalBytes: result.totalBytes, + }); + console.log( + `Exported ${formatCount(result.totalRows)} rows, ${formatBytes(result.totalBytes)} in ${secs}s`, + ); + console.log(`Wrote manifest → ${dir}/manifest.json`); + }, + ); + +program + .command('load-snapshot ') + .description( + 'Load a Parquet snapshot for a scenario into ClickHouse (M2 fast-seed path). ' + + 'Ensures tables + rollup MVs exist, truncates, then inserts each Parquet ' + + 'file — rollups repopulate automatically via the MVs.', + ) + .requiredOption('--dir ', 'Directory containing the Parquet snapshot') + .action(async (scenarioName: string, cmdOpts: { dir: string }) => { + const opts = program.opts(); + const scenario = getScenario(scenarioName); + const http = buildChHttp(opts); + const dir = resolve(cmdOpts.dir); + const client = buildClient(opts); + const manifest = readManifest(dir); + if (manifest) { + console.log( + `Loading snapshot for "${scenario.name}" (built ${manifest.createdAt}, ` + + `anchor ${manifest.anchorIso}, volumeFactor ${manifest.volumeFactor})`, + ); + if (manifest.scenarioName !== scenario.name) { + throw new Error( + `Snapshot manifest is for scenario "${manifest.scenarioName}", not "${scenario.name}"`, + ); + } + } else { + console.log( + `Loading snapshot for "${scenario.name}" (no manifest.json found in ${dir})`, + ); + } + try { + const start = Date.now(); + const result = await loadScenarioSnapshot({ + http, + ensure: () => ensureScenarioTables(client, scenario.name), + truncate: () => truncateScenarioTables(client, scenario.name), + dropMaterializedViews: tables => + dropRollupMaterializedViews(client, tables), + backfillRollups: tables => backfillRollups(client, tables), + createMaterializedViews: tables => + createRollupMaterializedViews(client, tables), + scenarioName: scenario.name, + dir, + }); + const secs = ((Date.now() - start) / 1000).toFixed(1); + for (const f of result.files) { + console.log(` ${f.table}: ${formatCount(f.rows)} rows`); + } + console.log(`Loaded ${formatCount(result.totalRows)} rows in ${secs}s`); + } finally { + await client.close(); + } + }); + program .command('setup-hyperdx') .description( diff --git a/packages/hdx-eval/src/clickhouse/parquetSnapshot.ts b/packages/hdx-eval/src/clickhouse/parquetSnapshot.ts new file mode 100644 index 0000000000..73b882dc13 --- /dev/null +++ b/packages/hdx-eval/src/clickhouse/parquetSnapshot.ts @@ -0,0 +1,350 @@ +import { createHash } from 'crypto'; +import { + createReadStream, + createWriteStream, + existsSync, + mkdirSync, + readdirSync, + readFileSync, + statSync, + writeFileSync, +} from 'fs'; +import { dirname, join, resolve } from 'path'; +import { pipeline } from 'stream/promises'; + +import { EVAL_DATABASE, type ScenarioTables } from './schema'; + +/** + * Milestone 2 — kill the seed bottleneck. + * + * Full-volume generation of synthetic telemetry takes hours because it is a + * CPU-bound JS loop. Instead we generate the data ONCE, export each eval table + * to a Parquet file, and on subsequent runs load those Parquet files straight + * into ClickHouse (I/O-bound, ~an order of magnitude faster). The Parquet + * snapshot is keyed by a hash of the seed-generation source, so it is only + * regenerated when the seeding logic actually changes. + * + * The rollup metadata tables are intentionally NOT part of the snapshot: they + * are repopulated automatically by the materialized views attached to the raw + * tables when we INSERT the Parquet rows (same path as a live seed insert). + */ + +// ClickHouse HTTP connection derived from the eval client config. We use raw +// HTTP (not the @clickhouse/client insert API) because Parquet is a binary +// stream we want to pipe directly to/from disk without buffering in memory. +export type ChHttp = { + url: string; // base HTTP URL, e.g. http://localhost:8123 + username: string; + password: string; +}; + +/** The raw data tables that make up a snapshot (rollups are excluded). */ +export function snapshotTableFields(): Array { + return [ + 'traces', + 'logs', + 'metricsGauge', + 'metricsSum', + 'metricsHistogram', + 'metricsExponentialHistogram', + 'metricsSummary', + ]; +} + +function authHeaders(http: ChHttp): Record { + return { + 'X-ClickHouse-User': http.username, + 'X-ClickHouse-Key': http.password, + }; +} + +async function chExec(http: ChHttp, sql: string): Promise { + const res = await fetch(http.url, { + method: 'POST', + headers: { ...authHeaders(http), 'Content-Type': 'text/plain' }, + body: sql, + }); + const text = await res.text(); + if (!res.ok) { + throw new Error( + `ClickHouse query failed (${res.status}): ${text.slice(0, 500)}`, + ); + } + return text; +} + +async function tableRowCount(http: ChHttp, table: string): Promise { + const out = await chExec( + http, + `SELECT count() FROM ${EVAL_DATABASE}.${table}`, + ); + return Number(out.trim()) || 0; +} + +/** Filename for a table's Parquet file within a snapshot directory. */ +export function snapshotFileName(table: string): string { + return `${table}.parquet`; +} + +export type ExportResult = { + files: Array<{ table: string; path: string; rows: number; bytes: number }>; + totalRows: number; + totalBytes: number; +}; + +/** + * Export every non-empty snapshot table for a scenario to Parquet files under + * `dir`. Empty tables are skipped (no file written) so the loader can treat a + * missing file as "zero rows". + */ +export async function exportScenarioSnapshot(args: { + http: ChHttp; + scenarioName: string; + dir: string; + tables: ScenarioTables; +}): Promise { + mkdirSync(args.dir, { recursive: true }); + const files: ExportResult['files'] = []; + let totalRows = 0; + let totalBytes = 0; + + for (const field of snapshotTableFields()) { + const table = args.tables[field]; + const rows = await tableRowCount(args.http, table); + if (rows === 0) continue; + + const outPath = join(args.dir, snapshotFileName(table)); + // zstd is ClickHouse's default Parquet codec and already near the size + // floor for this data; set it explicitly so snapshots are reproducible + // regardless of server defaults. + const sql = + `SELECT * FROM ${EVAL_DATABASE}.${table} ` + + `FORMAT Parquet ` + + `SETTINGS output_format_parquet_compression_method='zstd'`; + const res = await fetch(args.http.url, { + method: 'POST', + headers: { ...authHeaders(args.http), 'Content-Type': 'text/plain' }, + body: sql, + }); + if (!res.ok || !res.body) { + const errText = await res.text().catch(() => ''); + throw new Error( + `Parquet export of ${table} failed (${res.status}): ${errText.slice(0, 500)}`, + ); + } + mkdirSync(dirname(outPath), { recursive: true }); + // Pipe the response stream directly to disk to avoid buffering large + // Parquet payloads in memory. + await pipeline(res.body, createWriteStream(outPath)); + const bytes = statSync(outPath).size; + files.push({ table, path: outPath, rows, bytes }); + totalRows += rows; + totalBytes += bytes; + } + + return { files, totalRows, totalBytes }; +} + +export type LoadResult = { + files: Array<{ table: string; rows: number }>; + totalRows: number; +}; + +/** Stream one Parquet file from disk into an INSERT over the HTTP interface. */ +async function insertParquetFile( + http: ChHttp, + table: string, + filePath: string, +): Promise { + const query = `INSERT INTO ${EVAL_DATABASE}.${table} FORMAT Parquet`; + const url = `${http.url}/?query=${encodeURIComponent(query)}`; + // Node's fetch accepts a Readable body but requires `duplex: 'half'`; neither + // is in the DOM RequestInit type, so build the init as `unknown` and cast. + const init = { + method: 'POST', + headers: { + ...authHeaders(http), + 'Content-Type': 'application/octet-stream', + }, + body: createReadStream(filePath), + duplex: 'half', + } as unknown as RequestInit; + const res = await fetch(url, init); + if (!res.ok) { + const errText = await res.text().catch(() => ''); + throw new Error( + `Parquet load of ${table} failed (${res.status}): ${errText.slice(0, 500)}`, + ); + } +} + +/** + * Load a previously exported Parquet snapshot for a scenario into ClickHouse. + * + * Fast path: rather than let the rollup materialized views fan out on the bulk + * insert (aggregating + writing the SummingMergeTree rollups per insert block, + * the dominant cost of a large load), we + * 1. drop the rollup MVs, + * 2. bulk-insert the raw Parquet with no fan-out, + * 3. backfill the rollups in one shot (same SELECTs the MVs run), and + * 4. recreate the MVs so any later inserts still maintain them. + * This produces byte-identical rollup content and is meaningfully faster at + * scale. The drop/backfill/recreate steps are supplied by the caller (they + * need the @clickhouse/client instance) via the same callback pattern as + * ensure/truncate. + */ +export async function loadScenarioSnapshot(args: { + http: ChHttp; + // DDL callbacks backed by the @clickhouse/client instance. + ensure: () => Promise; + truncate: () => Promise; + dropMaterializedViews: (tables: ScenarioTables) => Promise; + backfillRollups: (tables: ScenarioTables) => Promise; + createMaterializedViews: (tables: ScenarioTables) => Promise; + scenarioName: string; + dir: string; +}): Promise { + if (!existsSync(args.dir)) { + throw new Error(`Snapshot directory not found: ${args.dir}`); + } + const tables = await args.ensure(); + await args.truncate(); + + // Detach the MVs so the raw bulk insert below does not trigger per-block + // rollup maintenance. + await args.dropMaterializedViews(tables); + + const files: LoadResult['files'] = []; + let totalRows = 0; + + for (const field of snapshotTableFields()) { + const table = tables[field]; + const filePath = join(args.dir, snapshotFileName(table)); + if (!existsSync(filePath)) continue; // table had zero rows at export time + + await insertParquetFile(args.http, table, filePath); + const rows = await tableRowCount(args.http, table); + files.push({ table, rows }); + totalRows += rows; + } + + // Rebuild the rollups in one shot from the loaded raw tables, then re-attach + // the MVs for consistency with a live-seeded scenario. + await args.backfillRollups(tables); + await args.createMaterializedViews(tables); + + return { files, totalRows }; +} + +/** + * Metadata written next to a snapshot so a loader can verify what it is + * loading (scenario, anchor, volume, and the seed-logic hash it was built + * from). + */ +export type SnapshotManifest = { + scenarioName: string; + seedLogicHash: string; + volumeFactor: number; + seed: number; + anchorMs: number; + anchorIso: string; + createdAt: string; + tables: Array<{ table: string; rows: number; bytes: number }>; + totalRows: number; + totalBytes: number; +}; + +export function manifestPath(dir: string): string { + return join(dir, 'manifest.json'); +} + +export function writeManifest(dir: string, manifest: SnapshotManifest): void { + mkdirSync(dir, { recursive: true }); + writeFileSync( + manifestPath(dir), + JSON.stringify(manifest, null, 2) + '\n', + 'utf8', + ); +} + +export function readManifest(dir: string): SnapshotManifest | null { + const p = manifestPath(dir); + if (!existsSync(p)) return null; + try { + return JSON.parse(readFileSync(p, 'utf8')) as SnapshotManifest; + } catch { + return null; + } +} + +// --------------------------------------------------------------------------- +// Seed-logic hash +// --------------------------------------------------------------------------- + +/** + * Files whose content determines the generated seed data. When ANY of these + * change, the Parquet snapshot must be regenerated. Used as the cache key in + * CI so the snapshot is reused only while the seeding logic is unchanged. + * + * Paths are relative to the hdx-eval package root (src/...). + */ +const SEED_LOGIC_GLOBS: string[] = [ + 'src/generators', + 'src/scenarios', + 'src/rng', + 'src/clickhouse/insert.ts', + 'src/clickhouse/schema.ts', + 'src/clickhouse/parquetSnapshot.ts', +]; + +function packageRoot(): string { + // dist/clickhouse/parquetSnapshot.js → dist → packageRoot, or + // src/clickhouse/parquetSnapshot.ts via tsx → src → packageRoot. + return resolve(__dirname, '..', '..'); +} + +/** Recursively collect .ts/.json files under a path (or a single file). */ +function collectFiles(absPath: string): string[] { + if (!existsSync(absPath)) return []; + const st = statSync(absPath); + if (st.isFile()) return [absPath]; + const out: string[] = []; + for (const entry of readdirSync(absPath).sort()) { + const child = join(absPath, entry); + const cst = statSync(child); + if (cst.isDirectory()) { + out.push(...collectFiles(child)); + } else if (/\.(ts|json)$/.test(entry) && !/\.test\.ts$/.test(entry)) { + out.push(child); + } + } + return out; +} + +/** + * Deterministic hash of all seed-generation source files. Stable across + * machines: hashes file contents (not mtimes) in sorted path order, with the + * package-relative path mixed in so renames change the hash. + */ +export function seedLogicHash(): string { + const root = packageRoot(); + const hash = createHash('sha256'); + const files: string[] = []; + for (const glob of SEED_LOGIC_GLOBS) { + files.push(...collectFiles(join(root, glob))); + } + files.sort(); + for (const abs of files) { + const rel = abs.slice(root.length + 1).replace(/\\/g, '/'); + hash.update(rel); + hash.update('\0'); + hash.update(readFileSync(abs)); + hash.update('\0'); + } + return hash.digest('hex'); +} + +/** Short (12-char) form of the seed-logic hash for cache keys / tags. */ +export function seedLogicHashShort(): string { + return seedLogicHash().slice(0, 12); +} diff --git a/packages/hdx-eval/src/clickhouse/schema.ts b/packages/hdx-eval/src/clickhouse/schema.ts index 3a76303a11..a5af286bfb 100644 --- a/packages/hdx-eval/src/clickhouse/schema.ts +++ b/packages/hdx-eval/src/clickhouse/schema.ts @@ -391,6 +391,90 @@ async function ensureRollupTables( }); } +/** + * Drop only the rollup materialized views (not the rollup tables). Used by the + * Parquet snapshot load path so a bulk raw insert does NOT trigger the MV + * fan-out (which is the dominant cost of a large load — each inserted block is + * aggregated and written to the SummingMergeTree rollups per-block, then merged). + * After the raw load we backfill the rollups in one shot and recreate the MVs. + */ +export async function dropRollupMaterializedViews( + client: ClickHouseClient, + tables: ScenarioTables, +): Promise { + const db = EVAL_DATABASE; + for (const mv of [ + mvName(tables.tracesKvRollup), + keyMvName(tables.tracesKeyRollup), + mvName(tables.logsKvRollup), + keyMvName(tables.logsKeyRollup), + ]) { + await client.command({ query: `DROP VIEW IF EXISTS ${db}.${mv}` }); + } +} + +/** (Re)create just the rollup materialized views for a scenario's tables. */ +export async function createRollupMaterializedViews( + client: ClickHouseClient, + tables: ScenarioTables, +): Promise { + const db = EVAL_DATABASE; + await client.command({ + query: `CREATE MATERIALIZED VIEW IF NOT EXISTS ${db}.${mvName(tables.tracesKvRollup)} TO ${db}.${tables.tracesKvRollup} AS ${tracesKvMvSelect(db, tables.traces)}`, + }); + await client.command({ + query: `CREATE MATERIALIZED VIEW IF NOT EXISTS ${db}.${keyMvName(tables.tracesKeyRollup)} TO ${db}.${tables.tracesKeyRollup} AS ${keyRollupMvSelect(db, tables.tracesKvRollup)}`, + }); + await client.command({ + query: `CREATE MATERIALIZED VIEW IF NOT EXISTS ${db}.${mvName(tables.logsKvRollup)} TO ${db}.${tables.logsKvRollup} AS ${logsKvMvSelect(db, tables.logs)}`, + }); + await client.command({ + query: `CREATE MATERIALIZED VIEW IF NOT EXISTS ${db}.${keyMvName(tables.logsKeyRollup)} TO ${db}.${tables.logsKeyRollup} AS ${keyRollupMvSelect(db, tables.logsKvRollup)}`, + }); +} + +/** + * One-shot bulk backfill of the rollup tables from already-loaded raw tables, + * reusing the exact same SELECTs the materialized views run. This is the fast + * alternative to letting the MVs fan out on a bulk Parquet insert: aggregating + * the whole table once is far cheaper than maintaining the SummingMergeTree + * incrementally per insert block. Produces byte-identical rollup content. + * + * Only backfills a raw table's rollups when that raw table has rows (so a + * traces-only scenario doesn't write empty log rollups). + */ +export async function backfillRollups( + client: ClickHouseClient, + tables: ScenarioTables, +): Promise { + const db = EVAL_DATABASE; + + const hasRows = async (table: string): Promise => { + const rs = await client.query({ + query: `SELECT 1 FROM ${db}.${table} LIMIT 1`, + format: 'JSONEachRow', + }); + return (await rs.json()).length > 0; + }; + + if (await hasRows(tables.traces)) { + await client.command({ + query: `INSERT INTO ${db}.${tables.tracesKvRollup} ${tracesKvMvSelect(db, tables.traces)}`, + }); + await client.command({ + query: `INSERT INTO ${db}.${tables.tracesKeyRollup} ${keyRollupMvSelect(db, tables.tracesKvRollup)}`, + }); + } + if (await hasRows(tables.logs)) { + await client.command({ + query: `INSERT INTO ${db}.${tables.logsKvRollup} ${logsKvMvSelect(db, tables.logs)}`, + }); + await client.command({ + query: `INSERT INTO ${db}.${tables.logsKeyRollup} ${keyRollupMvSelect(db, tables.logsKvRollup)}`, + }); + } +} + async function dropRollupTables( client: ClickHouseClient, tables: ScenarioTables,