diff --git a/meerkat-browser/package.json b/meerkat-browser/package.json index 28d2537f..ddf89312 100644 --- a/meerkat-browser/package.json +++ b/meerkat-browser/package.json @@ -1,6 +1,6 @@ { "name": "@devrev/meerkat-browser", - "version": "0.0.133", + "version": "0.0.134", "dependencies": { "tslib": "^2.3.0", "@devrev/meerkat-core": "*", diff --git a/meerkat-browser/src/browser-cube-to-sql/browser-cube-to-sql.ts b/meerkat-browser/src/browser-cube-to-sql/browser-cube-to-sql.ts index ecf7165e..0298a6a4 100644 --- a/meerkat-browser/src/browser-cube-to-sql/browser-cube-to-sql.ts +++ b/meerkat-browser/src/browser-cube-to-sql/browser-cube-to-sql.ts @@ -2,6 +2,7 @@ import { BASE_TABLE_NAME, ContextParams, Query, + SqlGenerationOnEvent, TableSchema, applyFilterParamsToBaseSQL, applyProjectionToSQLQuery, @@ -13,6 +14,7 @@ import { getCombinedTableSchema, getFilterParamsSQL, getFinalBaseSQL, + timePhase, } from '@devrev/meerkat-core'; import { AsyncDuckDBConnection } from '@duckdb/duckdb-wasm'; @@ -30,6 +32,11 @@ export interface CubeQueryToSQLParams { query: Query; tableSchemas: TableSchema[]; contextParams?: ContextParams; + /** + * Optional per-call callback for SQL-generation phase timings. Invoked once + * per phase plus once for `total`. No-op when omitted. + */ + onEvent?: SqlGenerationOnEvent; } export const cubeQueryToSQL = async ({ @@ -37,19 +44,27 @@ export const cubeQueryToSQL = async ({ query, tableSchemas, contextParams, + onEvent, }: CubeQueryToSQLParams) => { - const updatedTableSchemas: TableSchema[] = await Promise.all( - tableSchemas.map(async (schema: TableSchema) => { - const baseFilterParamsSQL = await getFinalBaseSQL({ - query, - tableSchema: schema, - getQueryOutput: (query) => getQueryOutput(query, connection), - }); - return { - ...schema, - sql: baseFilterParamsSQL, - }; - }) + const totalStart = performance.now(); + + const updatedTableSchemas = await timePhase( + 'base_sql', + () => + Promise.all( + tableSchemas.map(async (schema: TableSchema) => { + const baseFilterParamsSQL = await getFinalBaseSQL({ + query, + tableSchema: schema, + getQueryOutput: (query) => getQueryOutput(query, connection), + }); + return { + ...schema, + sql: baseFilterParamsSQL, + }; + }) + ), + onEvent ); const updatedTableSchema = await getCombinedTableSchema( @@ -57,63 +72,92 @@ export const cubeQueryToSQL = async ({ query ); - const ast = cubeToDuckdbAST(query, updatedTableSchema, { - filterType: 'PROJECTION_FILTER', - }); + const ast = await timePhase( + 'ast_build', + async () => + cubeToDuckdbAST(query, updatedTableSchema, { + filterType: 'PROJECTION_FILTER', + }), + onEvent + ); if (!ast) { throw new Error('Could not generate AST'); } - const queryTemp = astDeserializerQuery(ast); + const preBaseQuery = await timePhase( + 'ast_deserialize_roundtrip', + async () => { + const queryTemp = astDeserializerQuery(ast); + const arrowResult = await connection.query(queryTemp); + const parsedOutputQuery = arrowResult + .toArray() + .map((row) => row.toJSON()); + return deserializeQuery(parsedOutputQuery); + }, + onEvent + ); - const arrowResult = await connection.query(queryTemp); - const parsedOutputQuery = arrowResult.toArray().map((row) => row.toJSON()); + const filterParamsSQL = await timePhase( + 'filter_params', + () => + getFilterParamsSQL({ + getQueryOutput: (query) => getQueryOutput(query, connection), + query, + tableSchema: updatedTableSchema, + filterType: 'PROJECTION_FILTER', + }), + onEvent + ); - const preBaseQuery = deserializeQuery(parsedOutputQuery); - const filterParamsSQL = await getFilterParamsSQL({ - getQueryOutput: (query) => getQueryOutput(query, connection), - query, - tableSchema: updatedTableSchema, - filterType: 'PROJECTION_FILTER', - }); + const finalQuery = await timePhase( + 'projections', + async () => { + const filterParamQuery = applyFilterParamsToBaseSQL( + updatedTableSchema.sql, + filterParamsSQL + ); - const filterParamQuery = applyFilterParamsToBaseSQL( - updatedTableSchema.sql, - filterParamsSQL - ); + /** + * Replace CONTEXT_PARAMS with context params + */ + const baseQuery = detectApplyContextParamsToBaseSQL( + filterParamQuery, + contextParams || {} + ); - /** - * Replace CONTEXT_PARAMS with context params - */ - const baseQuery = detectApplyContextParamsToBaseSQL( - filterParamQuery, - contextParams || {} - ); + /** + * Replace BASE_TABLE_NAME with cube query + */ + const replaceBaseTableName = preBaseQuery.replace( + BASE_TABLE_NAME, + `(${baseQuery}) AS ${updatedTableSchema.name}` + ); - /** - * Replace BASE_TABLE_NAME with cube query - */ - const replaceBaseTableName = preBaseQuery.replace( - BASE_TABLE_NAME, - `(${baseQuery}) AS ${updatedTableSchema.name}` - ); + /** + * Add measures to the query + */ + const measures = query.measures; + const dimensions = query.dimensions || []; + const queryWithProjections = applyProjectionToSQLQuery( + dimensions, + measures, + updatedTableSchemas, + replaceBaseTableName + ); - /** - * Add measures to the query - */ - const measures = query.measures; - const dimensions = query.dimensions || []; - const queryWithProjections = applyProjectionToSQLQuery( - dimensions, - measures, - updatedTableSchemas, - replaceBaseTableName + /** + * Replace SQL expression placeholders with actual SQL + */ + return applySQLExpressions(queryWithProjections); + }, + onEvent ); - /** - * Replace SQL expression placeholders with actual SQL - */ - const finalQuery = applySQLExpressions(queryWithProjections); + onEvent?.({ + event_name: 'sql_generation_duration', + phase: 'total', + duration: performance.now() - totalStart, + }); return finalQuery; }; diff --git a/meerkat-core/package.json b/meerkat-core/package.json index 46797006..f74bf625 100644 --- a/meerkat-core/package.json +++ b/meerkat-core/package.json @@ -1,6 +1,6 @@ { "name": "@devrev/meerkat-core", - "version": "0.0.133", + "version": "0.0.134", "dependencies": { "tslib": "^2.3.0" }, diff --git a/meerkat-core/src/index.ts b/meerkat-core/src/index.ts index c91e1f4e..7c9382d8 100644 --- a/meerkat-core/src/index.ts +++ b/meerkat-core/src/index.ts @@ -47,3 +47,4 @@ export type { EnsureColumnAliasBatchResult, } from './utils/ensure-sql-expression-column-alias'; export * from './utils/ensure-table-schema-alias-sql'; +export * from './instrumentation/sql-generation-event'; diff --git a/meerkat-core/src/instrumentation/sql-generation-event.spec.ts b/meerkat-core/src/instrumentation/sql-generation-event.spec.ts new file mode 100644 index 00000000..e5b056a1 --- /dev/null +++ b/meerkat-core/src/instrumentation/sql-generation-event.spec.ts @@ -0,0 +1,60 @@ +import { SqlGenerationEvent, timePhase } from './sql-generation-event'; + +describe('timePhase', () => { + it('returns the wrapped result unchanged', async () => { + const result = await timePhase('base_sql', async () => 'sql-string'); + expect(result).toBe('sql-string'); + }); + + it('does not invoke onEvent when omitted', async () => { + // Simply must not throw when no callback is supplied. + await expect( + timePhase('ast_build', async () => 42) + ).resolves.toBe(42); + }); + + it('emits a sql_generation_duration event for the phase when onEvent is supplied', async () => { + const events: SqlGenerationEvent[] = []; + + await timePhase('filter_params', async () => 'x', (e) => events.push(e)); + + expect(events).toHaveLength(1); + expect(events[0]).toEqual( + expect.objectContaining({ + event_name: 'sql_generation_duration', + phase: 'filter_params', + }) + ); + expect(typeof events[0].duration).toBe('number'); + expect(events[0].duration).toBeGreaterThanOrEqual(0); + }); + + it('forwards metadata onto the event', async () => { + const events: SqlGenerationEvent[] = []; + + await timePhase( + 'projections', + async () => null, + (e) => events.push(e), + { queryId: 'q-1' } + ); + + expect(events[0].metadata).toEqual({ queryId: 'q-1' }); + }); + + it('does not emit an event when the phase throws', async () => { + const events: SqlGenerationEvent[] = []; + + await expect( + timePhase( + 'ast_deserialize_roundtrip', + async () => { + throw new Error('boom'); + }, + (e) => events.push(e) + ) + ).rejects.toThrow('boom'); + + expect(events).toHaveLength(0); + }); +}); diff --git a/meerkat-core/src/instrumentation/sql-generation-event.ts b/meerkat-core/src/instrumentation/sql-generation-event.ts new file mode 100644 index 00000000..275ba3ea --- /dev/null +++ b/meerkat-core/src/instrumentation/sql-generation-event.ts @@ -0,0 +1,59 @@ +/** + * Instrumentation for the cube-query -> SQL generation path. This lives in + * meerkat-core (which meerkat-node/meerkat-browser already depend on) so the + * SQL builders can emit timings without taking a dependency on meerkat-dbm's + * DBMEvent. It is intentionally a separate event shape from DBMEvent: the two + * cover different phases (generation vs. execution) and different packages. + */ + +/** + * Distinct phases of `cubeQueryToSQL`. `total` wraps the whole call; the rest + * are the individually timed steps that add up to (roughly) it. + */ +export type SqlGenerationPhase = + | 'base_sql' + | 'ast_build' + | 'ast_deserialize_roundtrip' + | 'filter_params' + | 'projections' + | 'total'; + +export interface SqlGenerationEvent { + event_name: 'sql_generation_duration'; + phase: SqlGenerationPhase; + /** Duration in milliseconds. */ + duration: number; + metadata?: object; +} + +/** + * Optional per-call callback for SQL-generation timings. Supplied by the caller + * of `cubeQueryToSQL`; invoked once per phase (plus once for `total`). + */ +export type SqlGenerationOnEvent = (event: SqlGenerationEvent) => void; + +/** + * Time an async phase and, when an `onEvent` is supplied, emit its duration. + * Returns the phase's result untouched so call sites read as a thin wrapper. + * The event fires even if `fn` throws is NOT desired here — a failed phase has + * no meaningful duration to report, so timing is emitted only on success. + */ +export const timePhase = async ( + phase: SqlGenerationPhase, + fn: () => Promise, + onEvent?: SqlGenerationOnEvent, + metadata?: object +): Promise => { + if (!onEvent) { + return fn(); + } + const start = performance.now(); + const result = await fn(); + onEvent({ + event_name: 'sql_generation_duration', + phase, + duration: performance.now() - start, + metadata, + }); + return result; +}; diff --git a/meerkat-dbm/package.json b/meerkat-dbm/package.json index 5693972c..b46efb7c 100644 --- a/meerkat-dbm/package.json +++ b/meerkat-dbm/package.json @@ -1,6 +1,6 @@ { "name": "@devrev/meerkat-dbm", - "version": "0.1.47", + "version": "0.1.48", "dependencies": { "@duckdb/duckdb-wasm": "1.32.0", "apache-arrow": "^17.0.0", diff --git a/meerkat-dbm/src/dbm/__test__/dbm-parallel.spec.ts b/meerkat-dbm/src/dbm/__test__/dbm-parallel.spec.ts index 19d4e9a7..95b46e04 100644 --- a/meerkat-dbm/src/dbm/__test__/dbm-parallel.spec.ts +++ b/meerkat-dbm/src/dbm/__test__/dbm-parallel.spec.ts @@ -25,6 +25,8 @@ const iFrameRunnerManager = { addIFrameManager: jest.fn(), areRunnersRunning: jest.fn(), messageListener: jest.fn(), + registerQueryEventCallback: jest.fn(), + unregisterQueryEventCallback: jest.fn(), } as unknown as jest.Mocked; const runnerMock = { @@ -465,6 +467,35 @@ describe('DBMParallel', () => { ); }); + it('registers the per-query onEvent and strips it from the iframe options', async () => { + const onEvent = jest.fn(); + + runnerMock.communication.sendRequest.mockResolvedValue({ + message: { isError: false, data: [{ data: 1 }] }, + }); + + await dbmParallel.queryWithTables({ + query: 'SELECT * FROM table', + tables: [], + options: { onEvent }, + }); + + expect(iFrameRunnerManager.registerQueryEventCallback).toHaveBeenCalledWith( + expect.any(String), + onEvent + ); + expect(iFrameRunnerManager.unregisterQueryEventCallback).toHaveBeenCalledWith( + expect.any(String) + ); + expect(runnerMock.communication.sendRequest).toHaveBeenCalledWith( + expect.objectContaining({ + payload: expect.objectContaining({ + options: expect.objectContaining({ onEvent: undefined }), + }), + }) + ); + }); + it('should not log error when query is aborted by user', async () => { const abortController = new AbortController(); @@ -496,5 +527,38 @@ describe('DBMParallel', () => { // Logger should not have been called since it was a user-initiated abort expect(loggerMock.error).not.toHaveBeenCalled(); }); + + it('unregisters the per-query onEvent when the query is aborted', async () => { + const abortController = new AbortController(); + const onEvent = jest.fn(); + + runnerMock.communication.sendRequest.mockImplementation( + () => + new Promise(() => { + // Never resolves + }) + ); + + const queryPromise = dbmParallel.queryWithTables({ + query: 'SELECT * FROM table', + tables: [], + options: { signal: abortController.signal, onEvent }, + }); + + await new Promise((resolve) => setTimeout(resolve, 10)); + + abortController.abort(); + + await expect(queryPromise).rejects.toThrow('Query aborted by user'); + + // The callback registered for this runner must be torn down even though + // the abort handler cleared activeQueries first. + const registeredRunnerId = ( + iFrameRunnerManager.registerQueryEventCallback as jest.Mock + ).mock.calls[0][0]; + expect(iFrameRunnerManager.unregisterQueryEventCallback).toHaveBeenCalledWith( + registeredRunnerId + ); + }); }); }); diff --git a/meerkat-dbm/src/dbm/__test__/dbm.spec.ts b/meerkat-dbm/src/dbm/__test__/dbm.spec.ts index 4156e611..f80fbbe7 100644 --- a/meerkat-dbm/src/dbm/__test__/dbm.spec.ts +++ b/meerkat-dbm/src/dbm/__test__/dbm.spec.ts @@ -192,6 +192,68 @@ describe('DBM', () => { expect(result).toEqual(['SELECT * FROM table1']); }); + it('should invoke the per-query onEvent for this query’s engine phases', async () => { + const onEvent = jest.fn(); + + await dbm.queryWithTables({ + query: 'SELECT * FROM table1', + tables: tables, + options: { onEvent }, + }); + + const emittedNames = onEvent.mock.calls.map((call) => call[0].event_name); + expect(emittedNames).toContain('query_execution_duration'); + expect(emittedNames).toContain('mount_file_buffer_duration'); + }); + + it('routes exclusively: query-lifecycle events skip the instance onEvent when a per-query onEvent is supplied', async () => { + const instanceOnEvent = jest.fn(); + const perQueryOnEvent = jest.fn(); + + const dbmWithSink = new DBM({ + instanceManager: new InstanceManager(), + fileManager, + logger: log, + onEvent: instanceOnEvent, + options: { shutdownInactiveTime: 100 }, + onCreateConnection, + }); + + await dbmWithSink.queryWithTables({ + query: 'SELECT * FROM table1', + tables: tables, + options: { onEvent: perQueryOnEvent }, + }); + + const perQueryNames = perQueryOnEvent.mock.calls.map( + (call) => call[0].event_name + ); + expect(perQueryNames).toContain('query_execution_duration'); + const instanceNames = instanceOnEvent.mock.calls.map( + (call) => call[0].event_name + ); + expect(instanceNames).not.toContain('query_execution_duration'); + expect(instanceNames).not.toContain('mount_file_buffer_duration'); + }); + + it('should not invoke a per-query onEvent registered on a different query', async () => { + const onEventA = jest.fn(); + + await dbm.queryWithTables({ + query: 'SELECT * FROM table1', + tables: tables, + options: { onEvent: onEventA }, + }); + onEventA.mockClear(); + + await dbm.queryWithTables({ + query: 'SELECT * FROM table1', + tables: tables, + }); + + expect(onEventA).not.toHaveBeenCalled(); + }); + it('should execute multiple queries with table names', async () => { const promise1 = dbm.queryWithTables({ query: 'SELECT * FROM table1', diff --git a/meerkat-dbm/src/dbm/__test__/runner-manager.spec.ts b/meerkat-dbm/src/dbm/__test__/runner-manager.spec.ts index 0f13c2a8..914e6866 100644 --- a/meerkat-dbm/src/dbm/__test__/runner-manager.spec.ts +++ b/meerkat-dbm/src/dbm/__test__/runner-manager.spec.ts @@ -110,4 +110,109 @@ describe('IFrameRunnerManager', () => { message.message.payload.tables ); }); + + const onEventMessage = (scope?: 'query' | 'instance') => ({ + uuid: 'mock-uuid', + message: { + type: BROWSER_RUNNER_TYPE.RUNNER_ON_EVENT, + payload: { event_name: 'query_execution_duration', duration: 5 }, + scope, + }, + target_app: 'runner', + timestamp: 0, + }); + + it('dispatches a query-scoped RUNNER_ON_EVENT to the callback registered for the emitting runner', () => { + const onEvent = jest.fn(); + manager.registerQueryEventCallback('0', onEvent); + + manager['messageListener']('0', onEventMessage('query') as never); + + expect(onEvent).toHaveBeenCalledWith( + expect.objectContaining({ + event_name: 'query_execution_duration', + duration: 5, + }) + ); + }); + + it('stops dispatching after the per-query callback is unregistered', () => { + const onEvent = jest.fn(); + manager.registerQueryEventCallback('0', onEvent); + manager.unregisterQueryEventCallback('0'); + + manager['messageListener']('0', onEventMessage('query') as never); + + expect(onEvent).not.toHaveBeenCalled(); + }); + + it('does not dispatch to a callback registered for a different runner', () => { + const onEvent = jest.fn(); + manager.registerQueryEventCallback('0', onEvent); + + manager['messageListener']('1', onEventMessage('query') as never); + + expect(onEvent).not.toHaveBeenCalled(); + }); + + it('routes exclusively: the instance onEvent does not fire for a query-scoped event when a per-runner callback is registered', () => { + const instanceOnEvent = jest.fn(); + manager['onEvent'] = instanceOnEvent; + const perQueryOnEvent = jest.fn(); + manager.registerQueryEventCallback('0', perQueryOnEvent); + + manager['messageListener']('0', onEventMessage('query') as never); + + expect(perQueryOnEvent).toHaveBeenCalledTimes(1); + expect(instanceOnEvent).not.toHaveBeenCalled(); + }); + + it('falls back to the instance onEvent when no per-runner callback is registered', () => { + const instanceOnEvent = jest.fn(); + manager['onEvent'] = instanceOnEvent; + + manager['messageListener']('0', onEventMessage('query') as never); + + expect(instanceOnEvent).toHaveBeenCalledWith( + expect.objectContaining({ event_name: 'query_execution_duration' }) + ); + }); + + it('routes an instance-scoped event to the instance sink even while a per-runner callback is registered', () => { + const instanceOnEvent = jest.fn(); + manager['onEvent'] = instanceOnEvent; + const perQueryOnEvent = jest.fn(); + manager.registerQueryEventCallback('0', perQueryOnEvent); + + manager['messageListener']( + '0', + { + uuid: 'mock-uuid', + message: { + type: BROWSER_RUNNER_TYPE.RUNNER_ON_EVENT, + payload: { event_name: 'clone_buffer_duration', duration: 9 }, + scope: 'instance', + }, + target_app: 'runner', + timestamp: 0, + } as never + ); + + expect(instanceOnEvent).toHaveBeenCalledWith( + expect.objectContaining({ event_name: 'clone_buffer_duration' }) + ); + expect(perQueryOnEvent).not.toHaveBeenCalled(); + }); + + it('falls back to the instance sink when a RUNNER_ON_EVENT carries no scope (older runner bundle)', () => { + const instanceOnEvent = jest.fn(); + manager['onEvent'] = instanceOnEvent; + const perQueryOnEvent = jest.fn(); + manager.registerQueryEventCallback('0', perQueryOnEvent); + + manager['messageListener']('0', onEventMessage(undefined) as never); + + expect(instanceOnEvent).toHaveBeenCalledTimes(1); + expect(perQueryOnEvent).not.toHaveBeenCalled(); + }); }); diff --git a/meerkat-dbm/src/dbm/dbm-parallel/dbm-parallel.ts b/meerkat-dbm/src/dbm/dbm-parallel/dbm-parallel.ts index 8ada82b0..f28a184d 100644 --- a/meerkat-dbm/src/dbm/dbm-parallel/dbm-parallel.ts +++ b/meerkat-dbm/src/dbm/dbm-parallel/dbm-parallel.ts @@ -254,6 +254,10 @@ export class DBMParallel extends TableLockManager { // Deduplicate tables by name const tables = uniqBy(_tables, 'name'); + // Hoisted so the finally block can unregister the per-query event callback + // even when the query is aborted (the abort handler clears activeQueries). + let selectedRunnerId: string | undefined; + try { const start = performance.now(); /** @@ -279,9 +283,12 @@ export class DBMParallel extends TableLockManager { const runners = this.iFrameRunnerManager.getRunnerIds(); this.counter = roundRobin(this.counter, runners.length - 1); - const runner = this.iFrameRunnerManager.iFrameManagers.get( - runners[this.counter] - ); + selectedRunnerId = runners[this.counter]; + // Local const so narrowing to `string` survives into the closures below; + // the hoisted `let` is reserved for the finally block. + const runnerId = selectedRunnerId; + + const runner = this.iFrameRunnerManager.iFrameManagers.get(runnerId); /** * StartRunners only initiates the runners, it does not guarantee that the runner is ready to accept the query @@ -294,12 +301,17 @@ export class DBMParallel extends TableLockManager { } this.activeQueries.set(queryId, { - runnerId: runners[this.counter], + runnerId, signal, }); + this.iFrameRunnerManager.registerQueryEventCallback( + runnerId, + options?.onEvent + ); + const abortPromise = new Promise((_, reject) => { - this._signalListener(queryId, runners[this.counter], reject, signal); + this._signalListener(queryId, runnerId, reject, signal); }); /** @@ -320,8 +332,10 @@ export class DBMParallel extends TableLockManager { tables, options: { ...options, - // Don't pass signal to iframe as it's not serializable + // Not serializable across postMessage; onEvent is dispatched + // main-side. signal: undefined, + onEvent: undefined, }, }, } @@ -364,6 +378,13 @@ export class DBMParallel extends TableLockManager { if (queryInfo?.signal && queryInfo.abortHandler) { queryInfo.signal.removeEventListener('abort', queryInfo.abortHandler); } + // Unregister independently of activeQueries: on an aborted query the abort + // handler has already removed the activeQueries entry, so gating this on + // queryInfo would leak the runner's event callback (and misroute trailing + // events). Guarded only against an early throw before a runner was picked. + if (selectedRunnerId !== undefined) { + this.iFrameRunnerManager.unregisterQueryEventCallback(selectedRunnerId); + } this.activeQueries.delete(queryId); /** diff --git a/meerkat-dbm/src/dbm/dbm-parallel/runner-manager.ts b/meerkat-dbm/src/dbm/dbm-parallel/runner-manager.ts index eb3600c8..cac6cead 100644 --- a/meerkat-dbm/src/dbm/dbm-parallel/runner-manager.ts +++ b/meerkat-dbm/src/dbm/dbm-parallel/runner-manager.ts @@ -16,6 +16,13 @@ export interface IFrameRunnerManagerConstructor { fetchPreQuery: (runnerId: string, tables: Table[]) => string[]; totalRunners: number; logger: DBMLogger; + /** + * Instance-level callback for events that are not scoped to a single query + * (e.g. runner-side `clone_buffer_duration`, emitted inside the iframe where + * a per-query callback cannot be reached across `postMessage`). Query- + * lifecycle events are dispatched per-query via + * {@link IFrameRunnerManager.registerQueryEventCallback}. + */ onEvent?: (event: DBMEvent) => void; } @@ -39,6 +46,8 @@ export class IFrameRunnerManager { private runnerURL: string; private logger: DBMLogger; private onEvent?: (event: DBMEvent) => void; + private perRunnerEventCallbacks: Map void> = + new Map(); private fetchTableFileBuffers: ( tables: TableConfig[] @@ -63,6 +72,19 @@ export class IFrameRunnerManager { this.fetchPreQuery = fetchPreQuery; } + public registerQueryEventCallback( + runnerId: string, + onEvent?: (event: DBMEvent) => void + ) { + if (onEvent) { + this.perRunnerEventCallbacks.set(runnerId, onEvent); + } + } + + public unregisterQueryEventCallback(runnerId: string) { + this.perRunnerEventCallbacks.delete(runnerId); + } + private addIFrameManager(uuid: string) { this.iFrameReadyMap.set(uuid, false); this.iFrameManagers.set( @@ -154,11 +176,16 @@ export class IFrameRunnerManager { break; } - case BROWSER_RUNNER_TYPE.RUNNER_ON_EVENT: - if (this.onEvent) { - this.onEvent(message.message.payload); + case BROWSER_RUNNER_TYPE.RUNNER_ON_EVENT: { + const payload = message.message.payload; + const perQueryOnEvent = this.perRunnerEventCallbacks.get(runnerId); + if (message.message.scope === 'query' && perQueryOnEvent) { + perQueryOnEvent(payload); + } else if (this.onEvent) { + this.onEvent(payload); } break; + } case BROWSER_RUNNER_TYPE.RUNNER_PRE_QUERY: { if (this.fetchPreQuery) { diff --git a/meerkat-dbm/src/dbm/dbm.ts b/meerkat-dbm/src/dbm/dbm.ts index 9fe3669d..66b7485a 100644 --- a/meerkat-dbm/src/dbm/dbm.ts +++ b/meerkat-dbm/src/dbm/dbm.ts @@ -243,7 +243,17 @@ export class DBM extends TableLockManager { } } - private _emitEvent(event: DBMEvent) { + private _emitQueryEvent(event: DBMEvent, options?: QueryOptions) { + if (options?.onEvent) { + options.onEvent(event); + return; + } + if (this.onEvent) { + this.onEvent(event); + } + } + + private _emitInstanceEvent(event: DBMEvent) { if (this.onEvent) { this.onEvent(event); } @@ -292,11 +302,14 @@ export class DBM extends TableLockManager { query ); - this._emitEvent({ - event_name: 'mount_file_buffer_duration', - duration: endMountTime - startMountTime, - metadata: options?.metadata, - }); + this._emitQueryEvent( + { + event_name: 'mount_file_buffer_duration', + duration: endMountTime - startMountTime, + metadata: options?.metadata, + }, + options + ); const tablesFileData = await this.fileManager.getFilesNameForTables(tables); @@ -323,11 +336,14 @@ export class DBM extends TableLockManager { query ); - this._emitEvent({ - event_name: 'query_execution_duration', - duration: queryQueueDuration, - metadata: options?.metadata, - }); + this._emitQueryEvent( + { + event_name: 'query_execution_duration', + duration: queryQueueDuration, + metadata: options?.metadata, + }, + options + ); return result; } @@ -353,7 +369,7 @@ export class DBM extends TableLockManager { private async _startQueryExecution(metadata?: object) { this.logger.debug('Query queue length:', this.queriesQueue.length); - this._emitEvent({ + this._emitInstanceEvent({ event_name: 'query_queue_length', value: this.queriesQueue.length, metadata, @@ -388,11 +404,14 @@ export class DBM extends TableLockManager { this.currentQueryItem.query ); - this._emitEvent({ - event_name: 'query_queue_duration', - duration: startTime - this.currentQueryItem.timestamp, - metadata, - }); + this._emitQueryEvent( + { + event_name: 'query_queue_duration', + duration: startTime - this.currentQueryItem.timestamp, + metadata: this.currentQueryItem.options?.metadata ?? metadata, + }, + this.currentQueryItem.options + ); /** * Execute the query diff --git a/meerkat-dbm/src/dbm/types.ts b/meerkat-dbm/src/dbm/types.ts index 986470ef..3cdea6c1 100644 --- a/meerkat-dbm/src/dbm/types.ts +++ b/meerkat-dbm/src/dbm/types.ts @@ -27,7 +27,18 @@ export interface DBMConstructorOptions { /** * @description - * A callback function that handles events emitted by the DBM. + * Instance-level callback for events that are NOT scoped to a single query. + * It receives cross-query and load-time events — `query_queue_length` (a + * queue gauge that spans queries), `json_to_buffer_conversion_duration` + * (emitted at `registerJSON`/load time, outside any `queryWithTables` run), + * and runner-side `clone_buffer_duration` (emitted inside the iframe runner, + * which a per-query callback cannot cross via `postMessage`). + * + * For events that DO belong to one query run — `mount_file_buffer_duration`, + * `query_execution_duration`, `query_queue_duration` — use the per-query + * {@link QueryOptions.onEvent}. Routing is exclusive: when a query supplies + * its own callback those events go there and NOT here; when it does not, they + * fall back to this instance sink. */ onEvent?: (event: DBMEvent) => void; @@ -97,6 +108,26 @@ export interface QueryOptions { */ metadata?: object; + /** + * @description + * Per-query event callback for the query-lifecycle events of THIS query — + * `mount_file_buffer_duration`, `query_execution_duration` and + * `query_queue_duration` — so a caller can scope those timings to a single + * query run without correlating on `metadata`. Routing is exclusive: when + * this callback is supplied those events go here and NOT to the instance-level + * {@link DBMConstructorOptions.onEvent}. + * + * Cross-query and load-time events (`query_queue_length`, + * `json_to_buffer_conversion_duration`, runner-side `clone_buffer_duration`) + * are NOT delivered here — they have no single owning query and reach only the + * instance sink. + * + * For the parallel/iframe path the callback stays in the calling window; the + * runner manager dispatches to it by the id of the runner executing the query + * (one query per runner), so no query id crosses `postMessage`. + */ + onEvent?: (event: DBMEvent) => void; + /** * @description * An AbortSignal object instance which can be used to abort the query execution. diff --git a/meerkat-dbm/src/logger/event-types.ts b/meerkat-dbm/src/logger/event-types.ts index ef8e30c0..68aac881 100644 --- a/meerkat-dbm/src/logger/event-types.ts +++ b/meerkat-dbm/src/logger/event-types.ts @@ -2,7 +2,6 @@ export interface DurationEvents { event_name: | 'query_execution_duration' | 'mount_file_buffer_duration' - | 'unmount_file_buffer_duration' | 'query_queue_duration' | 'json_to_buffer_conversion_duration' | 'clone_buffer_duration'; diff --git a/meerkat-dbm/src/window-communication/runner-types.ts b/meerkat-dbm/src/window-communication/runner-types.ts index 98005b76..1c1ccb2e 100644 --- a/meerkat-dbm/src/window-communication/runner-types.ts +++ b/meerkat-dbm/src/window-communication/runner-types.ts @@ -56,6 +56,13 @@ export interface BrowserRunnerPreQueryMessage { export interface BrowserRunnerOnEventMessage { type: typeof BROWSER_RUNNER_TYPE.RUNNER_ON_EVENT; payload: DBMEvent; + /** + * Routing scope decided at the emit site inside the runner. `'query'` events + * belong to the runner's in-flight query and go to that query's per-query + * callback; anything else (or omitted) goes to the instance sink. Optional so + * an older runner bundle that doesn't set it falls back to the instance sink. + */ + scope?: 'query' | 'instance'; } export interface BrowserRunnerCancelQueryMessage { diff --git a/meerkat-node/package.json b/meerkat-node/package.json index 630b4e94..9cea0ac3 100644 --- a/meerkat-node/package.json +++ b/meerkat-node/package.json @@ -1,6 +1,6 @@ { "name": "@devrev/meerkat-node", - "version": "0.0.133", + "version": "0.0.134", "dependencies": { "@devrev/meerkat-core": "*", "@swc/helpers": "~0.5.0", diff --git a/meerkat-node/src/cube-to-sql/cube-to-sql.ts b/meerkat-node/src/cube-to-sql/cube-to-sql.ts index c7aa2494..14411451 100644 --- a/meerkat-node/src/cube-to-sql/cube-to-sql.ts +++ b/meerkat-node/src/cube-to-sql/cube-to-sql.ts @@ -2,6 +2,7 @@ import { BASE_TABLE_NAME, ContextParams, Query, + SqlGenerationOnEvent, TableSchema, applyFilterParamsToBaseSQL, applyProjectionToSQLQuery, @@ -13,6 +14,7 @@ import { getCombinedTableSchema, getFilterParamsSQL, getFinalBaseSQL, + timePhase, } from '@devrev/meerkat-core'; import { duckdbExec } from '../duckdb-exec'; @@ -20,85 +22,128 @@ export interface CubeQueryToSQLParams { query: Query; tableSchemas: TableSchema[]; contextParams?: ContextParams; + /** + * Optional per-call callback for SQL-generation phase timings. Invoked once + * per phase plus once for `total`. No-op when omitted. + */ + onEvent?: SqlGenerationOnEvent; } export const cubeQueryToSQL = async ({ query, tableSchemas, contextParams, + onEvent, }: CubeQueryToSQLParams) => { - const updatedTableSchemas: TableSchema[] = await Promise.all( - tableSchemas.map(async (schema: TableSchema) => { - const baseFilterParamsSQL = await getFinalBaseSQL({ - query, - tableSchema: schema, - getQueryOutput: duckdbExec, - }); - return { - ...schema, - sql: baseFilterParamsSQL, - }; - }) + const totalStart = performance.now(); + + const updatedTableSchemas = await timePhase( + 'base_sql', + () => + Promise.all( + tableSchemas.map(async (schema: TableSchema) => { + const baseFilterParamsSQL = await getFinalBaseSQL({ + query, + tableSchema: schema, + getQueryOutput: duckdbExec, + }); + return { + ...schema, + sql: baseFilterParamsSQL, + }; + }) + ), + onEvent ); const updatedTableSchema = getCombinedTableSchema(updatedTableSchemas, query); - const ast = cubeToDuckdbAST(query, updatedTableSchema, { - filterType: 'PROJECTION_FILTER', - }); + const ast = await timePhase( + 'ast_build', + async () => + cubeToDuckdbAST(query, updatedTableSchema, { + filterType: 'PROJECTION_FILTER', + }), + onEvent + ); if (!ast) { throw new Error('Could not generate AST'); } - const queryTemp = astDeserializerQuery(ast); + const preBaseQuery = await timePhase( + 'ast_deserialize_roundtrip', + async () => { + const queryTemp = astDeserializerQuery(ast); + const queryOutput = (await duckdbExec(queryTemp)) as Record< + string, + string + >[]; + return deserializeQuery(queryOutput); + }, + onEvent + ); - const queryOutput = (await duckdbExec(queryTemp)) as Record[]; - const preBaseQuery = deserializeQuery(queryOutput); + const filterParamsSQL = await timePhase( + 'filter_params', + () => + getFilterParamsSQL({ + query, + tableSchema: updatedTableSchema, + filterType: 'PROJECTION_FILTER', + getQueryOutput: duckdbExec, + }), + onEvent + ); - const filterParamsSQL = await getFilterParamsSQL({ - query, - tableSchema: updatedTableSchema, - filterType: 'PROJECTION_FILTER', - getQueryOutput: duckdbExec, - }); + const finalQuery = await timePhase( + 'projections', + async () => { + const filterParamQuery = applyFilterParamsToBaseSQL( + updatedTableSchema.sql, + filterParamsSQL + ); - const filterParamQuery = applyFilterParamsToBaseSQL( - updatedTableSchema.sql, - filterParamsSQL - ); + /** + * Replace CONTEXT_PARAMS with context params + */ + const baseQuery = detectApplyContextParamsToBaseSQL( + filterParamQuery, + contextParams || {} + ); - /** - * Replace CONTEXT_PARAMS with context params - */ - const baseQuery = detectApplyContextParamsToBaseSQL( - filterParamQuery, - contextParams || {} - ); + /** + * Replace BASE_TABLE_NAME with cube query + */ + const replaceBaseTableName = preBaseQuery.replace( + BASE_TABLE_NAME, + `(${baseQuery}) AS ${updatedTableSchema.name}` + ); - /** - * Replace BASE_TABLE_NAME with cube query - */ - const replaceBaseTableName = preBaseQuery.replace( - BASE_TABLE_NAME, - `(${baseQuery}) AS ${updatedTableSchema.name}` - ); + /** + * Add measures to the query + */ + const measures = query.measures; + const dimensions = query.dimensions || []; + const queryWithProjections = applyProjectionToSQLQuery( + dimensions, + measures, + updatedTableSchemas, + replaceBaseTableName + ); - /** - * Add measures to the query - */ - const measures = query.measures; - const dimensions = query.dimensions || []; - const queryWithProjections = applyProjectionToSQLQuery( - dimensions, - measures, - updatedTableSchemas, - replaceBaseTableName + /** + * Replace SQL expression placeholders with actual SQL + */ + return applySQLExpressions(queryWithProjections); + }, + onEvent ); - /** - * Replace SQL expression placeholders with actual SQL - */ - const finalQuery = applySQLExpressions(queryWithProjections); + onEvent?.({ + event_name: 'sql_generation_duration', + phase: 'total', + duration: performance.now() - totalStart, + }); return finalQuery; };