import fs from 'node:fs'; import os from 'node:os'; import path from 'node:path'; import { performance } from 'node:perf_hooks'; import { CondensedGraph } from '../src/core/CondensedGraph.js'; import { ShardedSnapshotBuilder } from '../src/core/shards/ShardedSnapshotBuilder.js'; import { ShardedSnapshot } from '../src/core/shards/ShardedSnapshot.js'; import { FileShardStorage } from '../src/core/shards/FileShardStorage.js'; import { DeltaShardBinary } from '../src/core/shards/DeltaShardBinary.js'; import { WaveletShardBinary } from '../src/core/shards/WaveletShardBinary.js'; function parseArgs(argv) { const args = new Map(); for (let i = 2; i < argv.length; i++) { const value = argv[i]; if (!value.startsWith('--')) continue; const [key, inline] = value.slice(2).split('='); if (inline !== undefined) { args.set(key, inline); continue; } const next = argv[i + 1]; if (next && !next.startsWith('--')) { args.set(key, next); i++; } else { args.set(key, true); } } return args; } function makeRng(seed = 1337) { let state = seed >>> 0; return () => { state = (state * 1664525 + 1013904223) >>> 0; return state / 0xffffffff; }; } function makeZipfSampler(count, skew, seed = 1337) { const weights = new Float64Array(count); let sum = 0; for (let i = 1; i <= count; i++) { const w = 1 / Math.pow(i, skew); weights[i - 1] = w; sum += w; } const cdf = new Float64Array(count); let acc = 0; for (let i = 0; i < count; i++) { acc += weights[i] / sum; cdf[i] = acc; } let state = seed >>> 0; const rand = () => { state = (state * 1664525 + 1013904223) >>> 0; return state / 0xffffffff; }; return () => { const r = rand(); let lo = 0; let hi = cdf.length - 1; while (lo < hi) { const mid = (lo + hi) >> 1; if (r <= cdf[mid]) { hi = mid; } else { lo = mid + 1; } } return lo; }; } function buildGraph({ edges, users, docs, seed }) { const graph = new CondensedGraph(); const rng = makeRng(seed); for (let i = 0; i < edges; i++) { const userId = Math.floor(rng() * users); const docId = Math.floor(rng() * docs); graph.addEdge(`user:${userId}`, 'owner', `doc:${docId}`); } graph.finalizePerfectHash(); graph.finalizeWaveletAdjacency({ dropAdjacencyList: true }); return graph; } function buildDeltaLayer(snapshot, relationId, deltaEdges, seed, dir, skew) { const rng = makeRng(seed); const nodeCount = snapshot.nodeCount; const sampler = skew ? makeZipfSampler(nodeCount, skew, seed + 11) : null; const buckets = new Map(); let addCount = 0; let removeCount = 0; for (let i = 0; i < deltaEdges; i++) { const srcId = sampler ? sampler() : Math.floor(rng() * nodeCount); const dstId = sampler ? sampler() : Math.floor(rng() * nodeCount); const shardMeta = snapshot._selectShardMeta(relationId, 'out', srcId); if (!shardMeta) continue; const localSource = snapshot._localSource(srcId, shardMeta); const key = shardMeta.cacheKey; let entry = buckets.get(key); if (!entry) { entry = { shardMeta, additions: [], removals: [] }; buckets.set(key, entry); } if (rng() < 0.5) { entry.removals.push({ srcLocal: localSource, otherId: dstId }); removeCount++; } else { entry.additions.push({ srcLocal: localSource, otherId: dstId, possBits: 65535, relBits: 65535 }); addCount++; } } fs.mkdirSync(dir, { recursive: true }); const shards = []; for (const entry of buckets.values()) { const shardKey = `delta-${entry.shardMeta.key}`; const buffer = DeltaShardBinary.serialize({ relationId: entry.shardMeta.relationId, direction: entry.shardMeta.direction, rangeStart: entry.shardMeta.rangeStart, rangeEnd: entry.shardMeta.rangeEnd, nodeCount: snapshot.nodeCount, additions: entry.additions, removals: entry.removals }); fs.writeFileSync(path.join(dir, shardKey), new Uint8Array(buffer)); shards.push({ key: shardKey, relationId: entry.shardMeta.relationId, direction: entry.shardMeta.direction, rangeStart: entry.shardMeta.rangeStart, rangeEnd: entry.shardMeta.rangeEnd, cacheKey: entry.shardMeta.cacheKey }); } return { shards, storage: new FileShardStorage(dir), addCount, removeCount }; } function buildQueries(snapshot, relationId, count, seed) { const rng = makeRng(seed); const queries = []; for (let i = 0; i < count; i++) { const srcId = Math.floor(rng() * snapshot.nodeCount); const edges = snapshot.getOutEdgesSync(srcId, relationId); if (edges.length) { const edge = edges[Math.floor(rng() * edges.length)]; queries.push([srcId, edge.dst]); } else { const dstId = Math.floor(rng() * snapshot.nodeCount); queries.push([srcId, dstId]); } } return queries; } function timeFindEdges(snapshot, relationId, queries) { const start = performance.now(); let found = 0; for (const [srcId, dstId] of queries) { if (snapshot.findEdgeSync(srcId, relationId, dstId)) found++; } const duration = performance.now() - start; return { duration, found }; } function compactLayer(snapshot, layer, outputDir) { fs.mkdirSync(outputDir, { recursive: true }); let compacted = 0; for (const shardMeta of layer.shards) { const base = snapshot._cacheIndex.get(shardMeta.cacheKey); if (!base) continue; const shard = snapshot._loadShardSync(base.relationId, base.direction, base.rangeStart); if (!shard) continue; const deltaBuffer = layer.storage.getSync(shardMeta.key); if (!deltaBuffer) continue; const deltaShard = DeltaShardBinary.deserialize(deltaBuffer); const rangeSize = shard.rangeEnd - shard.rangeStart; const sources = new Array(rangeSize); for (let localSource = 0; localSource < rangeSize; localSource++) { const range = snapshot._rangeForSource(shard, localSource); const list = []; if (range) { for (let pos = range.start; pos < range.end; pos++) { list.push({ otherId: shard.dstIds[pos], possBits: shard.possBits[pos], relBits: shard.relBits[pos] }); } } sources[localSource] = list; } for (const removal of deltaShard.removals) { const list = sources[removal.srcLocal]; if (!list) continue; const idx = list.findIndex((item) => item.otherId === removal.otherId); if (idx !== -1) list.splice(idx, 1); } for (const addition of deltaShard.additions) { const list = sources[addition.srcLocal] || (sources[addition.srcLocal] = []); list.push({ otherId: addition.otherId, possBits: addition.possBits, relBits: addition.relBits }); } const buffer = WaveletShardBinary.serialize({ relationId: shard.relationId, direction: shard.direction, rangeStart: shard.rangeStart, rangeEnd: shard.rangeEnd, nodeCount: snapshot.nodeCount, sources }); fs.writeFileSync(path.join(outputDir, base.key), new Uint8Array(buffer)); compacted++; } return compacted; } function sumDeltaBytes(dir) { let total = 0; const entries = fs.readdirSync(dir, { withFileTypes: true }); for (const entry of entries) { if (!entry.isFile()) continue; const stat = fs.statSync(path.join(dir, entry.name)); total += stat.size; } return total; } function sumShardBytes(dir, manifest) { let total = 0; if (manifest.nodeTableKey) { const nodeTablePath = path.join(dir, manifest.nodeTableKey); if (fs.existsSync(nodeTablePath)) total += fs.statSync(nodeTablePath).size; } if (manifest.componentKey) { const componentPath = path.join(dir, manifest.componentKey); if (fs.existsSync(componentPath)) total += fs.statSync(componentPath).size; } for (const shard of manifest.shards || []) { const shardPath = path.join(dir, shard.key); if (fs.existsSync(shardPath)) total += fs.statSync(shardPath).size; } return total; } function shardSizeStats(dir, manifest) { const sizes = []; for (const shard of manifest.shards || []) { const shardPath = path.join(dir, shard.key); if (fs.existsSync(shardPath)) sizes.push(fs.statSync(shardPath).size); } sizes.sort((a, b) => a - b); if (!sizes.length) return { p50: 0, p95: 0, p99: 0, max: 0, min: 0 }; const pct = (p) => sizes[Math.min(sizes.length - 1, Math.floor(p * sizes.length))]; return { p50: pct(0.5), p95: pct(0.95), p99: pct(0.99), max: sizes[sizes.length - 1], min: sizes[0] }; } const args = parseArgs(process.argv); const edges = Number(args.get('edges') || 200000); const users = Number(args.get('users') || 50000); const docs = Number(args.get('docs') || 100000); const bucketSize = Number(args.get('bucket') || 65536); const queryCount = Number(args.get('queries') || 20000); const seed = Number(args.get('seed') || 1337); const deltaPowers = String(args.get('delta-powers') || '0,1,2,3').split(',').map((v) => Number(v.trim())); const compactAt = Number(args.get('compact-at') || 0); const deltaSkew = Number(args.get('delta-skew') || 0); const shardMode = String(args.get('shard-mode') || 'chain'); const componentRelations = args.get('component-relations') ? String(args.get('component-relations')).split(',').map((v) => v.trim()).filter(Boolean) : ['owner']; console.log('Sharded delta overlay bench'); console.log(` edges: ${edges}`); console.log(` users: ${users}`); console.log(` docs: ${docs}`); console.log(` queries: ${queryCount}`); console.log(` delta powers: ${deltaPowers.join(',')}`); if (compactAt > 0) console.log(` compact at: ${compactAt}`); if (deltaSkew > 0) console.log(` delta skew: ${deltaSkew}`); console.log(` shard mode: ${shardMode}`); console.log(` component relations: ${componentRelations.join(',')}`); console.log(' note: timings are per-query averages; expect noise and cache effects. Compare trends, not single points.'); const graph = buildGraph({ edges, users, docs, seed }); const baseDir = fs.mkdtempSync(path.join(os.tmpdir(), 'sharded-delta-bench-')); let shardDir = path.join(baseDir, 'base'); const builder = new ShardedSnapshotBuilder({ bucketSize, includeDirections: ['out', 'in'], shardMode, componentRelations }); const manifest = builder.build(graph, shardDir); let storage = new FileShardStorage(shardDir); let snapshot = new ShardedSnapshot(manifest, storage, { cacheLimit: 64, recentLimit: 256 }); snapshot.initializeSync(); const relationId = snapshot.relationIdToName.indexOf('owner'); const queries = buildQueries(snapshot, relationId, queryCount, seed + 9); console.log('Delta size vs avg findEdge time'); console.log('edges | base_bytes | shard_files | shard_p50 | shard_p95 | shard_p99 | deltaEdges | avg_us_per_query | delta_bytes | compact_ms | compact_us_per_edge'); let baseEdges = edges; const baseline = timeFindEdges(snapshot, relationId, queries); const baseBytes = sumShardBytes(shardDir, manifest); const shardFiles = countShardFiles(manifest); const shardStats = shardSizeStats(shardDir, manifest); console.log(`${baseEdges} | ${baseBytes} | ${shardFiles} | ${shardStats.p50} | ${shardStats.p95} | ${shardStats.p99} | 0 | ${(baseline.duration / queryCount) * 1000} | 0 | - | -`); for (const power of deltaPowers) { const deltaEdges = Math.max(1, Math.floor(Math.pow(10, power))); const layerDir = path.join(baseDir, `delta-${power}`); const layer = buildDeltaLayer(snapshot, relationId, deltaEdges, seed + power + 1, layerDir, deltaSkew); snapshot.setDeltaLayers([layer]); snapshot.clearDeltaCache(); const result = timeFindEdges(snapshot, relationId, queries); const avgUs = (result.duration / queryCount) * 1000; const deltaBytes = sumDeltaBytes(layerDir); let baseBytesNow = baseBytes; let shardFilesNow = shardFiles; let shardStatsNow = shardStats; let compactMs = '-'; let compactUsPerEdge = '-'; if (compactAt > 0 && deltaEdges >= compactAt) { const compactDir = path.join(baseDir, `compact-${power}`); const start = performance.now(); compactLayer(snapshot, layer, compactDir); const totalMs = performance.now() - start; compactMs = totalMs.toFixed(2); const denom = Math.max(1, layer.addCount + layer.removeCount); compactUsPerEdge = ((totalMs * 1000) / denom).toFixed(3); baseEdges = Math.max(0, baseEdges + layer.addCount - layer.removeCount); const newBaseDir = path.join(baseDir, `base-${power}`); fs.rmSync(newBaseDir, { recursive: true, force: true }); fs.mkdirSync(newBaseDir, { recursive: true }); if (manifest.nodeTableKey) { fs.copyFileSync(path.join(shardDir, manifest.nodeTableKey), path.join(newBaseDir, manifest.nodeTableKey)); } if (manifest.componentKey) { fs.copyFileSync(path.join(shardDir, manifest.componentKey), path.join(newBaseDir, manifest.componentKey)); } for (const shard of manifest.shards || []) { const basePath = path.join(shardDir, shard.key); const compactPath = path.join(compactDir, shard.key); const sourcePath = fs.existsSync(compactPath) ? compactPath : basePath; fs.copyFileSync(sourcePath, path.join(newBaseDir, shard.key)); } shardDir = newBaseDir; storage = new FileShardStorage(shardDir); snapshot = new ShardedSnapshot(manifest, storage, { cacheLimit: 64, recentLimit: 256 }); snapshot.initializeSync(); baseBytesNow = sumShardBytes(shardDir, manifest); shardFilesNow = countShardFiles(manifest); shardStatsNow = shardSizeStats(shardDir, manifest); } console.log(`${baseEdges} | ${baseBytesNow} | ${shardFilesNow} | ${shardStatsNow.p50} | ${shardStatsNow.p95} | ${shardStatsNow.p99} | ${deltaEdges} | ${avgUs} | ${deltaBytes} | ${compactMs} | ${compactUsPerEdge}`); } fs.rmSync(baseDir, { recursive: true, force: true }); function countShardFiles(manifest) { let total = 0; if (manifest.nodeTableKey) total += 1; if (manifest.componentKey) total += 1; total += (manifest.shards || []).length; return total; }