diff --git a/docs-mintlify/reference/configuration/environment-variables.mdx b/docs-mintlify/reference/configuration/environment-variables.mdx index d5f60fa323368..b7b2053a04d0c 100644 --- a/docs-mintlify/reference/configuration/environment-variables.mdx +++ b/docs-mintlify/reference/configuration/environment-variables.mdx @@ -195,11 +195,8 @@ and we don't recommend enabling it in production yet. Enables the preview fast-track path for Cube Store query queues. When enabled, Cube can atomically enqueue and retrieve a query when queue concurrency is -available, reducing queue coordination round trips. It applies to queries at -queue priority 10 and above, which is where user-facing queries and the -pre-aggregation builds a request waits on are submitted; background refresh runs -below that and keeps using the regular path. This requires Cube Store -1.7.25 or newer; against an older version Cube keeps using the regular path. +available, reducing queue coordination round trips. This requires Cube and Cube Store +1.7.26 or newer; against an older version Cube keeps using the regular path. | Possible Values | Default in Development | Default in Production | | --------------- | ---------------------- | --------------------- | diff --git a/packages/cubejs-base-driver/src/queue-driver.interface.ts b/packages/cubejs-base-driver/src/queue-driver.interface.ts index 5980031cd59b2..0d4b2c7efd01a 100644 --- a/packages/cubejs-base-driver/src/queue-driver.interface.ts +++ b/packages/cubejs-base-driver/src/queue-driver.interface.ts @@ -9,25 +9,11 @@ export type QueryKeyHash = string & { __type: 'QueryKeyHash' }; export type QueryKeysTuple = [keyHash: QueryKeyHash, queueId: QueueId]; export type GetActiveAndToProcessResponse = [active: QueryKeysTuple[], toProcess: QueryKeysTuple[]]; export type QueryStageStateResponse = [active: string[], toProcess: string[]] | [active: string[], toProcess: string[], defs: Record]; -export type RetrieveForProcessingSuccess = [ - added: unknown, - // Identifies the retrieved generation of the queue item. - queueId: QueueId | null, +export type RetrieveForProcessingSuccess = { active: QueryKeyHash[], - pending: number, + queueSize: number, def: QueryDef, - retrieved: true -]; -export type RetrieveForProcessingFail = [ - added: unknown, - // Null when no queue item was retrieved. - queueId: QueueId | null, - active: QueryKeyHash[], - pending: number, - def: null, - retrieved: false -]; -export type RetrieveForProcessingResponse = RetrieveForProcessingSuccess | RetrieveForProcessingFail | null; +}; export type AddToQueueResponse = [ added: number, queueId: QueueId | null, @@ -106,7 +92,7 @@ export interface QueueDriverConnectionInterface { updateHeartBeat(hash: QueryKeyHash, queueId: QueueId | null): Promise; // Atomically moves a queue item to active. Returns null when another node is already // processing the query or the concurrency budget is full. - retrieveForProcessing(hash: QueryKeyHash, queueId: QueueId): Promise; + retrieveForProcessing(hash: QueryKeyHash, queueId: QueueId): Promise; optimisticQueryUpdate(hash: QueryKeyHash, toUpdate: unknown, queueId: QueueId): Promise; cancelQuery(queryKey: QueryKey, queueId: QueueId | null): Promise; getQueryAndRemove(hash: QueryKeyHash, queueId: QueueId | null): Promise<[QueryDef]>; diff --git a/packages/cubejs-cubestore-driver/src/CubeStoreQueueDriver.ts b/packages/cubejs-cubestore-driver/src/CubeStoreQueueDriver.ts index 95563eb392041..5067b71a82865 100644 --- a/packages/cubejs-cubestore-driver/src/CubeStoreQueueDriver.ts +++ b/packages/cubejs-cubestore-driver/src/CubeStoreQueueDriver.ts @@ -4,7 +4,6 @@ import { QueueDriverConnectionInterface, QueryStageStateResponse, QueryDef, - RetrieveForProcessingResponse, RetrieveForProcessingSuccess, QueueDriverOptions, AddToQueueQuery, @@ -357,30 +356,20 @@ export class CubestoreQueueDriverConnection implements QueueDriverConnectionInte return null; } - return [ - 1, - row.id ? parseInt(row.id, 10) : null, - this.decodeActiveKeysFromRow(row.active), - parseInt(row.pending, 10), - this.decodeQueryDefFromRow(row as { payload: string, extra?: string | null }, method), - true - ]; + return { + active: this.decodeActiveKeysFromRow(row.active), + queueSize: parseInt(row.pending, 10), + def: this.decodeQueryDefFromRow(row as { payload: string, extra?: string | null }, method), + }; } - public async retrieveForProcessing(hash: QueryKeyHash, _queueId: QueueId): Promise { - const rows = await this.driver.query('QUEUE RETRIEVE EXTENDED CONCURRENCY ? ?', [ + public async retrieveForProcessing(hash: QueryKeyHash, _queueId: QueueId): Promise { + const rows = await this.driver.query('QUEUE RETRIEVE CONCURRENCY ? ?', [ this.options.concurrency, this.prefixKey(hash), ]); - if (rows && rows.length) { - return this.decodeRetrievedFromRow(rows[0], 'retrieveForProcessing') || [ - 0, - null, - this.decodeActiveKeysFromRow(rows[0].active), - parseInt(rows[0].pending, 10), - null, - false - ]; + if (rows.length) { + return this.decodeRetrievedFromRow(rows[0], 'retrieveForProcessing'); } return null; diff --git a/packages/cubejs-query-orchestrator/DEVELOPMENT.md b/packages/cubejs-query-orchestrator/DEVELOPMENT.md index f51d405843f47..243a1aaafde3b 100644 --- a/packages/cubejs-query-orchestrator/DEVELOPMENT.md +++ b/packages/cubejs-query-orchestrator/DEVELOPMENT.md @@ -79,7 +79,8 @@ enum ResultStatus { `EXTENDED` on `QUEUE RETRIEVE` changes only the failure shape: without it a failed retrieval returns zero rows, with it a single row where `payload` and `id` are `NULL` but `pending` -and `active` are filled. The driver always sends `EXTENDED`. +and `active` are filled. The driver uses the non-extended form and treats zero rows as a failed +retrieval. ## Enqueue and wait: `executeInQueue` @@ -221,13 +222,13 @@ sequenceDiagram participant QueryOrchestrator QueryQueue->>QueueDriver: retrieveForProcessing - QueueDriver->>CubeStore: QUEUE RETRIEVE EXTENDED CONCURRENCY ?n ?path + QueueDriver->>CubeStore: QUEUE RETRIEVE CONCURRENCY ?n ?path CubeStore-->>QueueDriver: RetrieveResponse - QueueDriver-->>QueryQueue: [added, queueId, activeKeys, queueSize, def, retrieved] + QueueDriver-->>QueryQueue: { active, queueSize, def } | null Note over QueueDriver,CubeStore: The retrieval is atomic in Cube Store:
only one node moves the item to active - alt def && added && activeKeys includes our key && retrieved - QueryQueue-)Background: sendProcessMessageFn(RetrievedQuery) + alt retrieved + QueryQueue-)Background: sendProcessMessageFn(queryKeyHash, queueId, retrieved) Note over QueryQueue,Background: Detached from here on: the hand-off returns,
the execution keeps running Background->>QueueDriver: optimisticQueryUpdate @@ -356,7 +357,53 @@ returns items highest priority first (oldest first within a priority) and reconc The fast track selects itself, so it is priority blind with nothing to compensate. That is what the `active + pending < concurrency` condition rules out: retrieving leaves a free slot for every item already pending, so no item is jumped over, and once the budget gets tight -the fast track steps aside and lets reconcile pick by priority. Note that a retrieved item -goes straight to active and never becomes pending, so a burst onto an idle queue still -fast-tracks every query — items only start accumulating in pending once the concurrency -budget is exhausted, which is exactly when the condition should stop firing. +the fast track steps aside and lets reconcile pick by priority. A retrieved item goes straight +to active and never becomes pending, so a burst onto an idle queue fast-tracks exactly one +concurrency budget's worth and no more: items start accumulating in pending the moment the +budget is exhausted, which is when the condition stops firing. S3 measures this exactly — +50 retrievals out of 1000 at concurrency 50, 200 out of 1000 at concurrency 200 — so on a +burst the saving scales as `concurrency / burst size`, not with the size of the burst. + +## Benchmarks + +The harness lives in `test/benchmarks/` and is compiled by the ordinary `yarn tsc` — jest never +picks it up (`testMatch` is `*.test.ts`). Runs go through the suite runner: + +```bash +yarn tsc +yarn bench:suite --list # what the suites are +yarn bench:suite S1 --dry-run # the matrix with computed ρ and wall clock, run this first +yarn bench:suite S1 # off and on, one pass each +yarn bench:suite --report .context/bench-results/S1-….jsonl +``` + +Every run emits one `BENCH_RESULT {json}` line plus a `BENCH_TICK {json}` per second; the runner +collects both into `.context/bench-results/-.jsonl` and prints a markdown summary. +A single run can also be driven straight from env vars against +`dist/test/benchmarks/QueueCubestore.bench.js` (or `QueueMemory.bench.js`) — see `readSettings` +in `QueueBench.abstract.ts` for the full list. + +The one axis that decides everything is the load factor: + +``` +ρ = arrival_rate / capacity, capacity = concurrency / handler_latency +``` + +The fast track saves a round trip only while the concurrency budget has a free slot, so ρ is +what the suites sweep. `driverCalls.fastTrack.missRate` is the direct measure of the cost side: +an `ADD_AND_RETRIEVE` that comes back without a retrieval is a round trip spent for nothing. +Eligibility is read off the connection's `useFastTrack`, so a driver that cannot fast track +never registers an attempt. + +Two things about the numbers are artifacts of the harness rather than production behaviour: + +- Workers poll `reconcileQueue` on a timer (`BENCH_WORKER_RECONCILE_MS`, default 50ms) because a + worker never submits and so has no submit-time reconcile to bootstrap from. Production + reconcile is event-driven. This poll dominates `getQueriesToCancel` / `getActiveAndToProcess` + and is the entire traffic of the idle-floor suite, which is why S5 measures two intervals. +- Payload defaults are 5MB responses / 256KB query bodies, but the suites deliberately run at + 64KB / 16KB. S7 owns the payload axis. + +`driverCalls` is snapshotted after drain and before the idle tail, so on a suite with +`BENCH_IDLE_TAIL_MS` the tail's polling lands in `idle.driverCalls` and nowhere else — and +`main` plus `workers` add up to `total` exactly. diff --git a/packages/cubejs-query-orchestrator/package.json b/packages/cubejs-query-orchestrator/package.json index 223f668c8da16..35f95486d71ce 100644 --- a/packages/cubejs-query-orchestrator/package.json +++ b/packages/cubejs-query-orchestrator/package.json @@ -21,6 +21,7 @@ "unit": "jest --runInBand --forceExit --coverage --verbose test/unit", "integration": "jest --runInBand --verbose test/integration", "integration:cubestore": "jest --runInBand --verbose test/integration/cubestore", + "bench:suite": "node dist/test/benchmarks/run-suite.js", "lint": "eslint src/* test/* --ext .ts,.js", "lint:fix": "eslint --fix src/* test/* --ext .ts,.js" }, @@ -41,9 +42,11 @@ "@types/jest": "^29", "@types/node": "^22", "@types/ramda": "^0.27.32", + "@types/yargs": "^17.0.31", "jest": "^29", "ts-jest": "^29", - "typescript": "~5.2.2" + "typescript": "~5.2.2", + "yargs": "^17.7.1" }, "license": "Apache-2.0", "eslintConfig": { diff --git a/packages/cubejs-query-orchestrator/src/orchestrator/LocalQueueDriverConnection.ts b/packages/cubejs-query-orchestrator/src/orchestrator/LocalQueueDriverConnection.ts index 55d4211440364..9ac867ae02c78 100644 --- a/packages/cubejs-query-orchestrator/src/orchestrator/LocalQueueDriverConnection.ts +++ b/packages/cubejs-query-orchestrator/src/orchestrator/LocalQueueDriverConnection.ts @@ -11,7 +11,7 @@ import { QueryKeysTuple, GetActiveAndToProcessResponse, QueryStageStateResponse, - RetrieveForProcessingResponse, + RetrieveForProcessingSuccess, QueueDriverOptions, QueuePriority } from '@cubejs-backend/base-driver'; @@ -272,7 +272,7 @@ export class LocalQueueDriverConnection implements QueueDriverConnectionInterfac } } - public async retrieveForProcessing(queryKeyHash: QueryKeyHash, queueId: QueueId): Promise { + public async retrieveForProcessing(queryKeyHash: QueryKeyHash, queueId: QueueId): Promise { const query = this.state.queryDef[queryKeyHash]; const activeKeys = this.queueArray(this.state.active) as QueryKeyHash[]; @@ -283,14 +283,7 @@ export class LocalQueueDriverConnection implements QueueDriverConnectionInterfac this.state.active[queryKeyHash] || activeKeys.length >= this.concurrency ) { - return [ - 0, - null, - activeKeys, - Object.keys(this.state.toProcess).length, - null, - false - ]; + return null; } this.state.active[queryKeyHash] = { key: queryKeyHash, order: Number(queueId), queueId }; @@ -298,14 +291,11 @@ export class LocalQueueDriverConnection implements QueueDriverConnectionInterfac this.state.heartBeat[queryKeyHash] = { key: queryKeyHash, order: new Date().getTime(), queueId }; - return [ - 1, - query.queueId, - this.queueArray(this.state.active) as QueryKeyHash[], - Object.keys(this.state.toProcess).length, - query, - true - ]; + return { + active: this.queueArray(this.state.active) as QueryKeyHash[], + queueSize: Object.keys(this.state.toProcess).length, + def: query, + }; } public async optimisticQueryUpdate(queryKeyHash: QueryKeyHash, toUpdate: any, queueId: QueueId): Promise { diff --git a/packages/cubejs-query-orchestrator/src/orchestrator/QueryQueue.ts b/packages/cubejs-query-orchestrator/src/orchestrator/QueryQueue.ts index d36558735f70c..8ad5b2f4f81d1 100644 --- a/packages/cubejs-query-orchestrator/src/orchestrator/QueryQueue.ts +++ b/packages/cubejs-query-orchestrator/src/orchestrator/QueryQueue.ts @@ -25,14 +25,11 @@ export type QueryHandlerFn = (query: QueryDef, cancelHandler: CancelHandlerFn) = export type StreamHandlerFn = (query: QueryDef, stream: QueryStream) => Promise; export type QueryHandlersMap = Record; -export type RetrievedQuery = { - queryKeyHash: QueryKeyHash; - queueId: QueueId; - queueSize: number; - query: QueryDef; -}; - -export type SendProcessMessageFn = (retrieved: RetrievedQuery) => Promise | void; +export type SendProcessMessageFn = ( + queryKeyHash: QueryKeyHash, + queueId: QueueId, + retrieved: RetrieveForProcessingSuccess +) => Promise | void; export type SendCancelMessageFn = (query: QueryDef, queueId: QueueId | null) => Promise | void; export type ExecuteInQueueOptions = Omit & { @@ -131,7 +128,9 @@ export class QueryQueue { this.orphanedTimeout = options.orphanedTimeout || 120; this.heartBeatInterval = options.heartBeatInterval || 30; - this.sendProcessMessageFn = options.sendProcessMessageFn || ((retrieved) => { this.executeQuery(retrieved); }); + this.sendProcessMessageFn = options.sendProcessMessageFn || ( + (queryKeyHash, queueId, retrieved) => { this.executeQuery(queryKeyHash, queueId, retrieved); } + ); this.sendCancelMessageFn = options.sendCancelMessageFn || ((query, queueId) => { this.processCancel(query, queueId); }); this.queryHandlers = options.queryHandlers; this.streamHandler = options.streamHandler; @@ -307,13 +306,13 @@ export class QueryQueue { if (retrieved) { // The item is active already, there is nothing for reconcile to pick up - await this.dispatchQuery(this.retrievedQuery(queryKeyHash, queueId, retrieved)); + await this.dispatchQuery(queryKeyHash, queueId, retrieved); } else { await this.reconcileQueue(); } if (!added) { - const queryDef = retrieved ? retrieved[4] : await queueConnection.getQueryDef(queryKeyHash, queueId); + const queryDef = retrieved ? retrieved.def : await queueConnection.getQueryDef(queryKeyHash, queueId); if (queryDef) { waitingContext = { queueId, @@ -328,7 +327,7 @@ export class QueryQueue { // A retrieval carries the active keys of its prefix, and a retrieved query is never pending, // so it has no place in the queue to report - const [active, toProcess] = retrieved ? [retrieved[2], undefined] : await queueConnection.getQueryStageState(true); + const [active, toProcess] = retrieved ? [retrieved.active, undefined] : await queueConnection.getQueryStageState(true); this.logger('Waiting for query', { ...waitingContext, @@ -365,17 +364,6 @@ export class QueryQueue { } } - protected retrievedQuery(queryKeyHash: QueryKeyHash, queueId: QueueId, retrieved: RetrieveForProcessingSuccess): RetrievedQuery { - const [, , , queueSize, query] = retrieved; - - return { - queryKeyHash, - queueId, - queueSize, - query, - }; - } - /** * `dispose` releases the listener and the timer, it's a no-op once the promise resolved. */ @@ -796,17 +784,21 @@ export class QueryQueue { return; } - await this.dispatchQuery(retrieved); + await this.dispatchQuery(queryKeyHashed, queueId, retrieved); } - protected async dispatchQuery(retrieved: RetrievedQuery): Promise { + protected async dispatchQuery( + queryKeyHash: QueryKeyHash, + queueId: QueueId, + retrieved: RetrieveForProcessingSuccess + ): Promise { try { - await this.sendProcessMessageFn(retrieved); + await this.sendProcessMessageFn(queryKeyHash, queueId, retrieved); } catch (e: any) { this.logger('Error while processing message', { - queueId: retrieved.queueId, - queryKey: retrieved.query.queryKey, - requestId: retrieved.query.requestId, + queueId, + queryKey: retrieved.def.queryKey, + requestId: retrieved.def.requestId, error: (e.stack || e).toString(), queuePrefix: this.redisQueuePrefix }); @@ -817,27 +809,16 @@ export class QueryQueue { * Atomically moves the query specified by `queryKeyHashed` to the active set. Returns `null` * when another node is already running the query or the concurrency budget is full. */ - protected async retrieveQueryForProcessing(queryKeyHashed: QueryKeyHash, queueId: QueueId): Promise { + protected async retrieveQueryForProcessing(queryKeyHashed: QueryKeyHash, queueId: QueueId): Promise { const queueConnection = await this.queueDriver.createConnection(); - let insertedCount; - let activeKeys; - let queueSize; let query; - let retrievalSucceeded; try { - const retrieveResult = await queueConnection.retrieveForProcessing(queryKeyHashed, queueId); - if (retrieveResult) { - [insertedCount, , activeKeys, queueSize, query, retrievalSucceeded] = retrieveResult; - } - - const activated = activeKeys && activeKeys.indexOf(queryKeyHashed) !== -1; - if (!query) { + const retrieved = await queueConnection.retrieveForProcessing(queryKeyHashed, queueId); + if (!retrieved) { query = await queueConnection.getQueryDef(queryKeyHashed, null); - } - if (!query || !insertedCount || !activated || !retrievalSucceeded) { // TODO Ideally streaming queries should reconcile queue here after waiting on open slot however in practice continue wait timeout reconciles faster CPU-wise // if (query?.queryHandler === 'stream') { // const [active] = await queueConnection.getQueryStageState(true); @@ -852,22 +833,13 @@ export class QueryQueue { queryKey: query && query.queryKey || queryKeyHashed, requestId: query && query.requestId, queuePrefix: this.redisQueuePrefix, - retrievalSucceeded, query, - insertedCount, - activeKeys, - activated, queryExists: !!query }); return null; } - return { - queryKeyHash: queryKeyHashed, - queueId, - queueSize, - query, - }; + return retrieved; } catch (e: any) { this.logger('Queue storage error', { queueId, @@ -888,8 +860,12 @@ export class QueryQueue { * the result. It's the counterpart of `sendProcessMessageFn` and the entry point for a custom * implementation which hands the query over to another process. */ - public async executeQuery(retrieved: RetrievedQuery): Promise { - const { queryKeyHash: queryKeyHashed, queueId, queueSize, query } = retrieved; + public async executeQuery( + queryKeyHashed: QueryKeyHash, + queueId: QueueId, + retrieved: RetrieveForProcessingSuccess + ): Promise { + const { queueSize, def: query } = retrieved; const queueConnection = await this.queueDriver.createConnection(); diff --git a/packages/cubejs-query-orchestrator/test/benchmarks/QueueBench.abstract.ts b/packages/cubejs-query-orchestrator/test/benchmarks/QueueBench.abstract.ts index f558043c5e217..80d41018d7353 100644 --- a/packages/cubejs-query-orchestrator/test/benchmarks/QueueBench.abstract.ts +++ b/packages/cubejs-query-orchestrator/test/benchmarks/QueueBench.abstract.ts @@ -1,10 +1,21 @@ -import { CubeStoreQueueDriver } from '@cubejs-backend/cubestore-driver'; import crypto from 'crypto'; import path from 'path'; import { ChildProcess, fork } from 'child_process'; -import { createPromiseLock, MethodName, pausePromise } from '@cubejs-backend/shared'; -import { QueueDriverConnectionInterface, QueueDriverInterface, QueuePriority } from '@cubejs-backend/base-driver'; -import { LocalQueueDriver, QueryQueue, QueryQueueOptions } from '../../src'; +import { createPromiseLock, pausePromise } from '@cubejs-backend/shared'; +import { QueuePriority } from '@cubejs-backend/base-driver'; +import { ContinueWaitError, QueryQueueOptions, TimeoutError } from '../../src'; +import { + BenchCounters, + cloneMethods, + createBenchQueue, + createCounters, + driverCallsTotal, + MethodCounter, + mergeEvents, + mergeMethods, + percentiles, +} from './instrument'; +import { counterSnapshot, ParentMessage, WorkerSnapshot } from './protocol'; export type QueryQueueTestOptions = Pick & { beforeAll?: () => Promise, @@ -12,56 +23,161 @@ export type QueryQueueTestOptions = Pick>(methodName: M): any { - return async (...args: Parameters) => { - if (!(methodName in counters.methods)) { - counters.methods[methodName] = { - started: 1, - finished: 0, - }; - } else { - counters.methods[methodName].started++; - } +type PriorityBucket = { priority: number, weight: number }; + +type BenchSettings = { + driver: string, + fastTrack: boolean, + workers: number, + concurrency: number, + totalQueries: number, + periodMs: number, + pushIntervalMs: number, + priority: QueuePriority, + priorityMix: PriorityBucket[] | null, + handlerLatencyMs: number, + queueResponseSize: number, + queuePayloadSize: number, + workerReconcileMs: number, + warmupQueries: number, + idleTailMs: number, + tickMs: number, +}; - const result = await (connection[methodName] as any)(...args); - counters.methods[methodName].finished++; +type Phase = 'warmup' | 'measure' | 'drain' | 'idle'; - return result; - }; - } +type Aggregate = { + methods: Record, + events: Record, + handlersStarted: number, + handlersFinished: number, + fastTrack: { attempts: number, hits: number }, + driverCalls: number, +}; +const EMPTY_AGGREGATE = (): Aggregate => ({ + methods: {}, + events: {}, + handlersStarted: 0, + handlersFinished: 0, + fastTrack: { attempts: 0, hits: 0 }, + driverCalls: 0, +}); + +function toAggregate(source: Pick): Aggregate { return { - ...connection, - addToQueue: wrapAsyncMethod('addToQueue'), - getResult: wrapAsyncMethod('getResult'), - getQueriesToCancel: wrapAsyncMethod('getQueriesToCancel'), - getActiveAndToProcess: wrapAsyncMethod('getActiveAndToProcess'), - retrieveForProcessing: wrapAsyncMethod('retrieveForProcessing'), - getQueryDef: wrapAsyncMethod('getQueryDef'), - setResultAndRemoveQuery: wrapAsyncMethod('setResultAndRemoveQuery'), - getQueryStageState: wrapAsyncMethod('getQueryStageState'), - getResultBlocking: wrapAsyncMethod('getResultBlocking'), - optimisticQueryUpdate: wrapAsyncMethod('optimisticQueryUpdate'), - getQueryAndRemove: wrapAsyncMethod('getQueryAndRemove'), - release: connection.release, + methods: cloneMethods(source.methods), + events: { ...source.events }, + handlersStarted: source.handlersStarted, + handlersFinished: source.handlersFinished, + fastTrack: { ...source.fastTrack }, + driverCalls: driverCallsTotal(source), }; } -function patchQueueDriverForTrack(driver: QueueDriverInterface, counters: any): QueueDriverInterface { - return { - ...driver, - createConnection: async () => { - counters.connections++; +function addAggregate(into: Aggregate, from: Aggregate): Aggregate { + mergeMethods(into.methods, from.methods); + mergeEvents(into.events, from.events); + into.handlersStarted += from.handlersStarted; + into.handlersFinished += from.handlersFinished; + into.fastTrack.attempts += from.fastTrack.attempts; + into.fastTrack.hits += from.fastTrack.hits; + into.driverCalls += from.driverCalls; - return patchQueueDriverConnectionForTrack(await driver.createConnection(), counters); - }, - redisHash: (...args) => driver.redisHash(...args), - release: async (...args) => { - counters.connections--; + return into; +} - return driver.release(...args); +/** + * Warmup is charged against a baseline instead of a counter reset, so the workers need no reset + * round trip and the subtraction is identical on every source + */ +function subAggregate(a: Aggregate, b: Aggregate): Aggregate { + const methods: Record = {}; + for (const [name, m] of Object.entries(a.methods)) { + const base = b.methods[name] || { started: 0, finished: 0 }; + methods[name] = { started: m.started - base.started, finished: m.finished - base.finished }; + } + + const events: Record = {}; + for (const [name, count] of Object.entries(a.events)) { + events[name] = count - (b.events[name] || 0); + } + + return { + methods, + events, + handlersStarted: a.handlersStarted - b.handlersStarted, + handlersFinished: a.handlersFinished - b.handlersFinished, + fastTrack: { + attempts: a.fastTrack.attempts - b.fastTrack.attempts, + hits: a.fastTrack.hits - b.fastTrack.hits, }, + driverCalls: a.driverCalls - b.driverCalls, + }; +} + +function parsePriorityMix(raw: string | undefined): PriorityBucket[] | null { + if (!raw) { + return null; + } + + const buckets = raw.split(',').map((part) => { + const [priority, weight] = part.split(':'); + return { priority: parseInt(priority, 10), weight: parseInt(weight, 10) }; + }); + + if (buckets.some((b) => Number.isNaN(b.priority) || Number.isNaN(b.weight) || b.weight <= 0)) { + throw new Error(`Malformed BENCH_PRIORITY_MIX: ${raw}, expected "10:50,0:50"`); + } + + return buckets; +} + +/** + * Largest-remainder apportionment, so an uneven mix still interleaves instead of arriving in blocks + */ +function pickBucket(buckets: PriorityBucket[], assigned: number[], index: number): number { + const total = buckets.reduce((acc, b) => acc + b.weight, 0); + let best = 0; + let bestScore = -Infinity; + + for (let i = 0; i < buckets.length; i++) { + const score = (buckets[i].weight * (index + 1)) / total - assigned[i]; + if (score > bestScore) { + bestScore = score; + best = i; + } + } + + return best; +} + +const envInt = (name: string, fallback: number) => parseInt(process.env[name] || `${fallback}`, 10); + +function readSettings(driver: string, workers: number): BenchSettings { + const totalQueries = envInt('BENCH_TOTAL_QUERIES', 1000); + // BENCH_PERIOD_MS spreads the queries evenly over that window instead of pushing them + // as fast as the loop allows, which is what decides whether the queue ever backlogs + const periodMs = envInt('BENCH_PERIOD_MS', 0); + + return { + driver, + fastTrack: process.env.CUBEJS_QUEUE_FAST_TRACK === 'true', + workers, + concurrency: envInt('BENCH_CONCURRENCY', 50), + totalQueries, + periodMs, + pushIntervalMs: periodMs > 0 && totalQueries > 0 ? Math.max(1, Math.round(periodMs / totalQueries)) : 10, + priority: envInt('BENCH_PRIORITY', QueuePriority.Interactive), + priorityMix: parsePriorityMix(process.env.BENCH_PRIORITY_MIX), + handlerLatencyMs: envInt('BENCH_HANDLER_LATENCY_MS', 1500), + // eslint-disable-next-line no-bitwise + queueResponseSize: envInt('BENCH_RESPONSE_SIZE', 5 << 20), + queuePayloadSize: envInt('BENCH_PAYLOAD_SIZE', 256 * 1024), + workerReconcileMs: envInt('BENCH_WORKER_RECONCILE_MS', 50), + warmupQueries: envInt('BENCH_WARMUP_QUERIES', 0), + idleTailMs: envInt('BENCH_IDLE_TAIL_MS', 0), + tickMs: envInt('BENCH_TICK_MS', 1000), }; } @@ -71,88 +187,23 @@ export function QueryQueueBenchmark(name: string, options: QueryQueueTestOptions await options.beforeAll(); } - const createBenchmark = async (benchSettings: { totalQueries: number, queueResponseSize: number, queuePayloadSize: number, currency: number, pushIntervalMs: number, priority: QueuePriority }) => { - const counters = { - connections: 0, - methods: {}, - events: {}, - queueStarted: 0, - queueResolved: 0, - handlersStarted: 0, - handlersFinished: 0, - queueDriverQueriesStarted: 0, - }; - - const queueDriverFactory = (driverType, queueDriverOptions) => { - switch (driverType) { - case 'memory': - return patchQueueDriverForTrack( - new LocalQueueDriver( - queueDriverOptions - ) as any, - counters - ); - case 'cubestore': - return patchQueueDriverForTrack( - new CubeStoreQueueDriver( - async () => options.cubeStoreDriverFactory(), - queueDriverOptions - ), - counters - ); - default: - throw new Error(`Unsupported driver: ${driverType}`); - } - }; + const createBenchmark = async (benchSettings: BenchSettings) => { + const counters = createCounters(); const tenantPrefix = crypto.randomBytes(6).toString('hex'); - const queue = new QueryQueue(`${tenantPrefix}#test_query_queue`, { - queryHandlers: { - query: async (_query) => { - counters.handlersStarted++; - await pausePromise(1500); - counters.handlersFinished++; - - return { - payload: 'a'.repeat(benchSettings.queueResponseSize), - }; - }, - stream: async (_query, _stream) => { - throw new Error('streaming handler is not supported for testing'); - } - }, - cancelHandlers: { - query: async (_query) => { - console.error('Cancel handler was called for query'); - }, - }, - continueWaitTimeout: 60 * 2, - executionTimeout: 20, - orphanedTimeout: 60 * 5, - concurrency: benchSettings.currency, - logger: (event, _params) => { - // console.log(event, _params); - // console.log(event); - - if (event in counters.events) { - counters.events[event]++; - } else { - counters.events[event] = 1; - } - - if (event.includes('error')) { - console.log(event, _params); - } - }, - queueDriverFactory, - ...options - }); - - // Spawn worker processes for multi-process simulation (CubeStore only) - type WorkerCounters = { handlersStarted: number; handlersFinished: number; events: Record }; - type WorkerState = { worker: ChildProcess; counters: WorkerCounters; prevFinished: number }; + const queue = createBenchQueue(`${tenantPrefix}#test_query_queue`, counters, benchSettings, options); + + type WorkerState = { + worker: ChildProcess, + latest: WorkerSnapshot, + baseline: Aggregate, + awaiting: { seq: number, resolve: () => void } | null, + alive: boolean, + }; const workerStates: WorkerState[] = []; const numWorkers = options.workers || 0; + const shutdown = { requested: false }; + const diedEarly = { count: 0 }; if (numWorkers > 0) { const workerPath = path.resolve(__dirname, 'QueueBenchWorker.js'); @@ -165,13 +216,20 @@ export function QueryQueueBenchmark(name: string, options: QueryQueueTestOptions const state: WorkerState = { worker: w, - counters: { handlersStarted: 0, handlersFinished: 0, events: {} }, - prevFinished: 0, + latest: counterSnapshot(createCounters()), + baseline: EMPTY_AGGREGATE(), + awaiting: null, + alive: true, }; - w.on('message', (msg: { type: string; data?: WorkerCounters }) => { - if (msg.type === 'counters' && msg.data) { - state.counters = msg.data; + w.on('message', (msg: ParentMessage) => { + if (msg.type === 'counters') { + state.latest = msg.data; + + if (state.awaiting && (state.awaiting.seq === msg.seq || msg.seq === -1)) { + state.awaiting.resolve(); + state.awaiting = null; + } } }); @@ -179,14 +237,30 @@ export function QueryQueueBenchmark(name: string, options: QueryQueueTestOptions console.error(`[Worker ${i}] error:`, err); }); + // Without this a dead worker keeps its last snapshot and every tick waits the full + // timeout for a reply that cannot come, while the run reports as if it were still there + w.on('exit', (code, signal) => { + if (!shutdown.requested) { + // Counted into the result too: on stderr alone this reads as a healthy run in + // the .jsonl, and a degraded point would get compared against a whole one + diedEarly.count++; + console.error(`[Worker ${i}] exited early with ${signal || `code ${code}`} — its counters stop here`); + } + + state.alive = false; + state.awaiting?.resolve(); + state.awaiting = null; + }); + w.send({ type: 'start', tenantPrefix, benchSettings: { queueResponseSize: benchSettings.queueResponseSize, - currency: benchSettings.currency, + concurrency: benchSettings.concurrency, + handlerLatencyMs: benchSettings.handlerLatencyMs, }, - reconcileInterval: 50, + reconcileIntervalMs: benchSettings.workerReconcileMs, }); workerStates.push(state); @@ -195,140 +269,388 @@ export function QueryQueueBenchmark(name: string, options: QueryQueueTestOptions console.log(`Spawned ${numWorkers} worker processes`); } - const processingPromisses = []; + let tickSeq = 0; - async function awaitProcessing() { - // process query can call reconcileQueue - while (await queue.shutdown() || processingPromisses.length) { - console.log('awaitProcessing', { - counters, - processingPromisses: processingPromisses.length + async function collectWorkerSnapshots(timeoutMs = 200): Promise { + if (workerStates.length === 0) { + return 0; + } + + const seq = ++tickSeq; + const requestedAt = Date.now(); + + const waits = workerStates.filter((ws) => ws.alive).map((ws) => new Promise((resolve) => { + if (ws.awaiting) { + ws.awaiting.resolve(); + } + + if (!ws.worker.connected) { + ws.alive = false; + resolve(); + + return; + } + + ws.awaiting = { seq, resolve }; + // Never reject: this promise loses the race below whenever a worker is slow, and a + // rejection settled after the loser is dropped would surface as an unhandled rejection + ws.worker.send({ type: 'tickRequest', seq }, (err) => { + if (err) { + ws.awaiting = null; + resolve(); + } }); - await Promise.all(processingPromisses.splice(0)); + })); + + if (waits.length === 0) { + return 0; } - // Shutdown worker processes - if (workerStates.length > 0) { - await Promise.all(workerStates.map((ws) => new Promise((resolve) => { - const onMessage = (msg: { type: string; data?: WorkerCounters }) => { - if (msg.type === 'counters' && msg.data) { - ws.counters = msg.data; - } - if (msg.type === 'done') { - ws.worker.removeListener('message', onMessage); - resolve(); - } - }; + await Promise.race([Promise.all(waits), pausePromise(timeoutMs)]); - ws.worker.on('message', onMessage); - ws.worker.send({ type: 'shutdown' }); - }))); + return Date.now() - requestedAt; + } + + function snapshotAggregate(): Aggregate { + const total = toAggregate(counters); + for (const ws of workerStates) { + addAggregate(total, toAggregate(ws.latest)); } + + return total; } - const progressIntervalId = setInterval(() => { - console.log('running', { - ...counters, - processingPromisses: processingPromisses.length, - benchSettings, - ...(workerStates.length > 0 ? { - workers: workerStates.map((ws, i) => { - const finishedSinceLastTick = ws.counters.handlersFinished - ws.prevFinished; - ws.prevFinished = ws.counters.handlersFinished; - return { - id: i, - handlersStarted: ws.counters.handlersStarted, - handlersFinished: ws.counters.handlersFinished, - processing: ws.counters.handlersStarted - ws.counters.handlersFinished, - processedFromLastEvent: finishedSinceLastTick, - }; - }), - } : {}), - }); - }, 1000); - - const lock = createPromiseLock(); - - const pusherIntervalId = setInterval(async () => { - if (counters.queueStarted >= benchSettings.totalQueries) { - lock.resolve(); - clearInterval(pusherIntervalId); + const runStartedAt = Date.now(); + let phase: Phase = benchSettings.warmupQueries > 0 ? 'warmup' : 'measure'; + let baseline = EMPTY_AGGREGATE(); + let baselineMain = EMPTY_AGGREGATE(); - return; + let pushed = 0; + let completed = 0; + const failed = { continueWait: 0, timeout: 0, other: 0 }; + const errorSamples: string[] = []; + let latenciesByPriority: Record = {}; + const latencies = () => Object.values(latenciesByPriority).flat(); + + const inFlight = () => pushed - completed - failed.continueWait - failed.timeout - failed.other; + + const processingPromisses: Promise[] = []; + + const bucketAssigned = benchSettings.priorityMix ? benchSettings.priorityMix.map(() => 0) : []; + + function priorityFor(index: number): QueuePriority { + if (!benchSettings.priorityMix) { + return benchSettings.priority; } - counters.queueStarted++; + const bucket = pickBucket(benchSettings.priorityMix, bucketAssigned, index); + bucketAssigned[bucket]++; + + return benchSettings.priorityMix[bucket].priority; + } + + function pushOne(index: number) { + pushed++; + const priority = priorityFor(index); const queueId = crypto.randomBytes(12).toString('hex'); + const startedAt = process.hrtime.bigint(); + const running = (async () => { try { await queue.executeInQueue('query', queueId, { - // eslint-disable-next-line no-bitwise payload: { large_str: 'a'.repeat(benchSettings.queuePayloadSize) }, orphanedTimeout: 120 - }, benchSettings.priority, { + }, priority, { stageQueryKey: 1, requestId: 'request-id', spanId: 'span-id' }); - } catch (e) { - console.error(e); - } - counters.queueResolved++; + completed++; + + const latencyMs = Number(process.hrtime.bigint() - startedAt) / 1e6; + (latenciesByPriority[priority] ||= []).push(latencyMs); + } catch (e: any) { + if (e instanceof ContinueWaitError) { + failed.continueWait++; + } else if (e instanceof TimeoutError) { + failed.timeout++; + } else { + failed.other++; + } - // losing memory for a result + if (errorSamples.length < 3) { + errorSamples.push(`${e?.constructor?.name}: ${e?.message}`); + } + } + + // The result is dropped rather than returned, so 1000 payloads are not held alive + // by the promise array until the run ends return null; })(); processingPromisses.push(running); - await running; - }, benchSettings.pushIntervalMs); - - await lock.promise; - await awaitProcessing(); - clearInterval(progressIntervalId); - - const workerAgg = workerStates.reduce( - (acc, ws) => ({ - handlersStarted: acc.handlersStarted + ws.counters.handlersStarted, - handlersFinished: acc.handlersFinished + ws.counters.handlersFinished, - }), - { handlersStarted: 0, handlersFinished: 0 } - ); - - console.dir({ - message: 'Result', - benchSettings, - ...counters, - ...(workerStates.length > 0 ? { - workers: workerStates.map((ws, i) => ({ - id: i, - ...ws.counters, - processing: ws.counters.handlersStarted - ws.counters.handlersFinished, - })), - totalHandlersStarted: counters.handlersStarted + workerAgg.handlersStarted, - totalHandlersFinished: counters.handlersFinished + workerAgg.handlersFinished, - } : {}), - }, { depth: null }); + } + + function runPusher(count: number, intervalMs: number): Promise { + if (count <= 0) { + return Promise.resolve(); + } + + const lock = createPromiseLock(); + let index = 0; + + const pusherIntervalId = setInterval(() => { + if (index >= count) { + clearInterval(pusherIntervalId); + lock.resolve(); + + return; + } + + pushOne(index); + index++; + }, intervalMs); + + return lock.promise as Promise; + } + + async function drain() { + // process query can call reconcileQueue + while (await queue.shutdown() || processingPromisses.length) { + await Promise.all(processingPromisses.splice(0)); + } + } + + let prevDriverCalls = 0; + let prevHandlersFinished = 0; + let prevTickAt = runStartedAt; + let peakDriverCallsPerSec = 0; + let ticking = false; + + const tickIntervalId = setInterval(async () => { + if (ticking) { + return; + } + ticking = true; + + try { + const workerSnapshotAgeMs = await collectWorkerSnapshots(); + const now = Date.now(); + const agg = subAggregate(snapshotAggregate(), baseline); + + const driverCallsDelta = agg.driverCalls - prevDriverCalls; + const elapsedSinceTick = Math.max(1, now - prevTickAt); + const perSec = (driverCallsDelta * 1000) / elapsedSinceTick; + + if (phase === 'measure' || phase === 'drain') { + peakDriverCallsPerSec = Math.max(peakDriverCallsPerSec, perSec); + } + + console.log(`BENCH_TICK ${JSON.stringify({ + runId: process.env.BENCH_RUN_ID || null, + tMs: now - runStartedAt, + phase, + pushed, + inFlight: inFlight(), + completed, + failed: failed.continueWait + failed.timeout + failed.other, + driverCallsTotal: agg.driverCalls, + driverCallsDelta, + driverCallsPerSec: Math.round(perSec), + handlersFinished: agg.handlersFinished, + handlersFinishedDelta: agg.handlersFinished - prevHandlersFinished, + fastTrackAttempts: agg.fastTrack.attempts, + fastTrackHits: agg.fastTrack.hits, + workerSnapshotAgeMs, + })}`); + + prevDriverCalls = agg.driverCalls; + prevHandlersFinished = agg.handlersFinished; + prevTickAt = now; + } finally { + ticking = false; + } + }, benchSettings.tickMs); + + if (benchSettings.warmupQueries > 0) { + await runPusher(benchSettings.warmupQueries, Math.min(benchSettings.pushIntervalMs, 50)); + await drain(); + + await collectWorkerSnapshots(); + baseline = snapshotAggregate(); + baselineMain = toAggregate(counters); + for (const ws of workerStates) { + ws.baseline = toAggregate(ws.latest); + } + + pushed = 0; + completed = 0; + failed.continueWait = 0; + failed.timeout = 0; + failed.other = 0; + latenciesByPriority = {}; + errorSamples.splice(0); + prevDriverCalls = 0; + prevHandlersFinished = 0; + } + + phase = 'measure'; + const measureStartedAt = Date.now(); + + await runPusher(benchSettings.totalQueries, benchSettings.pushIntervalMs); + const pushEndedAt = Date.now(); + + phase = 'drain'; + await drain(); + const drainEndedAt = Date.now(); + + // Everything the run reports as its cost comes from this one snapshot, taken after drain + // and before the idle tail: worker polling during the tail is never charged to the + // queries, and because the three views come from a single instant the per-process + // breakdown adds up to the total exactly. Without a fresh pull the idle delta would + // absorb up to a tick of worker polling that predates it. + await collectWorkerSnapshots(); + const measured = subAggregate(snapshotAggregate(), baseline); + const measuredMain = subAggregate(toAggregate(counters), baselineMain); + const measuredWorkers = workerStates.map((ws) => subAggregate(toAggregate(ws.latest), ws.baseline)); + + if (benchSettings.idleTailMs > 0) { + phase = 'idle'; + await pausePromise(benchSettings.idleTailMs); + } + + clearInterval(tickIntervalId); + await collectWorkerSnapshots(); + + const idleDriverCalls = benchSettings.idleTailMs > 0 + ? subAggregate(snapshotAggregate(), baseline).driverCalls - measured.driverCalls + : 0; + + shutdown.requested = true; + + if (workerStates.length > 0) { + await Promise.all(workerStates.map((ws) => new Promise((resolve) => { + if (!ws.alive || !ws.worker.connected) { + resolve(); + + return; + } + + const onMessage = (msg: ParentMessage) => { + if (msg.type === 'counters') { + ws.latest = msg.data; + } + if (msg.type === 'done') { + ws.worker.removeListener('message', onMessage); + resolve(); + } + }; + + ws.worker.on('message', onMessage); + // A worker that dies before answering would otherwise hold this promise open forever + ws.worker.once('exit', resolve); + ws.worker.send({ type: 'shutdown' }, (err) => { + if (err) { + resolve(); + } + }); + }))); + } + + const pushWindowMs = pushEndedAt - measureStartedAt; + const capacityQps = (benchSettings.concurrency * 1000) / benchSettings.handlerLatencyMs; + const targetRateQps = benchSettings.periodMs > 0 + ? (benchSettings.totalQueries * 1000) / benchSettings.periodMs + : null; + const actualRateQps = pushWindowMs > 0 ? (pushed * 1000) / pushWindowMs : null; + const processes = benchSettings.workers + 1; + + const round = (v: number | null, digits = 3) => (v === null ? null : Number(v.toFixed(digits))); + + const result = { + runId: process.env.BENCH_RUN_ID || null, + suite: process.env.BENCH_SUITE || null, + label: process.env.BENCH_LABEL || null, + axis: process.env.BENCH_AXIS ? JSON.parse(process.env.BENCH_AXIS) : null, + settings: benchSettings, + derived: { + capacityQps: round(capacityQps), + targetRateQps: round(targetRateQps), + actualRateQps: round(actualRateQps), + targetRho: round(targetRateQps === null ? null : targetRateQps / capacityQps), + actualRho: round(actualRateQps === null ? null : actualRateQps / capacityQps), + }, + timing: { + startedAt: new Date(runStartedAt).toISOString(), + measureStartedAtMs: measureStartedAt - runStartedAt, + pushWindowMs, + drainMs: drainEndedAt - pushEndedAt, + elapsedMs: drainEndedAt - measureStartedAt, + idleTailMs: benchSettings.idleTailMs, + }, + outcome: { + pushed, + completed, + inFlightAtEnd: inFlight(), + failed: { ...failed, total: failed.continueWait + failed.timeout + failed.other }, + errorSamples, + workersDiedEarly: diedEarly.count, + }, + latencyMs: percentiles(latencies()), + latencyMsByPriority: Object.fromEntries( + Object.entries(latenciesByPriority).map(([priority, samples]) => [priority, percentiles(samples)]) + ), + driverCalls: { + total: measured.driverCalls, + perQuery: completed > 0 ? round(measured.driverCalls / completed) : null, + peakPerSec: Math.round(peakDriverCallsPerSec), + byMethod: measured.methods, + main: measuredMain.methods, + workers: measuredWorkers.map((agg) => agg.methods), + fastTrack: { + ...measured.fastTrack, + missRate: measured.fastTrack.attempts > 0 + ? round(1 - measured.fastTrack.hits / measured.fastTrack.attempts) + : null, + }, + }, + events: { + merged: measured.events, + main: measuredMain.events, + workers: measuredWorkers.map((agg) => agg.events), + }, + handlers: { + started: measured.handlersStarted, + finished: measured.handlersFinished, + main: measuredMain.handlersFinished, + workers: measuredWorkers.map((agg) => agg.handlersFinished), + }, + idle: { + driverCalls: idleDriverCalls, + callsPerSecPerProcess: benchSettings.idleTailMs > 0 + ? round((idleDriverCalls * 1000) / benchSettings.idleTailMs / processes) + : null, + }, + connections: counters.connections, + }; + + // stdout is a pipe under the suite runner, and a pipe write is asynchronous on POSIX — + // the process.exit below would be free to drop a half-written line. console.log offers + // no completion callback, so this one line goes out through write(). + await new Promise((resolve) => { + process.stdout.write(`BENCH_RESULT ${JSON.stringify(result)}\n`, () => resolve()); + }); + + if (!process.env.BENCH_RUN_ID) { + console.dir({ message: 'Result', ...result }, { depth: null }); + } }; - const totalQueries = parseInt(process.env.BENCH_TOTAL_QUERIES || '1000', 10); - // BENCH_PERIOD_MS spreads the queries evenly over that window instead of pushing them - // as fast as the loop allows, which is what decides whether the queue ever backlogs - const periodMs = parseInt(process.env.BENCH_PERIOD_MS || '0', 10); - - await createBenchmark({ - currency: parseInt(process.env.BENCH_CONCURRENCY || '50', 10), - totalQueries, - pushIntervalMs: periodMs > 0 ? Math.max(1, Math.round(periodMs / totalQueries)) : 10, - priority: parseInt(process.env.BENCH_PRIORITY || `${QueuePriority.Interactive}`, 10), - // eslint-disable-next-line no-bitwise - queueResponseSize: parseInt(process.env.BENCH_RESPONSE_SIZE || `${5 << 20}`, 10), - queuePayloadSize: parseInt(process.env.BENCH_PAYLOAD_SIZE || `${256 * 1024}`, 10), - }); + await createBenchmark(readSettings(name, options.workers || 0)); if (options.afterAll) { await options.afterAll(); diff --git a/packages/cubejs-query-orchestrator/test/benchmarks/QueueBenchWorker.ts b/packages/cubejs-query-orchestrator/test/benchmarks/QueueBenchWorker.ts index 624c1e48a2e1d..a5f4b01c860d5 100644 --- a/packages/cubejs-query-orchestrator/test/benchmarks/QueueBenchWorker.ts +++ b/packages/cubejs-query-orchestrator/test/benchmarks/QueueBenchWorker.ts @@ -2,94 +2,68 @@ import 'source-map-support/register'; import { CubeStoreDriver } from '@cubejs-backend/cubestore-driver'; -import { pausePromise } from '@cubejs-backend/shared'; import { QueryQueue } from '../../src'; +import { countEvent, createBenchQueue, createCounters } from './instrument'; +import { counterSnapshot, WorkerMessage, WorkerStartMessage } from './protocol'; if (!process.send) { throw new Error('QueueBenchWorker must be run as a child process with IPC'); } -const counters = { - handlersStarted: 0, - handlersFinished: 0, - events: {} as Record, -}; +// A killed run must not leave a worker polling Cube Store forever +process.on('disconnect', () => process.exit(0)); + +const RECONCILE_ERROR_EVENT = 'bench reconcile poll error'; + +const counters = createCounters(); let cubeStoreDriver: CubeStoreDriver; let queue: QueryQueue; let reconcileId: ReturnType; -let progressId: ReturnType; -process.on('message', async (msg: { type: string; tenantPrefix?: string; benchSettings?: { queueResponseSize: number; currency: number; handlerLatencyMs?: number }; reconcileInterval?: number }) => { +process.on('message', async (msg: WorkerMessage) => { if (msg.type === 'start') { - const { tenantPrefix, benchSettings, reconcileInterval } = msg as Required; + const { tenantPrefix, benchSettings, reconcileIntervalMs } = msg as WorkerStartMessage; cubeStoreDriver = new CubeStoreDriver({}); - queue = new QueryQueue(`${tenantPrefix}#test_query_queue`, { - queryHandlers: { - query: async () => { - counters.handlersStarted++; - await pausePromise(benchSettings.handlerLatencyMs || 1500); - counters.handlersFinished++; - - return { - payload: 'a'.repeat(benchSettings.queueResponseSize), - }; - }, - stream: async () => { - throw new Error('streaming handler is not supported for testing'); - } + queue = createBenchQueue( + `${tenantPrefix}#test_query_queue`, + counters, + benchSettings, + { + cacheAndQueueDriver: 'cubestore', + cubeStoreDriverFactory: async () => cubeStoreDriver, }, - cancelHandlers: { - query: async () => { - console.error('[Worker] Cancel handler was called for query'); - }, - }, - continueWaitTimeout: 60 * 2, - executionTimeout: 20, - orphanedTimeout: 60 * 5, - concurrency: benchSettings.currency, - cacheAndQueueDriver: 'cubestore', - cubeStoreDriverFactory: async () => cubeStoreDriver, - logger: (event, _params) => { - if (event in counters.events) { - counters.events[event]++; - } else { - counters.events[event] = 1; - } + '[Worker] ', + ); - if (event.includes('error')) { - console.log('[Worker]', event, _params); - } - }, - }); - - // Periodically reconcile to pick up pending queries from CubeStore + // A worker never submits, so it has no submit-time reconcile to bootstrap from. This poll is + // an artifact of the harness — production reconcile is event-driven — which is why its + // interval is a setting and lands in the reported run settings. reconcileId = setInterval(() => { - queue.reconcileQueue(); - }, reconcileInterval); - - // Report counters to main process periodically - progressId = setInterval(() => { - process.send!({ - type: 'counters', - data: { ...counters, events: { ...counters.events } }, + // reconcileQueue rethrows, and an unhandled rejection takes the process down under + // Node's default policy — the run would then keep reporting as if the worker were here + queue.reconcileQueue().catch((e) => { + countEvent(counters, RECONCILE_ERROR_EVENT); + console.error('[Worker]', RECONCILE_ERROR_EVENT, e); }); - }, 1000); + }, reconcileIntervalMs); + } + + // Answering on request instead of pushing on a timer keeps the worker numbers inside the tick + // they belong to + if (msg.type === 'tickRequest') { + process.send!({ type: 'counters', seq: msg.seq, data: counterSnapshot(counters) }); } if (msg.type === 'shutdown') { clearInterval(reconcileId); - clearInterval(progressId); await queue.shutdown(); await cubeStoreDriver.release(); - process.send!({ - type: 'counters', - data: { ...counters, events: { ...counters.events } }, - }); + process.send!({ type: 'counters', seq: -1, data: counterSnapshot(counters) }); process.send!({ type: 'done' }); process.disconnect(); process.exit(0); diff --git a/packages/cubejs-query-orchestrator/test/benchmarks/QueueCubestore.bench.ts b/packages/cubejs-query-orchestrator/test/benchmarks/QueueCubestore.bench.ts index 441a6b8aaa718..24bfd0d0871ac 100644 --- a/packages/cubejs-query-orchestrator/test/benchmarks/QueueCubestore.bench.ts +++ b/packages/cubejs-query-orchestrator/test/benchmarks/QueueCubestore.bench.ts @@ -28,7 +28,7 @@ const beforeAll = async () => { const workers = parseInt(process.env.WORKERS || '2', 10); QueryQueueBenchmark( - `CubeStore Queue (workers: ${workers})`, + 'cubestore', { cacheAndQueueDriver: 'cubestore', cubeStoreDriverFactory, diff --git a/packages/cubejs-query-orchestrator/test/benchmarks/QueueMemory.bench.ts b/packages/cubejs-query-orchestrator/test/benchmarks/QueueMemory.bench.ts index 59f3546264b60..5b0b5b06c459f 100644 --- a/packages/cubejs-query-orchestrator/test/benchmarks/QueueMemory.bench.ts +++ b/packages/cubejs-query-orchestrator/test/benchmarks/QueueMemory.bench.ts @@ -12,7 +12,7 @@ const beforeAll = async () => { }; QueryQueueBenchmark( - 'Memory Queue', + 'memory', { cacheAndQueueDriver: 'memory', beforeAll, diff --git a/packages/cubejs-query-orchestrator/test/benchmarks/instrument.ts b/packages/cubejs-query-orchestrator/test/benchmarks/instrument.ts new file mode 100644 index 0000000000000..036909be0fe15 --- /dev/null +++ b/packages/cubejs-query-orchestrator/test/benchmarks/instrument.ts @@ -0,0 +1,263 @@ +import { CubeStoreQueueDriver } from '@cubejs-backend/cubestore-driver'; +import { MethodName, pausePromise } from '@cubejs-backend/shared'; +import { + AddToQueueResponse, + QueueDriverConnectionInterface, + QueueDriverInterface, + QueuePriority, +} from '@cubejs-backend/base-driver'; +import { LocalQueueDriver, QueryQueue, QueryQueueOptions } from '../../src'; + +export type MethodCounter = { started: number, finished: number }; + +export type BenchCounters = { + connections: number, + methods: Record, + events: Record, + handlersStarted: number, + handlersFinished: number, + fastTrack: { attempts: number, hits: number }, +}; + +export function createCounters(): BenchCounters { + return { + connections: 0, + methods: {}, + events: {}, + handlersStarted: 0, + handlersFinished: 0, + fastTrack: { attempts: 0, hits: 0 }, + }; +} + +export function countEvent(counters: Pick, event: string) { + counters.events[event] = (counters.events[event] || 0) + 1; +} + +export function driverCallsTotal(counters: Pick): number { + return Object.values(counters.methods).reduce((acc, m) => acc + m.started, 0); +} + +export function mergeMethods(into: Record, from: Record) { + for (const [name, m] of Object.entries(from)) { + if (name in into) { + into[name].started += m.started; + into[name].finished += m.finished; + } else { + into[name] = { started: m.started, finished: m.finished }; + } + } + + return into; +} + +export function mergeEvents(into: Record, from: Record) { + for (const [name, count] of Object.entries(from)) { + into[name] = (into[name] || 0) + count; + } + + return into; +} + +export function cloneMethods(methods: Record): Record { + return mergeMethods({}, methods); +} + +export type Percentiles = { count: number, mean: number, p50: number, p90: number, p95: number, p99: number, max: number }; + +export function percentiles(samples: number[]): Percentiles { + if (samples.length === 0) { + return { count: 0, mean: 0, p50: 0, p90: 0, p95: 0, p99: 0, max: 0 }; + } + + const sorted = [...samples].sort((a, b) => a - b); + const at = (q: number) => Math.round(sorted[Math.min(sorted.length - 1, Math.max(0, Math.ceil(q * sorted.length) - 1))]); + + return { + count: sorted.length, + mean: Math.round(sorted.reduce((a, b) => a + b, 0) / sorted.length), + p50: at(0.5), + p90: at(0.9), + p95: at(0.95), + p99: at(0.99), + max: Math.round(sorted[sorted.length - 1]), + }; +} + +/** + * Asking the connection itself, rather than reading the env, covers the version capability check + * and the priority floor too, so a driver that cannot fast track never registers an attempt and + * the miss rate stays a property of the queue rather than of the deployment. + */ +async function fastTrackEligible(connection: QueueDriverConnectionInterface, priority: QueuePriority): Promise { + const { useFastTrack } = connection as any; + + return typeof useFastTrack === 'function' ? useFastTrack.call(connection, priority) : false; +} + +/** `addToQueue` is wrapped separately, so that it can also record the fast track outcome */ +const TRACKED_METHODS: MethodName[] = [ + 'getResult', + 'getQueriesToCancel', + 'getActiveAndToProcess', + 'retrieveForProcessing', + 'getQueryDef', + 'setResultAndRemoveQuery', + 'getQueryStageState', + 'getResultBlocking', + 'optimisticQueryUpdate', + 'getQueryAndRemove', +]; + +function patchQueueDriverConnectionForTrack(connection: QueueDriverConnectionInterface, counters: BenchCounters): QueueDriverConnectionInterface { + function wrapAsyncMethod>(methodName: M): any { + return async (...args: Parameters) => { + if (!(methodName in counters.methods)) { + counters.methods[methodName] = { + started: 1, + finished: 0, + }; + } else { + counters.methods[methodName].started++; + } + + const result = await (connection[methodName] as any)(...args); + counters.methods[methodName].finished++; + + return result; + }; + } + + const trackedAddToQueue = wrapAsyncMethod('addToQueue'); + + const tracked: Record = { + ...Object.fromEntries(TRACKED_METHODS.map((methodName) => [methodName, wrapAsyncMethod(methodName)])), + addToQueue: async (...args: Parameters) => { + const eligible = await fastTrackEligible(connection, args[3]); + if (eligible) { + counters.fastTrack.attempts++; + } + + const result: AddToQueueResponse = await trackedAddToQueue(...args); + if (eligible && result[4]) { + counters.fastTrack.hits++; + } + + return result; + }, + }; + + // Spreading the connection would copy own properties only and silently drop every method that + // lives on the prototype and is not re-listed above — `updateHeartBeat` among them, which + // QueryQueue calls on a timer during a long handler + return new Proxy(connection, { + get: (target, prop) => { + if (typeof prop === 'string' && prop in tracked) { + return tracked[prop]; + } + + const value = Reflect.get(target, prop, target); + + return typeof value === 'function' ? value.bind(target) : value; + }, + }); +} + +export function patchQueueDriverForTrack(driver: QueueDriverInterface, counters: BenchCounters): QueueDriverInterface { + return { + ...driver, + createConnection: async () => { + counters.connections++; + + return patchQueueDriverConnectionForTrack(await driver.createConnection(), counters); + }, + redisHash: (...args) => driver.redisHash(...args), + release: async (...args) => { + counters.connections--; + + return driver.release(...args); + }, + }; +} + +/** + * Both the main process and the workers count through the same wrapper, otherwise every run with + * `workers > 0` reports the driver calls of the main process only + */ +export function makeQueueDriverFactory(counters: BenchCounters, cubeStoreDriverFactory?: () => Promise) { + return (driverType: string, queueDriverOptions: any) => { + switch (driverType) { + case 'memory': + return patchQueueDriverForTrack( + new LocalQueueDriver(queueDriverOptions) as any, + counters + ); + case 'cubestore': + return patchQueueDriverForTrack( + new CubeStoreQueueDriver( + async () => cubeStoreDriverFactory(), + queueDriverOptions + ), + counters + ); + default: + throw new Error(`Unsupported driver: ${driverType}`); + } + }; +} + +export type BenchQueueSettings = { + queueResponseSize: number, + concurrency: number, + handlerLatencyMs: number, +}; + +/** + * The submitter and the workers have to run the same queue: any difference between the two + * would show up in the results as a difference between processes + */ +export function createBenchQueue( + queueName: string, + counters: BenchCounters, + settings: BenchQueueSettings, + options: Pick, + logPrefix = '', +): QueryQueue { + return new QueryQueue(queueName, { + queryHandlers: { + query: async () => { + counters.handlersStarted++; + await pausePromise(settings.handlerLatencyMs); + counters.handlersFinished++; + + return { + payload: 'a'.repeat(settings.queueResponseSize), + }; + }, + stream: async () => { + throw new Error('streaming handler is not supported for testing'); + }, + }, + cancelHandlers: { + query: async () => { + console.error(`${logPrefix}Cancel handler was called for query`); + }, + }, + continueWaitTimeout: 60 * 2, + executionTimeout: 20, + orphanedTimeout: 60 * 5, + concurrency: settings.concurrency, + logger: (event, params) => { + countEvent(counters, event); + + // stderr, not stdout: stdout carries the BENCH_RESULT/BENCH_TICK lines the suite + // runner parses, and it is shared with every worker + if (event.includes('error')) { + console.error(`${logPrefix}${event}`, params); + } + }, + queueDriverFactory: makeQueueDriverFactory(counters, options.cubeStoreDriverFactory), + cacheAndQueueDriver: options.cacheAndQueueDriver, + cubeStoreDriverFactory: options.cubeStoreDriverFactory, + }); +} diff --git a/packages/cubejs-query-orchestrator/test/benchmarks/protocol.ts b/packages/cubejs-query-orchestrator/test/benchmarks/protocol.ts new file mode 100644 index 0000000000000..c35ce80ba8438 --- /dev/null +++ b/packages/cubejs-query-orchestrator/test/benchmarks/protocol.ts @@ -0,0 +1,36 @@ +import { BenchCounters, BenchQueueSettings, cloneMethods, MethodCounter } from './instrument'; + +export type WorkerStartMessage = { + type: 'start', + tenantPrefix: string, + benchSettings: BenchQueueSettings, + reconcileIntervalMs: number, +}; + +export type WorkerMessage = + | WorkerStartMessage + | { type: 'tickRequest', seq: number } + | { type: 'shutdown' }; + +export type WorkerSnapshot = { + handlersStarted: number, + handlersFinished: number, + methods: Record, + events: Record, + fastTrack: { attempts: number, hits: number }, +}; + +export type ParentMessage = + | { type: 'counters', seq: number, data: WorkerSnapshot } + | { type: 'done' }; + +/** Detached from the live counters, so a snapshot in flight over IPC cannot keep moving */ +export function counterSnapshot(counters: Omit): WorkerSnapshot { + return { + handlersStarted: counters.handlersStarted, + handlersFinished: counters.handlersFinished, + methods: cloneMethods(counters.methods), + events: { ...counters.events }, + fastTrack: { ...counters.fastTrack }, + }; +} diff --git a/packages/cubejs-query-orchestrator/test/benchmarks/run-suite.ts b/packages/cubejs-query-orchestrator/test/benchmarks/run-suite.ts new file mode 100644 index 0000000000000..d4d25c9fac5a0 --- /dev/null +++ b/packages/cubejs-query-orchestrator/test/benchmarks/run-suite.ts @@ -0,0 +1,487 @@ +// eslint-disable-next-line import/no-extraneous-dependencies +import 'source-map-support/register'; + +import fs from 'fs'; +import path from 'path'; +import readline from 'readline'; +import { fork } from 'child_process'; +// eslint-disable-next-line import/no-extraneous-dependencies +import yargs from 'yargs'; +// eslint-disable-next-line import/no-extraneous-dependencies +import { hideBin } from 'yargs/helpers'; +import { pausePromise } from '@cubejs-backend/shared'; +import { driverCallsTotal } from './instrument'; +import { BenchRun, estimateRunMs, rhoOf, Suite, SUITES, suiteByName } from './suites'; + +type Args = { + suites: string[], + driver?: 'cubestore' | 'memory', + fastTrack: boolean[], + settleMs: number, + runTimeoutMs: number, + only?: string[], + repeat: number, + out?: string, + dryRun: boolean, + report?: string, + list: boolean, +}; + +const commaList = (v: unknown): string[] => String(v).split(',').map((s) => s.trim()).filter(Boolean); + +const FAST_TRACK_PASSES: Record = { on: true, true: true, off: false, false: false }; + +/** + * Every other option is validated by yargs — `choices`, `check`, `strict`. Mapping an + * unrecognised token to `off` would run a pass the caller did not ask for and label it as if + * they had. + */ +function parseFastTrackPasses(v: unknown): boolean[] { + const passes = commaList(v).map((token) => { + const pass = FAST_TRACK_PASSES[token.toLowerCase()]; + if (pass === undefined) { + throw new Error(`--fast-track: expected on/off, got "${token}"`); + } + + return pass; + }); + + if (passes.length === 0) { + throw new Error('--fast-track needs at least one pass, e.g. --fast-track=off,on'); + } + + return passes; +} + +function parseArgs(argv: string[]): Args { + const parsed = yargs(argv) + .scriptName('bench:suite') + // A variadic positional is only collected by a command builder, and angle brackets would make + // it required, which --list and --report do not satisfy + .command('$0 [suites..]', 'Run queue benchmark suites, off and on, into a .jsonl', (y) => y + .positional('suites', { + describe: 'Suite names, or "all" for every suite except the smoke one', + type: 'string', + array: true, + default: [] as string[], + })) + .example('$0 S1 --dry-run', 'price the sweep before starting it') + .example('$0 S1 S6 --settle=5000', 'run two suites back to back') + .example('$0 S1 --only=rho=2.5 --repeat=3', 'one point, three times') + .example('$0 --report results.jsonl', 'redraw the table from a finished run') + .option('driver', { + describe: 'Override the driver the suite declares', + choices: ['cubestore', 'memory'] as const, + }) + .option('fast-track', { + describe: 'Which passes to run for each configuration', + default: 'off,on', + coerce: parseFastTrackPasses, + }) + .option('settle', { + describe: 'Pause between runs, ms — lets the previous run\'s connections go away', + type: 'number', + default: 3000, + }) + .option('run-timeout', { + describe: 'Kill a run after this long, ms', + type: 'number', + default: 30 * 60 * 1000, + }) + .option('only', { + describe: 'Comma separated substrings; keeps just the matching runs', + coerce: commaList, + }) + // Above capacity a single pair is not enough to conclude from — the same point, several times + .option('repeat', { + describe: 'Run each selected point this many times, labelled #1, #2, …', + type: 'number', + default: 1, + }) + .option('out', { + describe: 'Write the .jsonl here instead of .context/bench-results', + type: 'string', + }) + .option('dry-run', { + describe: 'Print the matrix with computed \u03c1 and estimated wall clock, run nothing', + type: 'boolean', + default: false, + }) + .option('report', { + describe: 'Redraw the summary from an existing .jsonl and exit', + type: 'string', + }) + .option('list', { + describe: 'List the suites and exit', + type: 'boolean', + default: false, + }) + .check((a) => { + if (!a.list && !a.report && (a.suites as string[]).length === 0) { + throw new Error('Name at least one suite, or pass --list / --report'); + } + if (a.repeat < 1) { + throw new Error('--repeat must be at least 1'); + } + + return true; + }) + .strict() + .wrap(Math.min(120, process.stdout.columns || 120)) + .parseSync(); + + return { + suites: parsed.suites as string[], + driver: parsed.driver, + fastTrack: parsed.fastTrack, + settleMs: parsed.settle, + runTimeoutMs: parsed.runTimeout, + only: parsed.only, + repeat: parsed.repeat, + out: parsed.out, + dryRun: parsed.dryRun, + report: parsed.report, + list: parsed.list, + }; +} + +// dist/test/benchmarks -> dist -> package -> packages -> repo root +const repoRoot = path.resolve(__dirname, '../../../../..'); + +function resultsDir(): string { + const dir = path.resolve(repoRoot, '.context/bench-results'); + fs.mkdirSync(dir, { recursive: true }); + + return dir; +} + +// Seconds included: the sink appends, so a minute-granularity name silently merges two sweeps +// of the same suites into one file and the summary then keeps only the last pair per label +function stamp(): string { + return new Date().toISOString().replace(/[:.]/g, '-').slice(0, 19); +} + +type RunRecord = any; + +/** + * A throw inside the readline listener escapes the run loop entirely — no sink.end(), no + * summary, and every remaining point in the sweep skipped. A corrupt line is reachable: the + * bench child shares its stdout pipe with every worker it forks, and a pipe only guarantees + * atomic writes up to PIPE_BUF. + */ +function parseBenchLine(line: string, prefix: string, onError: (message: string) => void): any | null { + try { + return JSON.parse(line.slice(prefix.length)); + } catch (e: any) { + onError(`unparseable ${prefix.trim()} line (${e.message}): ${line.slice(0, 200)}`); + + return null; + } +} + +function selectRuns(suite: Suite, args: Args): BenchRun[] { + return args.only ? suite.runs.filter((r) => args.only!.some((o) => r.label.includes(o))) : suite.runs; +} + +async function executeRun(suite: Suite, benchRun: BenchRun, fastTrack: boolean, args: Args, sink: fs.WriteStream): Promise { + const driver = args.driver || suite.driver; + const entry = path.resolve(__dirname, driver === 'memory' ? 'QueueMemory.bench.js' : 'QueueCubestore.bench.js'); + const runId = `${suite.name}/${benchRun.label}/${fastTrack ? 'on' : 'off'}`; + + console.log(`\n=== ${runId} — ${JSON.stringify(benchRun.axis)} — est. ${Math.round(estimateRunMs(benchRun.env) / 1000)}s ===`); + + const child = fork(entry, [], { + execArgv: process.execArgv, + stdio: ['inherit', 'pipe', 'inherit', 'ipc'], + env: { + ...process.env, + ...benchRun.env, + CUBEJS_QUEUE_FAST_TRACK: `${fastTrack}`, + BENCH_RUN_ID: runId, + BENCH_SUITE: suite.name, + BENCH_LABEL: benchRun.label, + BENCH_AXIS: JSON.stringify(benchRun.axis), + }, + }); + + let result: RunRecord = null; + let malformed = 0; + const onMalformed = (message: string) => { + malformed++; + console.error(` !! ${runId}: ${message}`); + }; + + const rl = readline.createInterface({ input: child.stdout! }); + rl.on('line', (line) => { + if (line.startsWith('BENCH_RESULT ')) { + const parsed = parseBenchLine(line, 'BENCH_RESULT ', onMalformed); + if (parsed) { + result = parsed; + sink.write(`${JSON.stringify({ type: 'run', ...result })}\n`); + } + } else if (line.startsWith('BENCH_TICK ')) { + const parsed = parseBenchLine(line, 'BENCH_TICK ', onMalformed); + if (parsed) { + sink.write(`${JSON.stringify({ type: 'tick', ...parsed })}\n`); + } + } else { + console.log(` | ${line}`); + } + }); + + const timeoutMs = Math.max(args.runTimeoutMs, estimateRunMs(benchRun.env) * 2 + 120000); + let timedOut = false; + const timer = setTimeout(() => { + timedOut = true; + child.kill('SIGKILL'); + }, timeoutMs); + + // The child exits right after printing, so its last lines can still be unread at that point + const drained = new Promise((resolve) => rl.on('close', resolve)); + const code = await new Promise((resolve) => child.on('exit', resolve)); + await drained; + clearTimeout(timer); + + if (!result) { + let error: string; + if (timedOut) { + error = `timed out after ${timeoutMs}ms`; + } else if (malformed > 0) { + error = `exited with code ${code}, and ${malformed} line(s) were unparseable`; + } else { + error = `exited with code ${code} without a BENCH_RESULT`; + } + console.error(` !! ${runId}: ${error}`); + const failure = { runId, suite: suite.name, label: benchRun.label, axis: benchRun.axis, settings: { fastTrack }, error }; + sink.write(`${JSON.stringify({ type: 'run', ...failure })}\n`); + + return failure; + } + + return result; +} + +const fmt = (v: any, digits = 2) => (typeof v === 'number' ? Number(v.toFixed(digits)) : '—'); + +function markdown(header: string[], rows: (string | number)[][]): string { + return [ + `| ${header.join(' | ')} |`, + `|${header.map(() => '---').join('|')}|`, + ...rows.map((row) => `| ${row.join(' | ')} |`), + ].join('\n'); +} + +/** + * The workers poll reconcile on a timer that production does not have, and at a low ρ that poll is + * most of the traffic — it dilutes the headline badly. The submitting process is the honest view. + */ +function mainCallsPerQuery(record: RunRecord): number | null { + const methods = record?.driverCalls?.main; + const completed = record?.outcome?.completed; + if (!methods || !completed) { + return null; + } + + return driverCallsTotal({ methods }) / completed; +} + +/** + * Where queries drop, per-completed is an efficiency reading and per-pushed is the cost one — the + * two diverge sharply and quoting only the first turns a flat cost into an apparent saving + */ +function callsPerPushed(record: RunRecord): number | null { + const total = record?.driverCalls?.total; + const pushed = record?.outcome?.pushed; + + return total && pushed ? total / pushed : null; +} + +function pct(off: number | null | undefined, on: number | null | undefined): string { + if (!off || on === null || on === undefined) { + return '—'; + } + + const delta = ((on - off) / off) * 100; + + return `${delta >= 0 ? '+' : ''}${delta.toFixed(1)}%`; +} + +function markdownTable(records: RunRecord[]): string { + const byLabel = new Map(); + for (const r of records) { + const key = `${r.suite}/${r.label}`; + const pair = byLabel.get(key) || {}; + if (r.settings?.fastTrack) { + pair.on = r; + } else { + pair.off = r; + } + byLabel.set(key, pair); + } + + const header = ['run', 'ρ', 'rate q/s', 'done off→on', 'fail off→on', 'calls/pushed off→on', 'main calls/q off→on', 'Δ main', 'peak calls/s off→on', 'p95 ms off→on', 'elapsed s off→on', 'FT hit%', 'lost']; + const rows: (string | number)[][] = []; + + for (const [key, { off, on }] of byLabel) { + const sample = on || off; + const missRate = on?.driverCalls?.fastTrack?.missRate; + const mainOff = mainCallsPerQuery(off); + const mainOn = mainCallsPerQuery(on); + // A run that lost a worker measured fewer processes than its axis claims + const degraded = (off?.outcome?.workersDiedEarly ?? 0) > 0 || (on?.outcome?.workersDiedEarly ?? 0) > 0; + + if (off?.error || on?.error) { + rows.push([key, `off: ${off?.error || 'ok'}, on: ${on?.error || 'ok'}`, ...header.slice(2).map(() => '—')]); + } else if (sample) { + rows.push([ + degraded ? `⚠ ${key}` : key, + fmt(sample.derived?.actualRho), + fmt(sample.derived?.actualRateQps), + `${off?.outcome?.completed ?? '—'}→${on?.outcome?.completed ?? '—'}`, + `${off?.outcome?.failed?.total ?? '—'}→${on?.outcome?.failed?.total ?? '—'}`, + `${fmt(callsPerPushed(off))}→${fmt(callsPerPushed(on))}`, + `${fmt(mainOff)}→${fmt(mainOn)}`, + pct(mainOff, mainOn), + `${off?.driverCalls?.peakPerSec ?? '—'}→${on?.driverCalls?.peakPerSec ?? '—'}`, + `${fmt(off?.latencyMs?.p95, 0)}→${fmt(on?.latencyMs?.p95, 0)}`, + `${fmt((off?.timing?.elapsedMs ?? 0) / 1000, 1)}→${fmt((on?.timing?.elapsedMs ?? 0) / 1000, 1)}`, + typeof missRate === 'number' ? `${((1 - missRate) * 100).toFixed(1)}%` : '—', + `${off?.events?.merged?.['Orphaned execution result'] ?? 0}→${on?.events?.merged?.['Orphaned execution result'] ?? 0}`, + ]); + } + } + + return markdown(header, rows); +} + +function idleTable(records: RunRecord[]): string | null { + const idle = records.filter((r) => r.idle?.callsPerSecPerProcess !== null && r.idle?.callsPerSecPerProcess !== undefined); + if (idle.length === 0) { + return null; + } + + return markdown( + ['run', 'fast track', 'workers', 'reconcile ms', 'idle calls', 'calls/s/process'], + idle.map((r) => [ + `${r.suite}/${r.label}`, + r.settings.fastTrack ? 'on' : 'off', + r.settings.workers, + r.settings.workerReconcileMs, + r.idle.driverCalls, + fmt(r.idle.callsPerSecPerProcess), + ]), + ); +} + +function report(records: RunRecord[]) { + console.log('\n'); + console.log(markdownTable(records)); + + const idle = idleTable(records); + if (idle) { + console.log('\nIdle floor\n'); + console.log(idle); + } +} + +function dryRun(suites: Suite[], args: Args) { + let totalMs = 0; + const rows: (string | number)[][] = []; + + for (const suite of suites) { + for (const benchRun of selectRuns(suite, args)) { + const est = estimateRunMs(benchRun.env); + totalMs += (est + args.settleMs) * args.fastTrack.length * args.repeat; + + rows.push([ + suite.name, + benchRun.label, + JSON.stringify(benchRun.axis), + fmt(rhoOf(benchRun.env)), + benchRun.env.BENCH_TOTAL_QUERIES, + fmt(parseInt(benchRun.env.BENCH_PERIOD_MS || '0', 10) / 1000, 0), + Math.round(est / 1000), + ]); + } + } + + console.log(markdown(['suite', 'run', 'axis', 'ρ', 'queries', 'period s', 'est. s per pass'], rows)); + console.log(`\n${args.fastTrack.length} pass(es) per run — estimated total ${(totalMs / 60000).toFixed(0)} min`); +} + +(async () => { + const args = parseArgs(hideBin(process.argv)); + + if (args.list) { + for (const suite of SUITES) { + console.log(`${suite.name.padEnd(6)} ${suite.runs.length} runs, ${suite.driver} — ${suite.description}`); + } + + return; + } + + if (args.report) { + const records: RunRecord[] = []; + let unreadable = 0; + + // A sweep that was killed leaves a truncated last line, and one throw here used to lose + // the whole file rather than the one record + for (const line of fs.readFileSync(args.report, 'utf-8').split('\n').filter((l) => l.trim())) { + try { + const record = JSON.parse(line); + if (record.type === 'run') { + records.push(record); + } + } catch (e: any) { + unreadable++; + } + } + + if (unreadable > 0) { + console.error(`!! skipped ${unreadable} unreadable line(s) in ${args.report}`); + } + + report(records); + + return; + } + + const suites = args.suites.includes('all') + ? SUITES.filter((s) => s.name !== 'smoke') + : args.suites.map(suiteByName); + + if (args.dryRun) { + dryRun(suites, args); + + return; + } + + const outPath = args.out || path.resolve(resultsDir(), `${suites.map((s) => s.name).join('+')}-${stamp()}.jsonl`); + const sink = fs.createWriteStream(outPath, { flags: 'a' }); + console.log(`Writing to ${outPath}`); + + const records: RunRecord[] = []; + + for (const suite of suites) { + for (const benchRun of selectRuns(suite, args)) { + for (let pass = 0; pass < args.repeat; pass++) { + for (const fastTrack of args.fastTrack) { + const labelled = args.repeat > 1 + ? { ...benchRun, label: `${benchRun.label}#${pass + 1}` } + : benchRun; + records.push(await executeRun(suite, labelled, fastTrack, args, sink)); + // Let the previous run's connections and Cube Store's own bookkeeping settle + await pausePromise(args.settleMs); + } + } + } + } + + await new Promise((resolve) => sink.end(resolve)); + + report(records); + console.log(`\nResults: ${outPath}`); +})().catch((e) => { + console.error(e); + process.exit(1); +}); diff --git a/packages/cubejs-query-orchestrator/test/benchmarks/suites.ts b/packages/cubejs-query-orchestrator/test/benchmarks/suites.ts new file mode 100644 index 0000000000000..eb4056edf8641 --- /dev/null +++ b/packages/cubejs-query-orchestrator/test/benchmarks/suites.ts @@ -0,0 +1,258 @@ +import { QueuePriority } from '@cubejs-backend/base-driver'; + +export type BenchRun = { + label: string, + axis: Record, + env: Record, +}; + +export type Suite = { + name: string, + description: string, + driver: 'cubestore' | 'memory', + runs: BenchRun[], +}; + +/** + * The published fast track numbers were all taken at reduced payloads and that was never written + * down anywhere. These are the same reduced values, stated once; S7 owns the payload axis. + */ +const DEFAULTS: Record = { + WORKERS: '2', + BENCH_CONCURRENCY: '10', + BENCH_HANDLER_LATENCY_MS: '1500', + BENCH_RESPONSE_SIZE: `${64 * 1024}`, + BENCH_PAYLOAD_SIZE: `${16 * 1024}`, + BENCH_PRIORITY: `${QueuePriority.Interactive}`, + BENCH_WORKER_RECONCILE_MS: '50', + BENCH_WARMUP_QUERIES: '20', + BENCH_IDLE_TAIL_MS: '0', + BENCH_TICK_MS: '1000', +}; + +const num = (env: Record, key: string) => parseInt(env[key] ?? DEFAULTS[key], 10); + +function run(label: string, axis: Record, env: Record): BenchRun { + return { label, axis, env: { ...DEFAULTS, ...env } }; +} + +export function capacityQps(concurrency: number, handlerLatencyMs: number): number { + return (concurrency * 1000) / handlerLatencyMs; +} + +function periodForRho(totalQueries: number, rho: number, concurrency: number, handlerLatencyMs: number): number { + return Math.round((totalQueries * 1000) / (rho * capacityQps(concurrency, handlerLatencyMs))); +} + +export function rhoOf(env: Record): number | null { + const periodMs = num(env, 'BENCH_PERIOD_MS'); + const total = num(env, 'BENCH_TOTAL_QUERIES'); + if (!periodMs || !total) { + return null; + } + + const rate = (total * 1000) / periodMs; + + return rate / capacityQps(num(env, 'BENCH_CONCURRENCY'), num(env, 'BENCH_HANDLER_LATENCY_MS')); +} + +/** Wall clock a run cannot go below: the arrival window, or the time capacity needs to chew through it */ +export function estimateRunMs(env: Record): number { + const total = num(env, 'BENCH_TOTAL_QUERIES'); + const capacity = capacityQps(num(env, 'BENCH_CONCURRENCY'), num(env, 'BENCH_HANDLER_LATENCY_MS')); + const drainMs = total > 0 ? (total / capacity) * 1000 : 0; + const warmupMs = num(env, 'BENCH_WARMUP_QUERIES') > 0 + ? (num(env, 'BENCH_WARMUP_QUERIES') / capacity) * 1000 + 2000 + : 0; + + return Math.round(Math.max(num(env, 'BENCH_PERIOD_MS') || 0, drainMs) + num(env, 'BENCH_IDLE_TAIL_MS') + warmupMs); +} + +const S1: Suite = { + name: 'S1', + description: 'ρ-sweep — the main chart. Driver calls per completed query against the load factor, points clustered around ρ=1 where the knee is.', + driver: 'cubestore', + runs: [0.5, 0.75, 0.9, 1.0, 1.25, 1.75, 2.5].map((rho) => run( + `rho=${rho}`, + { rho }, + { + BENCH_TOTAL_QUERIES: '1000', + BENCH_PERIOD_MS: `${periodForRho(1000, rho, 10, 1500)}`, + } + )), +}; + +const S2: Suite = { + name: 'S2', + description: 'Fan-out at underload. The old report called the effect flat across processes, but measured it at ρ≈15 where the fast track degenerates.', + driver: 'cubestore', + runs: [0, 1, 2, 3, 4, 5].map((workers) => run( + `workers=${workers}`, + { workers, rho: 0.8 }, + { + WORKERS: `${workers}`, + BENCH_TOTAL_QUERIES: '500', + BENCH_PERIOD_MS: `${periodForRho(500, 0.8, 10, 1500)}`, + } + )), +}; + +const S3: Suite = { + name: 'S3', + description: 'Burst then silence. The one shape where the fast track can be a net loss: an ADD_AND_RETRIEVE that comes back empty is a round trip spent for nothing.', + driver: 'cubestore', + runs: [50, 200].map((concurrency) => run( + `burst-c${concurrency}`, + { concurrency }, + { + BENCH_CONCURRENCY: `${concurrency}`, + BENCH_TOTAL_QUERIES: '1000', + BENCH_PERIOD_MS: '5000', + BENCH_IDLE_TAIL_MS: '60000', + } + )), +}; + +const S4: Suite = { + name: 'S4', + description: 'Priority mix, half Interactive half Background. Checks that background never fast-tracks and that it does not starve while interactive does.', + driver: 'cubestore', + runs: [0.8, 1.5].map((rho) => run( + `mix-rho=${rho}`, + { rho }, + { + BENCH_PRIORITY_MIX: '10:50,0:50', + BENCH_TOTAL_QUERIES: '1000', + BENCH_PERIOD_MS: `${periodForRho(1000, rho, 10, 1500)}`, + } + )), +}; + +function idleRun(workers: number, reconcileMs: number): BenchRun { + // Nothing polls without a worker, so the interval is not an axis of the control run + const polled = workers > 0; + + return run( + polled ? `idle-w${workers}-r${reconcileMs}` : `idle-w${workers}`, + polled ? { workers, reconcileMs } : { workers }, + { + WORKERS: `${workers}`, + BENCH_TOTAL_QUERIES: '0', + BENCH_PERIOD_MS: '0', + BENCH_WARMUP_QUERIES: '0', + BENCH_IDLE_TAIL_MS: '60000', + BENCH_WORKER_RECONCILE_MS: `${reconcileMs}`, + } + ); +} + +const S5: Suite = { + name: 'S5', + description: 'Idle floor — driver calls per second per process on an empty queue. On idle the worker reconcile poll is the entire traffic, so both a harness-fast and a realistic interval are measured.', + driver: 'cubestore', + runs: [ + // Nothing polls here: the submitter has no timer of its own and there are no workers, so this + // run is zero by construction. It is the control that says the rest of the table is the poll + // and nothing else — and it carries no reconcile axis, because that setting only reaches workers + idleRun(0, 50), + ...[1, 2, 4].flatMap((workers) => [50, 1000].map((reconcileMs) => idleRun(workers, reconcileMs))), + ], +}; + +const S6: Suite = { + name: 'S6', + description: 'Handler latency sweep at ρ=0.8. The shorter the handler, the larger the overhead share — the saving as a fraction of wall clock should peak at 100ms.', + driver: 'cubestore', + runs: [ + { latencyMs: 100, windowMs: 60000 }, + { latencyMs: 500, windowMs: 60000 }, + { latencyMs: 1500, windowMs: 60000 }, + // 1.6 q/s over a minute is too few samples for a p99, so this point gets a longer window + { latencyMs: 5000, windowMs: 180000 }, + ].map(({ latencyMs, windowMs }) => { + const total = Math.round(0.8 * capacityQps(10, latencyMs) * (windowMs / 1000)); + + return run( + `L=${latencyMs}ms`, + { latencyMs, rho: 0.8 }, + { + BENCH_HANDLER_LATENCY_MS: `${latencyMs}`, + BENCH_TOTAL_QUERIES: `${total}`, + BENCH_PERIOD_MS: `${windowMs}`, + } + ); + }), +}; + +const S7: Suite = { + name: 'S7', + description: 'Payload sweep. ADD_AND_RETRIEVE carries the query def inline, so on fat payloads the saved round trips can be eaten by the bytes.', + driver: 'cubestore', + runs: [ + // eslint-disable-next-line no-bitwise + ...[0, 256 * 1024, 5 << 20, 20 << 20].map((responseSize) => run( + `response=${responseSize}`, + { responseSize, axis: 'response' }, + { + BENCH_RESPONSE_SIZE: `${responseSize}`, + BENCH_PAYLOAD_SIZE: `${16 * 1024}`, + BENCH_TOTAL_QUERIES: '300', + BENCH_PERIOD_MS: `${periodForRho(300, 0.8, 10, 1500)}`, + } + )), + ...[0, 16 * 1024, 256 * 1024, 1024 * 1024].map((payloadSize) => run( + `payload=${payloadSize}`, + { payloadSize, axis: 'payload' }, + { + BENCH_RESPONSE_SIZE: `${64 * 1024}`, + BENCH_PAYLOAD_SIZE: `${payloadSize}`, + BENCH_TOTAL_QUERIES: '300', + BENCH_PERIOD_MS: `${periodForRho(300, 0.8, 10, 1500)}`, + } + )), + ], +}; + +const S8: Suite = { + name: 'S8', + description: 'Crossing ρ=1 from the capacity side instead of the arrival side, at a fixed 8.33 q/s. Tests whether what decides is the budget or the ratio.', + driver: 'cubestore', + runs: [5, 10, 15, 20, 50].map((concurrency) => run( + `c=${concurrency}`, + { concurrency, rho: Number((8.3333 / capacityQps(concurrency, 1500)).toFixed(3)) }, + { + BENCH_CONCURRENCY: `${concurrency}`, + BENCH_TOTAL_QUERIES: '1000', + BENCH_PERIOD_MS: '120000', + } + )), +}; + +const SMOKE: Suite = { + name: 'smoke', + description: 'Two-minute self-check on the memory driver — verifies ticks, percentiles and the result line without a Cube Store.', + driver: 'memory', + runs: [0.5, 1.5].map((rho) => run( + `smoke-rho=${rho}`, + { rho }, + { + WORKERS: '0', + BENCH_CONCURRENCY: '5', + BENCH_HANDLER_LATENCY_MS: '100', + BENCH_TOTAL_QUERIES: '200', + BENCH_PERIOD_MS: `${periodForRho(200, rho, 5, 100)}`, + BENCH_WARMUP_QUERIES: '10', + } + )), +}; + +export const SUITES: Suite[] = [S1, S2, S3, S4, S5, S6, S7, S8, SMOKE]; + +export function suiteByName(name: string): Suite { + const suite = SUITES.find((s) => s.name.toLowerCase() === name.toLowerCase()); + if (!suite) { + throw new Error(`Unknown suite: ${name}. Known: ${SUITES.map((s) => s.name).join(', ')}`); + } + + return suite; +} diff --git a/packages/cubejs-query-orchestrator/test/unit/QueryQueue.abstract.ts b/packages/cubejs-query-orchestrator/test/unit/QueryQueue.abstract.ts index 2dc205f4c1be7..72814fd4b71d2 100644 --- a/packages/cubejs-query-orchestrator/test/unit/QueryQueue.abstract.ts +++ b/packages/cubejs-query-orchestrator/test/unit/QueryQueue.abstract.ts @@ -71,9 +71,9 @@ export const QueryQueueTest = (name: string, options: QueryQueueTestOptions) => readable.pipe(stream); }); }, - sendProcessMessageFn: async (retrieved) => { + sendProcessMessageFn: async (queryKeyHash, queueId, retrieved) => { streamCallOrder.push('dispatch'); - processMessagePromises.push(queue.executeQuery(retrieved)); + processMessagePromises.push(queue.executeQuery(queryKeyHash, queueId, retrieved)); }, sendCancelMessageFn: async (query) => { processCancelPromises.push(queue.processCancel.bind(queue)(query)); @@ -479,10 +479,14 @@ export const QueryQueueTest = (name: string, options: QueryQueueTestOptions) => ); const firstRetrieval = await connection.retrieveForProcessing(key, queueId); - expect(firstRetrieval?.[5]).toBe(true); + expect(firstRetrieval).toMatchObject({ + active: [key], + queueSize: 0, + def: { queryKey: key }, + }); const secondRetrieval = await connection2.retrieveForProcessing(key, queueId); - expect(secondRetrieval).toStrictEqual([0, null, [key], 0, null, false]); + expect(secondRetrieval).toBeNull(); } finally { await connection.getQueryAndRemove(key, null); queue.queueDriver.release(connection); @@ -490,12 +494,14 @@ export const QueryQueueTest = (name: string, options: QueryQueueTestOptions) => } }); - onlyLocalTest('a failed retrieval does not reserve a pending query', async () => { + test('a failed retrieval does not reserve a pending query', async () => { const connection = await queue.queueDriver.createConnection(); const connection2 = await queue.queueDriver.createConnection(); const priority = 10; - const firstKey = 'concurrency-first' as any; - const secondKey = 'concurrency-second' as any; + const firstKey: QueryKey = 'concurrency-first'; + const secondKey: QueryKey = 'concurrency-second'; + const firstHash = connection.redisHash(firstKey); + const secondHash = connection.redisHash(secondKey); try { const [, firstQueueId] = await connection.addToQueue( @@ -509,20 +515,25 @@ export const QueryQueueTest = (name: string, options: QueryQueueTestOptions) => } ); - expect((await connection.retrieveForProcessing(firstKey, firstQueueId))?.[5]).toBe(true); - expect(await connection2.retrieveForProcessing(secondKey, secondQueueId)).toStrictEqual([ - 0, null, [firstKey], 1, null, false - ]); - expect(await connection.getToProcessQueries()).toStrictEqual([[secondKey, secondQueueId]]); + expect(await connection.retrieveForProcessing(firstHash, firstQueueId)).toMatchObject({ + active: [firstHash], + queueSize: 1, + def: { queryKey: firstKey }, + }); + expect(await connection2.retrieveForProcessing(secondHash, secondQueueId)).toBeNull(); + expect(await connection.getToProcessQueries()).toStrictEqual([[secondHash, secondQueueId]]); - await connection.getQueryAndRemove(firstKey, firstQueueId); + await connection.getQueryAndRemove(firstHash, firstQueueId); - const secondRetrieval = await connection2.retrieveForProcessing(secondKey, secondQueueId); - expect(secondRetrieval?.[0]).toBe(1); - expect(secondRetrieval?.[5]).toBe(true); + const secondRetrieval = await connection2.retrieveForProcessing(secondHash, secondQueueId); + expect(secondRetrieval).toMatchObject({ + active: [secondHash], + queueSize: 0, + def: { queryKey: secondKey }, + }); } finally { - await connection.getQueryAndRemove(firstKey, null); - await connection.getQueryAndRemove(secondKey, null); + await connection.getQueryAndRemove(firstHash, null); + await connection.getQueryAndRemove(secondHash, null); queue.queueDriver.release(connection); queue.queueDriver.release(connection2); } @@ -540,7 +551,11 @@ export const QueryQueueTest = (name: string, options: QueryQueueTestOptions) => queueId: queue.generateQueueId(), stageQueryKey: key, requestId: '1' } ); - expect((await connection.retrieveForProcessing(key, staleQueueId))?.[5]).toBe(true); + expect(await connection.retrieveForProcessing(key, staleQueueId)).toMatchObject({ + active: [key], + queueSize: 0, + def: { queryKey }, + }); await connection.getQueryAndRemove(key, staleQueueId); const [, currentQueueId] = await connection.addToQueue( @@ -550,12 +565,14 @@ export const QueryQueueTest = (name: string, options: QueryQueueTestOptions) => ); if (options.cacheAndQueueDriver !== 'cubestore') { - expect(await connection.retrieveForProcessing(key, staleQueueId)).toStrictEqual([ - 0, null, [], 1, null, false - ]); + expect(await connection.retrieveForProcessing(key, staleQueueId)).toBeNull(); expect(await connection.getToProcessQueries()).toStrictEqual([[key, currentQueueId]]); } - expect((await connection.retrieveForProcessing(key, currentQueueId))?.[5]).toBe(true); + expect(await connection.retrieveForProcessing(key, currentQueueId)).toMatchObject({ + active: [key], + queueSize: 0, + def: { queryKey, query: ['new'] }, + }); const staleUpdateResult = await connection.optimisticQueryUpdate(key, { stale: true }, staleQueueId); if (options.cacheAndQueueDriver !== 'cubestore') { @@ -756,8 +773,7 @@ export const QueryQueueTest = (name: string, options: QueryQueueTestOptions) => // concurrency is 1, the first query takes the only slot const [added1, , , , retrieved1] = await addToQueue(first, 1); expect(added1).toBe(1); - expect(retrieved1?.[5]).toBe(true); - expect(retrieved1?.[4].queryKey).toStrictEqual(first); + expect(retrieved1?.def.queryKey).toStrictEqual(first); // an active item is never retrieved twice const [added1again, , , , retrieved1again] = await addToQueue(first, 1); diff --git a/packages/cubejs-schema-compiler/src/compiler/CubeEvaluator.ts b/packages/cubejs-schema-compiler/src/compiler/CubeEvaluator.ts index 4b2fa35409886..2909d43bf8c2e 100644 --- a/packages/cubejs-schema-compiler/src/compiler/CubeEvaluator.ts +++ b/packages/cubejs-schema-compiler/src/compiler/CubeEvaluator.ts @@ -299,6 +299,12 @@ export class CubeEvaluator extends CubeSymbols { return `${cube.name}.${match.name}`; }; + // A view extending another one inherits its `default_filters` entries by + // reference, so the resolved references have to go into a copy owned by this + // view. Written in place they would resolve to whichever view is prepared + // last, pointing the other views' filters at members they do not include. + cube.defaultFilters = (cube.defaultFilters as ViewDefaultValueFilter[]).map(f => ({ ...f })); + for (const filter of cube.defaultFilters as ViewDefaultValueFilter[]) { const rawMember = this.evaluateReferences(cube.name, filter.member); const resolved = resolveViewMember('member', rawMember); @@ -641,24 +647,33 @@ export class CubeEvaluator extends CubeSymbols { : {}), })); } + // `filter` and `grain` are nested objects, and a cube extending another + // one inherits them by reference, so the resolved references have to go + // into a copy owned by this cube. Written in place they would resolve to + // whichever cube is prepared last, pointing every member of the other + // cubes at that cube's dimensions. if (member.filter) { - if (typeof member.filter.exclude === 'function') { - member.filter.excludeReferences = this.evaluateReferences(cubeName, member.filter.exclude); + const filter = { ...member.filter }; + if (typeof filter.exclude === 'function') { + filter.excludeReferences = this.evaluateReferences(cubeName, filter.exclude); } - if (typeof member.filter.keepOnly === 'function') { - member.filter.keepOnlyReferences = this.evaluateReferences(cubeName, member.filter.keepOnly); + if (typeof filter.keepOnly === 'function') { + filter.keepOnlyReferences = this.evaluateReferences(cubeName, filter.keepOnly); } + member.filter = filter; } if (member.grain) { - if (typeof member.grain.exclude === 'function') { - member.grain.excludeReferences = this.evaluateReferences(cubeName, member.grain.exclude); + const grain = { ...member.grain }; + if (typeof grain.exclude === 'function') { + grain.excludeReferences = this.evaluateReferences(cubeName, grain.exclude); } - if (typeof member.grain.keepOnly === 'function') { - member.grain.keepOnlyReferences = this.evaluateReferences(cubeName, member.grain.keepOnly); + if (typeof grain.keepOnly === 'function') { + grain.keepOnlyReferences = this.evaluateReferences(cubeName, grain.keepOnly); } - if (typeof member.grain.include === 'function') { - member.grain.includeReferences = this.evaluateReferences(cubeName, member.grain.include); + if (typeof grain.include === 'function') { + grain.includeReferences = this.evaluateReferences(cubeName, grain.include); } + member.grain = grain; } } } @@ -711,7 +726,7 @@ export class CubeEvaluator extends CubeSymbols { protected preparePreAggregations(cube: any, errorReporter: ErrorReporter) { if (cube.preAggregations) { // eslint-disable-next-line no-restricted-syntax - for (const preAggregation of Object.values(cube.preAggregations) as any) { + for (const [preAggregationName, preAggregation] of Object.entries(cube.preAggregations) as any) { // preAggregation is actually (PreAggregationDefinitionRollup | PreAggregationDefinitionOriginalSql) if (preAggregation.timeDimension) { preAggregation.timeDimensionReference = preAggregation.timeDimension; @@ -767,10 +782,17 @@ export class CubeEvaluator extends CubeSymbols { delete preAggregation.buildRangeEnd; } + // `outputColumnTypes` names are resolved against this cube, and a cube + // extending another one inherits the pre-aggregation by reference, so + // both the entry and its columns have to be copies owned by this cube. if (preAggregation.outputColumnTypes) { - preAggregation.outputColumnTypes.forEach(column => { - column.name = this.evaluateReferences(cube.name, column.member, { originalSorting: true }); - }); + cube.preAggregations[preAggregationName] = { + ...preAggregation, + outputColumnTypes: preAggregation.outputColumnTypes.map(column => ({ + ...column, + name: this.evaluateReferences(cube.name, column.member, { originalSorting: true }), + })), + }; } } } diff --git a/packages/cubejs-schema-compiler/test/unit/extends-shared-definitions.test.ts b/packages/cubejs-schema-compiler/test/unit/extends-shared-definitions.test.ts new file mode 100644 index 0000000000000..0c239f617dd56 --- /dev/null +++ b/packages/cubejs-schema-compiler/test/unit/extends-shared-definitions.test.ts @@ -0,0 +1,255 @@ +import { PostgresQuery } from '../../src/adapter/PostgresQuery'; +import { prepareYamlCompiler } from './PrepareCompiler'; + +// `extends` hands the extending cube the very definitions of the cube it extends, +// so every reference resolved per cube — a multi-stage `grain:`/`filter:`, a view +// default filter, a pre-aggregation's output column names — has to be written into +// an object owned by that cube. Written into the shared one it resolves to whichever +// cube is prepared last, and the other cubes end up carrying member paths of a cube +// that is not theirs. +describe('Multi-stage members of a cube that another cube extends', () => { + const baseFact = ` + - name: base_fact + sql: "SELECT 1 AS id, 1 AS dim_id, '2026-01-01'::date AS d, 10 AS v" + joins: + - name: dims + sql: "{CUBE}.dim_id = {dims}.id" + relationship: many_to_one + dimensions: + - name: id + sql: "{CUBE}.id" + type: number + primary_key: true + - name: d + sql: "{CUBE}.d" + type: time + - name: v + sql: "{CUBE}.v" + type: number + measures: + - name: v_sum + sql: "{v}" + type: sum + - name: daily_v + multi_stage: true + sql: "{v_sum}" + type: number + - name: linked + multi_stage: true + sql: "{daily_v}" + type: sum + grain: + include: + - d + - name: combining + multi_stage: true + sql: "CASE WHEN {d} IS NOT NULL THEN {daily_v} ELSE 0 END" + type: max + grain: + include: + - d + - name: outer_combining + multi_stage: true + sql: "{combining}" + type: number + - name: v_sum_all_dates + multi_stage: true + sql: "{v_sum}" + type: number + filter: + exclude: + - d +`; + + const dims = ` + - name: dims + sql: "SELECT 1 AS id, 'a' AS name" + dimensions: + - name: id + sql: "{CUBE}.id" + type: number + primary_key: true + - name: name + sql: "{CUBE}.name" + type: string +`; + + const childFact = ` + - name: child_fact + extends: base_fact + sql: "SELECT 1 AS id, 1 AS dim_id, '2026-01-01'::date AS d, 10 AS v, 'x' AS tag" + dimensions: + - name: tag + sql: "{CUBE}.tag" + type: string +`; + + const model = (withChild: boolean) => `cubes:${dims}${baseFact}${withChild ? childFact : ''}`; + + const compile = async (withChild: boolean) => { + const compilers = prepareYamlCompiler(model(withChild)); + await compilers.compiler.compile(); + return compilers; + }; + + const buildSql = async (withChild: boolean, query: any, useNativeSqlPlanner: boolean) => { + const compilers = await compile(withChild); + return new PostgresQuery(compilers, { + timezone: 'UTC', + useNativeSqlPlanner, + ...query, + }).buildSqlAndParams(); + }; + + it('resolves grain references against the cube that declares the member', async () => { + const { cubeEvaluator } = await compile(true); + + expect(cubeEvaluator.evaluatedCubes.base_fact.measures.linked.grain?.includeReferences) + .toEqual(['base_fact.d']); + expect(cubeEvaluator.evaluatedCubes.child_fact.measures.linked.grain?.includeReferences) + .toEqual(['child_fact.d']); + }); + + it('resolves filter references against the cube that declares the member', async () => { + const { cubeEvaluator } = await compile(true); + + expect(cubeEvaluator.evaluatedCubes.base_fact.measures.v_sum_all_dates.filter?.excludeReferences) + .toEqual(['base_fact.d']); + expect(cubeEvaluator.evaluatedCubes.child_fact.measures.v_sum_all_dates.filter?.excludeReferences) + .toEqual(['child_fact.d']); + }); + + describe.each([ + ['native', true], + ['legacy', false], + ])('%s planner', (_name, useNativeSqlPlanner) => { + // Planning the parent's member must not depend on the extending cube being + // there at all, so the plan is compared against the same model without it. + const expectSamePlanWithAndWithoutChild = async (query: any) => { + const [withoutChild] = await buildSql(false, query, useNativeSqlPlanner); + const [withChild] = await buildSql(true, query, useNativeSqlPlanner); + expect(withChild).toEqual(withoutChild); + }; + + it('plans a multi-stage measure with an explicit grain', async () => { + await expectSamePlanWithAndWithoutChild({ measures: ['base_fact.linked'] }); + await expect(buildSql(true, { measures: ['child_fact.linked'] }, useNativeSqlPlanner)).resolves.toBeDefined(); + }); + + it('plans a chained multi-stage measure reading a grain dimension', async () => { + await expectSamePlanWithAndWithoutChild({ measures: ['base_fact.outer_combining'] }); + await expectSamePlanWithAndWithoutChild({ + measures: ['base_fact.outer_combining'], + dimensions: ['dims.name'], + }); + }); + + it('plans a multi-stage measure with a filter directive', async () => { + await expectSamePlanWithAndWithoutChild({ + measures: ['base_fact.v_sum_all_dates'], + timeDimensions: [{ + dimension: 'base_fact.d', + granularity: 'month', + dateRange: ['2026-01-01', '2026-01-31'], + }], + }); + }); + + it('plans a plain measure of a cube that another cube extends', async () => { + await expect(buildSql(true, { measures: ['base_fact.v_sum'] }, useNativeSqlPlanner)).resolves.toBeDefined(); + }); + }); +}); + +describe('View default filters of a view that another view extends', () => { + const model = ` +cubes: + - name: orders + sql: "SELECT 1 AS id, 'usd' AS currency" + dimensions: + - name: id + sql: "{CUBE}.id" + type: number + primary_key: true + - name: currency + sql: "{CUBE}.currency" + type: string + measures: + - name: count + type: count + +views: + - name: base_view + cubes: + - join_path: orders + includes: + - currency + - count + default_filters: + - member: currency + operator: equals + values: ["usd"] + + - name: child_view + extends: base_view +`; + + it('resolves the filter member against the view that declares the filter', async () => { + const { compiler, cubeEvaluator } = prepareYamlCompiler(model); + await compiler.compile(); + + expect(cubeEvaluator.evaluatedCubes.base_view.defaultFilters?.map(f => f.memberReference)) + .toEqual(['base_view.currency']); + expect(cubeEvaluator.evaluatedCubes.child_view.defaultFilters?.map(f => f.memberReference)) + .toEqual(['child_view.currency']); + }); +}); + +describe('Pre-aggregations of a cube that another cube extends', () => { + const model = ` +cubes: + - name: base + sql: "SELECT 1 AS id, '2026-01-01'::date AS d" + dimensions: + - name: id + sql: "{CUBE}.id" + type: number + primary_key: true + - name: d + sql: "{CUBE}.d" + type: time + measures: + - name: count + type: count + pre_aggregations: + - name: main + dimensions: + - id + measures: + - count + time_dimension: d + granularity: day + output_column_types: + - member: id + type: integer + + - name: child + extends: base + sql: "SELECT 1 AS id, '2026-01-01'::date AS d, 'x' AS tag" + dimensions: + - name: tag + sql: "{CUBE}.tag" + type: string +`; + + it('resolves output column names against the cube that declares the pre-aggregation', async () => { + const { compiler, cubeEvaluator } = prepareYamlCompiler(model); + await compiler.compile(); + + const names = (cube: string) => (cubeEvaluator.evaluatedCubes[cube].preAggregations.main as any) + .outputColumnTypes.map((c: any) => c.name); + + expect(names('base')).toEqual(['base.id']); + expect(names('child')).toEqual(['child.id']); + }); +}); diff --git a/rust/cube/cubesqlplanner/cubesqlplanner/src/planner/planners/multi_stage/multi_stage_query_planner.rs b/rust/cube/cubesqlplanner/cubesqlplanner/src/planner/planners/multi_stage/multi_stage_query_planner.rs index 845bbe7cba12b..97f9b488d943c 100644 --- a/rust/cube/cubesqlplanner/cubesqlplanner/src/planner/planners/multi_stage/multi_stage_query_planner.rs +++ b/rust/cube/cubesqlplanner/cubesqlplanner/src/planner/planners/multi_stage/multi_stage_query_planner.rs @@ -414,15 +414,100 @@ impl MultiStageQueryPlanner { }) { return Ok(()); } + let grain = Self::grain_members(grain_state, parent_state); + // A granularity of the very dimension the sql reads looks like a match in + // the listing, so the one case where reading the grain is not enough to + // tell them apart is spelled out. + let target = dimension.clone().resolve_reference_chain().full_name(); + let hint = if grain + .iter() + .any(|m| Self::granular_time_dimension_base(m).as_ref() == Some(&target)) + { + format!( + " The grain carries {target} at a granularity, which is a value of its own and not \ + the dimension itself." + ) + } else { + String::new() + }; + let grain = if grain.is_empty() { + "no dimensions".to_string() + } else { + grain + .iter() + .map(Self::describe_grain_member) + .collect::>() + .join(", ") + }; Err(CubeError::user(format!( "Multi-stage member {member} reads dimension {dimension}, which is not part of the \ - grain it is computed at. Add {dimension} to `grain.include` of {member}, or remove \ - it from the member's sql.", + grain it is computed at ({grain}).{hint} Add {dimension} to `grain.include` of \ + {member}, or remove it from the member's sql.", member = member.full_name(), dimension = dimension.full_name(), ))) } + // The grain the member is computed at: the stage's own dimensions and, when + // the assembly broadcasts back onto the query grid, the keys side. + fn grain_members( + grain_state: &QueryProperties, + parent_state: &QueryProperties, + ) -> Vec> { + let mut members: Vec> = Vec::new(); + for state in [grain_state, parent_state] { + for dimension in state + .dimensions() + .iter() + .chain(state.time_dimensions().iter()) + { + let resolved = dimension.clone().resolve_reference_chain(); + if !members + .iter() + .any(|m| m.full_name() == resolved.full_name()) + { + members.push(resolved); + } + } + } + members + } + + // A time dimension carries its granularity inside its name, which reads as a + // member of its own; the granularity is named separately instead. + fn describe_grain_member(member: &Rc) -> String { + match member.as_ref() { + MemberSymbol::TimeDimension(time_dimension) => match time_dimension.granularity() { + Some(granularity) => format!( + "{} ({})", + time_dimension + .base_symbol() + .clone() + .resolve_reference_chain() + .full_name(), + granularity + ), + None => member.full_name(), + }, + _ => member.full_name(), + } + } + + fn granular_time_dimension_base(member: &Rc) -> Option { + match member.as_ref() { + MemberSymbol::TimeDimension(time_dimension) => { + time_dimension.granularity().as_ref().map(|_| { + time_dimension + .base_symbol() + .clone() + .resolve_reference_chain() + .full_name() + }) + } + _ => None, + } + } + /// Plans CASE-SWITCH dependencies: collects, per dependency, the /// union of switch values it covers and renders each dependency /// under a state with an equality filter on the switch member diff --git a/rust/cube/cubesqlplanner/cubesqlplanner/src/test_fixtures/schemas/yaml_files/common/integration_multi_stage.yaml b/rust/cube/cubesqlplanner/cubesqlplanner/src/test_fixtures/schemas/yaml_files/common/integration_multi_stage.yaml index 43f33edd26866..86cd98c2bfddb 100644 --- a/rust/cube/cubesqlplanner/cubesqlplanner/src/test_fixtures/schemas/yaml_files/common/integration_multi_stage.yaml +++ b/rust/cube/cubesqlplanner/cubesqlplanner/src/test_fixtures/schemas/yaml_files/common/integration_multi_stage.yaml @@ -414,6 +414,16 @@ cubes: include: - orders.created_at + # The declared grain is a dimension of another cube, so it does not + # supply the dimension the sql reads. + - name: amount_first_half_of_month_grain_of_other_cube + type: sum + sql: "CASE WHEN EXTRACT(DAY FROM {CUBE.created_at}) <= 15 THEN {CUBE.total_amount} ELSE 0 END" + multi_stage: true + grain: + include: + - customers.city + # Undeclared read one stage further out: the dimension is read by a # measure whose aggregated dependency is itself multi-stage. - name: amount_prev_month_first_half diff --git a/rust/cube/cubesqlplanner/cubesqlplanner/src/tests/integration/multi_stage/dimension_deps.rs b/rust/cube/cubesqlplanner/cubesqlplanner/src/tests/integration/multi_stage/dimension_deps.rs index c5383476b0e58..aa8f17d2b6796 100644 --- a/rust/cube/cubesqlplanner/cubesqlplanner/src/tests/integration/multi_stage/dimension_deps.rs +++ b/rust/cube/cubesqlplanner/cubesqlplanner/src/tests/integration/multi_stage/dimension_deps.rs @@ -58,6 +58,36 @@ async fn test_undeclared_time_dimension_read_is_reported() { ); } +#[tokio::test(flavor = "multi_thread")] +async fn test_report_spells_out_the_grain_the_member_is_computed_at() { + let message = expect_error("amount_first_half_of_month"); + + assert!( + message.contains("orders.created_at (month)"), + "The error must spell out the grain the member is computed at:\n{}", + message + ); + assert!( + message.contains("at a granularity"), + "A granularity of the dimension the sql reads must not read as a match:\n{}", + message + ); +} + +/// A grain declared at another cube's dimension: naming the read dimension +/// alone would repeat what the model already says, so the grain the member is +/// actually computed at is what tells the two apart. +#[tokio::test(flavor = "multi_thread")] +async fn test_report_names_a_grain_declared_at_another_cube() { + let message = expect_error("amount_first_half_of_month_grain_of_other_cube"); + + assert!( + message.contains("orders.created_at") && message.contains("customers.city"), + "The error must name both the dimension the sql reads and the declared grain:\n{}", + message + ); +} + /// The reading member consumes a time-shifted multi-stage measure, so its own /// grain is settled one stage above the leaf that would have to carry the /// dimension. diff --git a/yarn.lock b/yarn.lock index a0e0d26cf1d44..08be06a73eb37 100644 --- a/yarn.lock +++ b/yarn.lock @@ -8781,7 +8781,7 @@ resolved "https://registry.yarnpkg.com/@types/yargs-parser/-/yargs-parser-20.2.1.tgz#3b9ce2489919d9e4fea439b76916abc34b2df129" integrity sha512-7tFImggNeNBVMsn0vLrpn1H1uPrUBdnARPTpZoitY37ZrdJREzf7I16tMrlK3hen349gr1NYh8CmZQa7CTG6Aw== -"@types/yargs@^17.0.8": +"@types/yargs@^17.0.31", "@types/yargs@^17.0.8": version "17.0.31" resolved "https://registry.yarnpkg.com/@types/yargs/-/yargs-17.0.31.tgz#8fd0089803fd55d8a285895a18b88cb71a99683c" integrity sha512-bocYSx4DI8TmdlvxqGpVNXOgCNR1Jj0gNPhhAY+iz1rgKDAaYrAYdFYnhDV1IFuiuVc9HkOwyDcFxaTElF3/wg==