Skip to content

Commit 8716d00

Browse files
perf(evals): faster Parquet load — bulk rollup backfill instead of MV fan-out
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.
1 parent b085e8e commit 8716d00

4 files changed

Lines changed: 141 additions & 52 deletions

File tree

‎.changeset/eval-parquet-snapshot-seed.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,4 +2,4 @@
22
"@hyperdx/hdx-eval": minor
33
---
44

5-
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.
5+
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.

‎packages/hdx-eval/src/cli.ts‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,9 @@ import {
1414
writeManifest,
1515
} from './clickhouse/parquetSnapshot';
1616
import {
17+
backfillRollups,
18+
createRollupMaterializedViews,
19+
dropRollupMaterializedViews,
1720
dropScenarioTables,
1821
ensureScenarioTables,
1922
scenarioIsSeeded,
@@ -402,6 +405,11 @@ program
402405
http,
403406
ensure: () => ensureScenarioTables(client, scenario.name),
404407
truncate: () => truncateScenarioTables(client, scenario.name),
408+
dropMaterializedViews: tables =>
409+
dropRollupMaterializedViews(client, tables),
410+
backfillRollups: tables => backfillRollups(client, tables),
411+
createMaterializedViews: tables =>
412+
createRollupMaterializedViews(client, tables),
405413
scenarioName: scenario.name,
406414
dir,
407415
});

‎packages/hdx-eval/src/clickhouse/parquetSnapshot.ts‎

Lines changed: 48 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -150,45 +150,57 @@ export type LoadResult = {
150150
totalRows: number;
151151
};
152152

