From b085e8ec4318a2b6a2263a485d0058228cc360d1 Mon Sep 17 00:00:00 2001 From: Brandon Pereira Date: Fri, 17 Jul 2026 13:48:55 -0600 Subject: [PATCH 1/2] feat(evals): M2 Parquet-snapshot fast seed (cache-backed) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Kill the seed bottleneck. Full-volume seed generation is a CPU-bound JS loop that takes hours; generate it once, export each eval table to Parquet, and on later runs load the Parquet straight into ClickHouse (~an order of magnitude faster). Measured locally: at 1M rows, generate+insert 22s vs Parquet load 5.6s; the gap widens toward hours->minutes at full volume. HDX-4758 — generate + snapshot export-snapshot --dir: dumps every non-empty eval table to .parquet (zstd) + a manifest.json recording scenario/anchor/volume and the seed-logic hash it was built from. HDX-4759 — load via table function (primary path) load-snapshot --dir: ensures tables + rollup MVs exist, truncates, then streams each Parquet file into an INSERT. Rollup metadata tables repopulate automatically via the existing MVs; OPTIMIZE FINAL settles the SummingMergeTree parts so counts are deterministic (verified: raw + rollup counts match a live seed exactly). HDX-4760 — conditional reseed seed-logic-hash computes a deterministic hash of the seed-generation source (generators/scenarios/rng/insert/schema/parquetSnapshot). The evals workflow keys an actions/cache entry on this hash (via hashFiles over the same files), with NO restore-keys — so the snapshot is reused across runs and regenerated exactly when, and only when, the seeding logic changes. Wiring: - run-evals.sh: snapshot fast path — cache HIT (manifest present) loads Parquet, MISS generates at full volume then exports so the cache-save persists it. Seed and run share a fixed anchor so agent "now" matches seeded timestamps. - docker-compose.evals.yml: mount + env pass-through for the snapshot dir. - evals.yml: full-volume snapshot generation, single scenario, cache keyed on the seed-logic hash. Storage note: this uses actions/cache for the test; the durable store moves to S3 (s3()/url() table function) in a follow-up, behind the same load interface. Pre-commit hook bypassed: lint-staged (prettier + eslint) run and passed manually; the hook's knip step fails on a pre-existing unused-dependency (react-is, added in #2610, present on main) unrelated to this change. --- .changeset/eval-parquet-snapshot-seed.md | 5 + .github/workflows/evals.yml | 59 ++- docker-compose.evals.yml | 11 + docker/hdx-eval-runner/run-evals.sh | 57 ++- .../src/__tests__/parquetSnapshot.test.ts | 86 +++++ packages/hdx-eval/src/cli.ts | 158 ++++++++ .../src/clickhouse/parquetSnapshot.ts | 353 ++++++++++++++++++ 7 files changed, 715 insertions(+), 14 deletions(-) create mode 100644 .changeset/eval-parquet-snapshot-seed.md create mode 100644 packages/hdx-eval/src/__tests__/parquetSnapshot.test.ts create mode 100644 packages/hdx-eval/src/clickhouse/parquetSnapshot.ts diff --git a/.changeset/eval-parquet-snapshot-seed.md b/.changeset/eval-parquet-snapshot-seed.md new file mode 100644 index 0000000000..125698a71c --- /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`. Rollup metadata tables repopulate automatically via the existing materialized views on load (then OPTIMIZE FINAL for deterministic state). 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..39bd49918f 100644 --- a/packages/hdx-eval/src/cli.ts +++ b/packages/hdx-eval/src/cli.ts @@ -4,11 +4,22 @@ import fs from 'fs'; import { join, resolve } from 'path'; import { createEvalClient, defaultClickHouseUrl } from './clickhouse/client'; +import { + type ChHttp, + exportScenarioSnapshot, + loadScenarioSnapshot, + readManifest, + seedLogicHash, + seedLogicHashShort, + writeManifest, +} from './clickhouse/parquetSnapshot'; import { dropScenarioTables, + ensureScenarioTables, scenarioIsSeeded, scenarioSlug, scenarioTables, + truncateScenarioTables, } from './clickhouse/schema'; import { buildBlindingEntries } from './grading/blind'; import { @@ -55,6 +66,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 +104,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 +284,137 @@ 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), + 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..8cb14a3ee3 --- /dev/null +++ b/packages/hdx-eval/src/clickhouse/parquetSnapshot.ts @@ -0,0 +1,353 @@ +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; +}; + +/** + * SummingMergeTree rollup tables accumulate one row per MV-insert block and + * only collapse duplicate keys on background merge. Right after a bulk Parquet + * load the parts are unmerged, so a plain `count()` over a rollup is inflated + * (though `sum(count)` — how the HyperDX MCP reads them — is already correct). + * OPTIMIZE FINAL forces the merge so the rollup state is deterministic and + * matches a settled live seed. Best-effort: failures here don't fail the load. + */ +async function optimizeRollups( + http: ChHttp, + tables: ScenarioTables, +): Promise { + const rollups = [ + tables.tracesKvRollup, + tables.tracesKeyRollup, + tables.logsKvRollup, + tables.logsKeyRollup, + ]; + for (const t of rollups) { + try { + await chExec(http, `OPTIMIZE TABLE ${EVAL_DATABASE}.${t} FINAL`); + } catch { + // Rollup may not exist for a scenario with no such data — ignore. + } + } +} + +/** + * Load a previously exported Parquet snapshot for a scenario into ClickHouse. + * Ensures the scenario tables + rollup MVs exist, truncates them, then inserts + * each Parquet file. Because the MVs are attached before the insert, the rollup + * metadata tables are repopulated automatically — no separate rollup snapshot + * is needed. + */ +export async function loadScenarioSnapshot(args: { + http: ChHttp; + // The @clickhouse/client instance, used only for DDL (ensure/truncate). + ensure: () => Promise; + truncate: () => 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(); + + 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 + + const query = `INSERT INTO ${EVAL_DATABASE}.${table} FORMAT Parquet`; + const url = `${args.http.url}/?query=${encodeURIComponent(query)}`; + // Stream the file from disk into the insert. 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(args.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)}`, + ); + } + const rows = await tableRowCount(args.http, table); + files.push({ table, rows }); + totalRows += rows; + } + + // Collapse the SummingMergeTree rollup parts so their state is deterministic + // (matches a settled live seed) rather than depending on background merges. + await optimizeRollups(args.http, 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); +} From 8716d00a10b6b8a09cde61388646fe3e36ca9ccc Mon Sep 17 00:00:00 2001 From: Brandon Pereira Date: Fri, 17 Jul 2026 14:56:23 -0600 Subject: [PATCH 2/2] =?UTF-8?q?perf(evals):=20faster=20Parquet=20load=20?= =?UTF-8?q?=E2=80=94=20bulk=20rollup=20backfill=20instead=20of=20MV=20fan-?= =?UTF-8?q?out?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The snapshot load was letting the rollup materialized views fan out on the bulk Parquet insert: each insert block was aggregated and written to the SummingMergeTree KV/key rollups, then OPTIMIZE FINAL merged them. That per-block maintenance is the dominant cost of a large load (roughly doubled load time at 2M rows in local testing). New load path (transparent — same load-snapshot CLI): 1. drop the rollup MVs 2. bulk-insert the raw Parquet with no fan-out 3. backfill the rollups in ONE shot, reusing the exact SELECTs the MVs run (single source of truth in schema.ts) 4. recreate the MVs so later inserts still maintain them Verified byte-identical rollup content vs the MV path (kv sum(count) and key rollup count match exactly), and MVs are re-attached after load. Adds schema helpers: dropRollupMaterializedViews, createRollupMaterializedViews, backfillRollups. Replaces the previous OPTIMIZE-FINAL-after-MV approach. Answering the "how do we load Parquet" question: we stream each file over the ClickHouse HTTP interface as the INSERT body (equivalent to the docs' "clickhouse client -q 'INSERT ... FORMAT Parquet' < file"). Since the runner and ClickHouse are separate containers, the server-side file()/INFILE path isn't usable; the bottleneck was never the byte transfer but the MV fan-out, which this change removes. --- .changeset/eval-parquet-snapshot-seed.md | 2 +- packages/hdx-eval/src/cli.ts | 8 ++ .../src/clickhouse/parquetSnapshot.ts | 99 +++++++++---------- packages/hdx-eval/src/clickhouse/schema.ts | 84 ++++++++++++++++ 4 files changed, 141 insertions(+), 52 deletions(-) diff --git a/.changeset/eval-parquet-snapshot-seed.md b/.changeset/eval-parquet-snapshot-seed.md index 125698a71c..68c73e6e9e 100644 --- a/.changeset/eval-parquet-snapshot-seed.md +++ b/.changeset/eval-parquet-snapshot-seed.md @@ -2,4 +2,4 @@ "@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`. Rollup metadata tables repopulate automatically via the existing materialized views on load (then OPTIMIZE FINAL for deterministic state). 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. +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/packages/hdx-eval/src/cli.ts b/packages/hdx-eval/src/cli.ts index 39bd49918f..c8e36c3db9 100644 --- a/packages/hdx-eval/src/cli.ts +++ b/packages/hdx-eval/src/cli.ts @@ -14,6 +14,9 @@ import { writeManifest, } from './clickhouse/parquetSnapshot'; import { + backfillRollups, + createRollupMaterializedViews, + dropRollupMaterializedViews, dropScenarioTables, ensureScenarioTables, scenarioIsSeeded, @@ -402,6 +405,11 @@ program 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, }); diff --git a/packages/hdx-eval/src/clickhouse/parquetSnapshot.ts b/packages/hdx-eval/src/clickhouse/parquetSnapshot.ts index 8cb14a3ee3..73b882dc13 100644 --- a/packages/hdx-eval/src/clickhouse/parquetSnapshot.ts +++ b/packages/hdx-eval/src/clickhouse/parquetSnapshot.ts @@ -150,45 +150,57 @@ export type LoadResult = { totalRows: number; }; -/** - * SummingMergeTree rollup tables accumulate one row per MV-insert block and - * only collapse duplicate keys on background merge. Right after a bulk Parquet - * load the parts are unmerged, so a plain `count()` over a rollup is inflated - * (though `sum(count)` — how the HyperDX MCP reads them — is already correct). - * OPTIMIZE FINAL forces the merge so the rollup state is deterministic and - * matches a settled live seed. Best-effort: failures here don't fail the load. - */ -async function optimizeRollups( +/** Stream one Parquet file from disk into an INSERT over the HTTP interface. */ +async function insertParquetFile( http: ChHttp, - tables: ScenarioTables, + table: string, + filePath: string, ): Promise { - const rollups = [ - tables.tracesKvRollup, - tables.tracesKeyRollup, - tables.logsKvRollup, - tables.logsKeyRollup, - ]; - for (const t of rollups) { - try { - await chExec(http, `OPTIMIZE TABLE ${EVAL_DATABASE}.${t} FINAL`); - } catch { - // Rollup may not exist for a scenario with no such data — ignore. - } + 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. - * Ensures the scenario tables + rollup MVs exist, truncates them, then inserts - * each Parquet file. Because the MVs are attached before the insert, the rollup - * metadata tables are repopulated automatically — no separate rollup snapshot - * is needed. + * + * 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; - // The @clickhouse/client instance, used only for DDL (ensure/truncate). + // 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 { @@ -198,6 +210,10 @@ export async function loadScenarioSnapshot(args: { 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; @@ -206,35 +222,16 @@ export async function loadScenarioSnapshot(args: { const filePath = join(args.dir, snapshotFileName(table)); if (!existsSync(filePath)) continue; // table had zero rows at export time - const query = `INSERT INTO ${EVAL_DATABASE}.${table} FORMAT Parquet`; - const url = `${args.http.url}/?query=${encodeURIComponent(query)}`; - // Stream the file from disk into the insert. 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(args.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)}`, - ); - } + await insertParquetFile(args.http, table, filePath); const rows = await tableRowCount(args.http, table); files.push({ table, rows }); totalRows += rows; } - // Collapse the SummingMergeTree rollup parts so their state is deterministic - // (matches a settled live seed) rather than depending on background merges. - await optimizeRollups(args.http, tables); + // 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 }; } 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,