153-
/**
154-
* SummingMergeTree rollup tables accumulate one row per MV-insert block and
155-
* only collapse duplicate keys on background merge. Right after a bulk Parquet
156-
* load the parts are unmerged, so a plain `count()` over a rollup is inflated
157-
* (though `sum(count)` — how the HyperDX MCP reads them — is already correct).
158-
* OPTIMIZE FINAL forces the merge so the rollup state is deterministic and
159-
* matches a settled live seed. Best-effort: failures here don't fail the load.
160-
*/
161-
async function optimizeRollups(
153+
/** Stream one Parquet file from disk into an INSERT over the HTTP interface. */
154+
async function insertParquetFile(
162155
http: ChHttp,
163-
tables: ScenarioTables,
156+
table: string,
157+
filePath: string,
164158
): Promise<void> {
165-
const rollups = [
166-
tables.tracesKvRollup,
167-
tables.tracesKeyRollup,
168-
tables.logsKvRollup,
169-
tables.logsKeyRollup,
170-
];
171-
for (const t of rollups) {
172-
try {
173-
await chExec(http, `OPTIMIZE TABLE ${EVAL_DATABASE}.${t} FINAL`);
174-
} catch {
175-
// Rollup may not exist for a scenario with no such data — ignore.
176-
}
159+
const query = `INSERT INTO ${EVAL_DATABASE}.${table} FORMAT Parquet`;
160+
const url = `${http.url}/?query=${encodeURIComponent(query)}`;
161+
// Node's fetch accepts a Readable body but requires `duplex: 'half'`; neither
162+
// is in the DOM RequestInit type, so build the init as `unknown` and cast.
163+
const init = {
164+
method: 'POST',
165+
headers: {
166+
...authHeaders(http),
167+
'Content-Type': 'application/octet-stream',
168+
},
169+
body: createReadStream(filePath),
170+
duplex: 'half',
171+
} as unknown as RequestInit;
172+
const res = await fetch(url, init);
173+
if (!res.ok) {
174+
const errText = await res.text().catch(() => '');
175+
throw new Error(
176+
`Parquet load of ${table} failed (${res.status}): ${errText.slice(0, 500)}`,
177+
);
177178
}
178179
}
179180

180181
/**
181182
* Load a previously exported Parquet snapshot for a scenario into ClickHouse.
182-
* Ensures the scenario tables + rollup MVs exist, truncates them, then inserts
183-
* each Parquet file. Because the MVs are attached before the insert, the rollup
184-
* metadata tables are repopulated automatically — no separate rollup snapshot
185-
* is needed.
183+
*
184+
* Fast path: rather than let the rollup materialized views fan out on the bulk
185+
* insert (aggregating + writing the SummingMergeTree rollups per insert block,
186+
* the dominant cost of a large load), we
187+
* 1. drop the rollup MVs,
188+
* 2. bulk-insert the raw Parquet with no fan-out,
189+
* 3. backfill the rollups in one shot (same SELECTs the MVs run), and
190+
* 4. recreate the MVs so any later inserts still maintain them.
191+
* This produces byte-identical rollup content and is meaningfully faster at
192+
* scale. The drop/backfill/recreate steps are supplied by the caller (they
193+
* need the @clickhouse/client instance) via the same callback pattern as
194+
* ensure/truncate.
186195
*/
187196
export async function loadScenarioSnapshot(args: {
188197
http: ChHttp;
189-
// The @clickhouse/client instance, used only for DDL (ensure/truncate).
198+
// DDL callbacks backed by the @clickhouse/client instance.
190199
ensure: () => Promise<ScenarioTables>;
191200
truncate: () => Promise<void>;
201+
dropMaterializedViews: (tables: ScenarioTables) => Promise<void>;
202+
backfillRollups: (tables: ScenarioTables) => Promise<void>;
203+
createMaterializedViews: (tables: ScenarioTables) => Promise<void>;
192204
scenarioName: string;
193205
dir: string;
194206
}): Promise<LoadResult> {
@@ -198,6 +210,10 @@ export async function loadScenarioSnapshot(args: {
198210
const tables = await args.ensure();
199211
await args.truncate();
200212

213+
// Detach the MVs so the raw bulk insert below does not trigger per-block
214+
// rollup maintenance.
215+
await args.dropMaterializedViews(tables);
216+
201217
const files: LoadResult['files'] = [];
202218
let totalRows = 0;
203219

@@ -206,35 +222,16 @@ export async function loadScenarioSnapshot(args: {
206222
const filePath = join(args.dir, snapshotFileName(table));
207223
if (!existsSync(filePath)) continue; // table had zero rows at export time
208224

209-
const query = `INSERT INTO ${EVAL_DATABASE}.${table} FORMAT Parquet`;
210-
const url = `${args.http.url}/?query=${encodeURIComponent(query)}`;
211-
// Stream the file from disk into the insert. Node's fetch accepts a
212-
// Readable body but requires `duplex: 'half'`; neither is in the DOM
213-
// RequestInit type, so build the init as `unknown` and cast.
214-
const init = {
215-
method: 'POST',
216-
headers: {
217-
...authHeaders(args.http),
218-
'Content-Type': 'application/octet-stream',
219-
},
220-
body: createReadStream(filePath),
221-
duplex: 'half',
222-
} as unknown as RequestInit;
223-
const res = await fetch(url, init);
224-
if (!res.ok) {
225-
const errText = await res.text().catch(() => '');
226-
throw new Error(
227-
`Parquet load of ${table} failed (${res.status}): ${errText.slice(0, 500)}`,
228-
);
229-
}
225+
await insertParquetFile(args.http, table, filePath);
230226
const rows = await tableRowCount(args.http, table);
231227
files.push({ table, rows });
232228
totalRows += rows;
233229
}
234230

235-
// Collapse the SummingMergeTree rollup parts so their state is deterministic
236-
// (matches a settled live seed) rather than depending on background merges.
237-
await optimizeRollups(args.http, tables);
231+
// Rebuild the rollups in one shot from the loaded raw tables, then re-attach
232+
// the MVs for consistency with a live-seeded scenario.
233+
await args.backfillRollups(tables);
234+
await args.createMaterializedViews(tables);
238235

239236
return { files, totalRows };
240237
}

‎packages/hdx-eval/src/clickhouse/schema.ts‎

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -391,6 +391,90 @@ async function ensureRollupTables(
391391
});
392392
}
393393

394+
/**
395+
* Drop only the rollup materialized views (not the rollup tables). Used by the
396+
* Parquet snapshot load path so a bulk raw insert does NOT trigger the MV
397+
* fan-out (which is the dominant cost of a large load — each inserted block is
398+
* aggregated and written to the SummingMergeTree rollups per-block, then merged).
399+
* After the raw load we backfill the rollups in one shot and recreate the MVs.
400+
*/
401+
export async function dropRollupMaterializedViews(
402+
client: ClickHouseClient,
403+
tables: ScenarioTables,
404+
): Promise<void> {
405+
const db = EVAL_DATABASE;
406+
for (const mv of [
407+
mvName(tables.tracesKvRollup),
408+
keyMvName(tables.tracesKeyRollup),
409+
mvName(tables.logsKvRollup),
410+
keyMvName(tables.logsKeyRollup),
411+
]) {
412+
await client.command({ query: `DROP VIEW IF EXISTS ${db}.${mv}` });
413+
}
414+
}
415+
416+
/** (Re)create just the rollup materialized views for a scenario's tables. */
417+
export async function createRollupMaterializedViews(
418+
client: ClickHouseClient,
419+
tables: ScenarioTables,
420+
): Promise<void> {
421+
const db = EVAL_DATABASE;
422+
await client.command({
423+
query: `CREATE MATERIALIZED VIEW IF NOT EXISTS ${db}.${mvName(tables.tracesKvRollup)} TO ${db}.${tables.tracesKvRollup} AS ${tracesKvMvSelect(db, tables.traces)}`,
424+
});
425+
await client.command({
426+
query: `CREATE MATERIALIZED VIEW IF NOT EXISTS ${db}.${keyMvName(tables.tracesKeyRollup)} TO ${db}.${tables.tracesKeyRollup} AS ${keyRollupMvSelect(db, tables.tracesKvRollup)}`,
427+
});
428+
await client.command({
429+
query: `CREATE MATERIALIZED VIEW IF NOT EXISTS ${db}.${mvName(tables.logsKvRollup)} TO ${db}.${tables.logsKvRollup} AS ${logsKvMvSelect(db, tables.logs)}`,
430+
});
431+
await client.command({
432+
query: `CREATE MATERIALIZED VIEW IF NOT EXISTS ${db}.${keyMvName(tables.logsKeyRollup)} TO ${db}.${tables.logsKeyRollup} AS ${keyRollupMvSelect(db, tables.logsKvRollup)}`,
433+
});
434+
}
435+
436+
/**
437+
* One-shot bulk backfill of the rollup tables from already-loaded raw tables,
438+
* reusing the exact same SELECTs the materialized views run. This is the fast
439+
* alternative to letting the MVs fan out on a bulk Parquet insert: aggregating
440+
* the whole table once is far cheaper than maintaining the SummingMergeTree
441+
* incrementally per insert block. Produces byte-identical rollup content.
442+
*
443+
* Only backfills a raw table's rollups when that raw table has rows (so a
444+
* traces-only scenario doesn't write empty log rollups).
445+
*/
446+
export async function backfillRollups(
447+
client: ClickHouseClient,
448+
tables: ScenarioTables,
449+
): Promise<void> {
450+
const db = EVAL_DATABASE;
451+
452+
const hasRows = async (table: string): Promise<boolean> => {
453+
const rs = await client.query({
454+
query: `SELECT 1 FROM ${db}.${table} LIMIT 1`,
455+
format: 'JSONEachRow',
456+
});
457+
return (await rs.json<unknown>()).length > 0;
458+
};
459+
460+
if (await hasRows(tables.traces)) {
461+
await client.command({
462+
query: `INSERT INTO ${db}.${tables.tracesKvRollup} ${tracesKvMvSelect(db, tables.traces)}`,
463+
});
464+
await client.command({
465+
query: `INSERT INTO ${db}.${tables.tracesKeyRollup} ${keyRollupMvSelect(db, tables.tracesKvRollup)}`,
466+
});
467+
}
468+
if (await hasRows(tables.logs)) {
469+
await client.command({
470+
query: `INSERT INTO ${db}.${tables.logsKvRollup} ${logsKvMvSelect(db, tables.logs)}`,
471+
});
472+
await client.command({
473+
query: `INSERT INTO ${db}.${tables.logsKeyRollup} ${keyRollupMvSelect(db, tables.logsKvRollup)}`,
474+
});
475+
}
476+
}
477+
394478
async function dropRollupTables(
395479
client: ClickHouseClient,
396480
tables: ScenarioTables,

0 commit comments

Comments
 (0)