diff --git a/dev/index.html b/dev/index.html index 71bcd4acb..280643511 100644 --- a/dev/index.html +++ b/dev/index.html @@ -113,7 +113,7 @@
diff --git a/dev/setup.ts b/dev/setup.ts index baa8cb71c..a06a5acb8 100644 --- a/dev/setup.ts +++ b/dev/setup.ts @@ -1,7 +1,8 @@ -import { DuckDBWASMConnector, RestConnector, SocketConnector } from '@uwdata/mosaic-core'; +import { DuckDBWASMConnector, observeLogger, RestConnector, SocketConnector } from '@uwdata/mosaic-core'; import { createAPIContext } from '@uwdata/vgplot'; export { parseSpec, astToDOM, astToESM } from '@uwdata/mosaic-spec'; +export { observeLogger }; export const vg = createAPIContext(); // make API accessible for console debugging diff --git a/packages/mosaic/core/src/Coordinator.ts b/packages/mosaic/core/src/Coordinator.ts index 7c66d285e..f2042e296 100644 --- a/packages/mosaic/core/src/Coordinator.ts +++ b/packages/mosaic/core/src/Coordinator.ts @@ -1,8 +1,11 @@ /* eslint-disable @typescript-eslint/no-explicit-any */ import { SocketConnector } from './connectors/socket.js'; import { type Connector } from './connectors/Connector.js'; -import { PreAggregator, type PreAggregateInfo, type PreAggregateOptions } from './preagg/PreAggregator.js'; -import { voidLogger } from './util/void-logger.js'; +import { + PreAggregator, + type PreAggregateInfo, + type PreAggregateOptions, +} from './preagg/PreAggregator.js'; import { QueryManager, Priority } from './QueryManager.js'; import { type Selection } from './Selection.js'; import { type Logger, type QueryType } from './types.js'; @@ -11,6 +14,9 @@ import { type MosaicClient } from './MosaicClient.js'; import { type SelectionClause } from './SelectionClause.js'; import { MaybeArray } from '@uwdata/mosaic-sql'; import { Table } from '@uwdata/flechette'; +import { EventType, MosaicErrorEvent, type MosaicEventMap } from './Events.js'; +import { ObserveDispatch } from './util/ObserveDispatch.js'; +import { observeLogger } from './logger.js'; interface FilterGroupEntry { selection: Selection; @@ -28,9 +34,7 @@ let _instance: Coordinator; * @param instance The coordinator instance to set * @returns The coordinator instance */ -export function coordinator( - instance?: Coordinator -): Coordinator { +export function coordinator(instance?: Coordinator): Coordinator { if (instance) { _instance = instance; } else if (_instance == null) { @@ -47,14 +51,16 @@ export function coordinator( export class Coordinator { public manager: QueryManager; public preaggregator: PreAggregator; - public clients = new Set; - public filterGroups = new Map; - protected _logger: Logger = voidLogger(); + public clients = new Set(); + public filterGroups = new Map(); + public readonly eventBus: ObserveDispatch; + protected _logger: Logger | null = null; + private _unobserveLogger: () => void = () => {}; /** * @param db Database connector. Defaults to a web socket connection. * @param options Coordinator options. - * @param options.logger The logger to use, defaults to `console`. + * @param options.logger The logger to use, defaults to `console`. Pass `null` to disable logging. * @param options.manager The query manager to use. * @param options.cache Boolean flag to enable/disable query caching. * @param options.consolidate Boolean flag to enable/disable query consolidation. @@ -68,16 +74,18 @@ export class Coordinator { cache?: boolean; consolidate?: boolean; preagg?: PreAggregateOptions; - } = {} + } = {}, ) { const { logger = console, manager = new QueryManager(), cache = true, consolidate = true, - preagg = {} + preagg = {}, } = options; + this.eventBus = new ObserveDispatch(); this.manager = manager; + this.manager.attachEventBus(this.eventBus); this.manager.cache(cache); this.manager.consolidate(consolidate); this.databaseConnector(db); @@ -96,10 +104,10 @@ export class Coordinator { const { clients = true, cache = true } = options; this.manager.clear(); if (clients) { - this.filterGroups?.forEach(group => group.disconnect()); - this.filterGroups = new Map; - this.clients?.forEach(client => this.disconnect(client)); - this.clients = new Set; + this.filterGroups?.forEach((group) => group.disconnect()); + this.filterGroups = new Map(); + this.clients?.forEach((client) => this.disconnect(client)); + this.clients = new Set(); } if (cache) this.manager.cache()!.clear(); } @@ -112,22 +120,25 @@ export class Coordinator { databaseConnector(): Connector | null; databaseConnector(db: Connector): Connector; databaseConnector(db?: Connector): Connector | null { - return db - ? this.manager.connector(db) - : this.manager.connector(); + return db ? this.manager.connector(db) : this.manager.connector(); } /** * Get or set the logger. - * @param logger The logger to use. - * @returns The current logger + * + * @deprecated Prefer observeLogger(coordinator, logger) for event-bus based logging. + * @param logger The logger to use. Pass `null` to disable logging. + * @returns The current logger. */ - logger(logger?: Logger | null): Logger { + logger(): Logger | null; + logger(logger: Logger | null): Logger | null; + logger(logger?: Logger | null): Logger | null { if (arguments.length) { - this._logger = logger || voidLogger(); - this.manager.logger(this._logger); + this._logger = logger ?? null; + this._unobserveLogger(); + this._unobserveLogger = observeLogger(this, this._logger); } - return this._logger!; + return this._logger; } // -- Query Management ---- @@ -150,7 +161,7 @@ export class Coordinator { */ exec( query: MaybeArray, - options: { priority?: number } = {} + options: { priority?: number } = {}, ): QueryResult { const { priority = Priority.Normal } = options; return this.manager.request({ type: 'exec', query }, priority); @@ -175,7 +186,7 @@ export class Coordinator { persist?: boolean; priority?: number; [key: string]: unknown; - } + }, ): QueryResult; query( query: QueryType, @@ -185,7 +196,7 @@ export class Coordinator { persist?: boolean; priority?: number; [key: string]: unknown; - } + }, ): QueryResult; query( query: QueryType, @@ -195,7 +206,7 @@ export class Coordinator { persist?: boolean; priority?: number; [key: string]: unknown; - } = {} + } = {}, ): QueryResult { const { type = 'arrow', @@ -203,7 +214,10 @@ export class Coordinator { priority = Priority.Normal, ...otherOptions } = options; - return this.manager.request({ type, query, cache, options: otherOptions }, priority); + return this.manager.request( + { type, query, cache, options: otherOptions }, + priority, + ); } /** @@ -216,17 +230,18 @@ export class Coordinator { */ prefetch( query: QueryType, - options?: { type?: 'arrow'; [key: string]: unknown } - ): QueryResult
- prefetch( - query: QueryType, - options?: { type?: 'json'; [key: string]: unknown } - ): QueryResult + options?: { type?: 'arrow'; [key: string]: unknown }, + ): QueryResult
; prefetch( query: QueryType, - options: any = {} - ): QueryResult { - return this.query(query, { ...options, cache: true, priority: Priority.Low }); + options?: { type?: 'json'; [key: string]: unknown }, + ): QueryResult; + prefetch(query: QueryType, options: any = {}): QueryResult { + return this.query(query, { + ...options, + cache: true, + priority: Priority.Low, + }); } // -- Client Management ---- @@ -242,15 +257,32 @@ export class Coordinator { updateClient( client: MosaicClient, query: QueryType, - priority: number = Priority.Normal + priority: number = Priority.Normal, ): Promise { client.queryPending(); - return client._pending = this.query(query, { priority }) + return (client._pending = this.query(query, { priority }) .then( - data => client.queryResult(data).update(), - err => { this._logger?.error(err); client.queryError(err); } + (data) => client.queryResult(data).update(), + (err) => { + this.eventBus.emit( + EventType.Error, + new MosaicErrorEvent({ + message: err instanceof Error ? err.message : String(err), + error: err, + }), + ); + client.queryError(err); + }, ) - .catch(err => this._logger?.error(err)); + .catch((err) => { + this.eventBus.emit( + EventType.Error, + new MosaicErrorEvent({ + message: err instanceof Error ? err.message : String(err), + error: err, + }), + ); + })); } /** @@ -260,7 +292,10 @@ export class Coordinator { * @param client The client to update. * @param query The query to issue. */ - requestQuery(client: MosaicClient, query?: QueryType | null): Promise { + requestQuery( + client: MosaicClient, + query?: QueryType | null, + ): Promise { this.preaggregator.clear(); return query ? this.updateClient(client, query) @@ -319,12 +354,13 @@ export class Coordinator { function connectSelection( mc: Coordinator, selection: Selection, - client: MosaicClient + client: MosaicClient, ): void { if (!selection) return; let entry = mc.filterGroups?.get(selection); if (!entry) { - const activate = (clause: SelectionClause) => activateSelection(mc, selection, clause); + const activate = (clause: SelectionClause) => + activateSelection(mc, selection, clause); const value = () => updateSelection(mc, selection); // @ts-expect-error todo: update selection dispatch types @@ -333,12 +369,12 @@ function connectSelection( entry = { selection, - clients: new Set, + clients: new Set(), disconnect() { // @ts-expect-error todo: update selection dispatch types selection.removeEventListener('activate', activate); selection.removeEventListener('value', value); - } + }, }; mc.filterGroups?.set(selection, entry); } @@ -356,7 +392,7 @@ function connectSelection( function activateSelection( mc: Coordinator, selection: Selection, - clause: SelectionClause + clause: SelectionClause, ): void { const { preaggregator, filterGroups } = mc; const { clients } = filterGroups.get(selection)!; @@ -376,27 +412,31 @@ function activateSelection( */ function updateSelection( mc: Coordinator, - selection: Selection + selection: Selection, ): Promise[]> { const { preaggregator, filterGroups } = mc; const { clients } = filterGroups!.get(selection)!; const { active } = selection; - return Promise.allSettled(Array.from(clients, async (client: MosaicClient) => { - // if client is not enabled, register a request for later - if (!client.enabled) return client.requestQuery(); - - // if client is initializing, wait for it to complete - if (!client.initialized) await client.pending; - - // check if we can handle selection update via preaggregation - const info = preaggregator.request(client, selection, active); - const filter = info ? null : selection.predicate(client); - - // skip due to cross-filtering - if (info?.skip || (!info && !filter)) return; - - // generate and issue update query - const query = (info as PreAggregateInfo)?.query(active.predicate!) ?? client.query(filter); - return mc.updateClient(client, query); - })); + return Promise.allSettled( + Array.from(clients, async (client: MosaicClient) => { + // if client is not enabled, register a request for later + if (!client.enabled) return client.requestQuery(); + + // if client is initializing, wait for it to complete + if (!client.initialized) await client.pending; + + // check if we can handle selection update via preaggregation + const info = preaggregator.request(client, selection, active); + const filter = info ? null : selection.predicate(client); + + // skip due to cross-filtering + if (info?.skip || (!info && !filter)) return; + + // generate and issue update query + const query = + (info as PreAggregateInfo)?.query(active.predicate!) ?? + client.query(filter); + return mc.updateClient(client, query); + }), + ); } diff --git a/packages/mosaic/core/src/Events.ts b/packages/mosaic/core/src/Events.ts new file mode 100644 index 000000000..63152fc0b --- /dev/null +++ b/packages/mosaic/core/src/Events.ts @@ -0,0 +1,104 @@ +export enum EventType { + QueryStart = 'query-start', + QueryEnd = 'query-end', + Warning = 'warning', + Error = 'error', +} + +abstract class MosaicEventBase { + readonly type: T; + readonly timestamp: number; + + protected constructor(type: T, timestamp: number = Date.now()) { + this.type = type; + this.timestamp = timestamp; + } +} + +export interface QueryLifecycleEventInit { + queryId: number; + query: string; + materialized: boolean; + timestamp?: number; +} + +export type QueryEndStatus = 'success' | 'error'; + +export interface QueryEndEventInit extends QueryLifecycleEventInit { + status: QueryEndStatus; +} + +export class MosaicQueryStartEvent extends MosaicEventBase { + readonly queryId: number; + readonly query: string; + readonly materialized: boolean; + + constructor({ queryId, query, materialized, timestamp }: QueryLifecycleEventInit) { + super(EventType.QueryStart, timestamp); + this.queryId = queryId; + this.query = query; + this.materialized = materialized; + } +} + +export class MosaicQueryEndEvent extends MosaicEventBase { + readonly queryId: number; + readonly query: string; + readonly materialized: boolean; + readonly status: QueryEndStatus; + + constructor({ queryId, query, materialized, status, timestamp }: QueryEndEventInit) { + super(EventType.QueryEnd, timestamp); + this.queryId = queryId; + this.query = query; + this.materialized = materialized; + this.status = status; + } +} + +export interface MosaicMessageEventInit { + message: string; + queryId?: number; + timestamp?: number; +} + +export interface MosaicErrorEventInit extends MosaicMessageEventInit { + error?: unknown; +} + +export class MosaicWarningEvent extends MosaicEventBase { + readonly message: string; + readonly queryId?: number; + + constructor({ message, queryId, timestamp }: MosaicMessageEventInit) { + super(EventType.Warning, timestamp); + this.message = message; + this.queryId = queryId; + } +} + +export class MosaicErrorEvent extends MosaicEventBase { + readonly message: string; + readonly queryId?: number; + readonly error?: unknown; + + constructor({ message, queryId, error, timestamp }: MosaicErrorEventInit) { + super(EventType.Error, timestamp); + this.message = message; + this.queryId = queryId; + this.error = error; + } +} + +export type MosaicEvent = + | MosaicQueryStartEvent + | MosaicQueryEndEvent + | MosaicWarningEvent + | MosaicErrorEvent; + +export type MosaicEventMap = { + [EventType.QueryStart]: MosaicQueryStartEvent; + [EventType.QueryEnd]: MosaicQueryEndEvent; + [EventType.Warning]: MosaicWarningEvent; + [EventType.Error]: MosaicErrorEvent; +}; diff --git a/packages/mosaic/core/src/QueryManager.ts b/packages/mosaic/core/src/QueryManager.ts index 776fbcd29..da137dbb6 100644 --- a/packages/mosaic/core/src/QueryManager.ts +++ b/packages/mosaic/core/src/QueryManager.ts @@ -1,10 +1,18 @@ import type { Connector } from './connectors/Connector.js'; -import type { Cache, Logger, QueryEntry, QueryRequest } from './types.js'; +import type { Cache, QueryEntry, QueryRequest } from './types.js'; import { consolidator } from './QueryConsolidator.js'; import { lruCache, voidCache } from './util/cache.js'; import { PriorityQueue } from './util/priority-queue.js'; import { QueryResult, QueryState } from './util/query-result.js'; -import { voidLogger } from './util/void-logger.js'; +import { + EventType, + MosaicErrorEvent, + type MosaicEventMap, + MosaicQueryEndEvent, + MosaicQueryStartEvent, + MosaicWarningEvent, +} from './Events.js'; +import { ObserveDispatch } from './util/ObserveDispatch.js'; export const Priority = Object.freeze({ High: 0, Normal: 1, Low: 2 }); @@ -12,28 +20,44 @@ export class QueryManager { private queue: PriorityQueue; private db: Connector | null; private clientCache: Cache | null; - private _logger: Logger; - private _logQueries: boolean; private _consolidate: ReturnType | null; /** Requests pending with the query manager. */ public pendingResults: QueryResult[]; private maxConcurrentRequests: number; private pendingExec: boolean; + private _eventBus: ObserveDispatch | null; + private _nextQueryId: number; constructor(maxConcurrentRequests: number = 32) { this.queue = new PriorityQueue(3); this.db = null; this.clientCache = null; - this._logger = voidLogger(); - this._logQueries = false; this._consolidate = null; this.pendingResults = []; this.maxConcurrentRequests = maxConcurrentRequests; this.pendingExec = false; + this._eventBus = null; + this._nextQueryId = 1; + } + + /** + * @internal Attach query lifecycle events to the owning coordinator's bus. + * Query managers do not own their own observable event surface; the + * coordinator attaches its bus once during construction. + */ + attachEventBus(eventBus: ObserveDispatch): void { + if (this._eventBus && this._eventBus !== eventBus) { + throw new Error('QueryManager event bus is already attached.'); + } + this._eventBus = eventBus; } next(): void { - if (this.queue.isEmpty() || this.pendingResults.length > this.maxConcurrentRequests || this.pendingExec) { + if ( + this.queue.isEmpty() || + this.pendingResults.length > this.maxConcurrentRequests || + this.pendingExec + ) { return; } @@ -47,12 +71,20 @@ export class QueryManager { this.submit(request, result).finally(() => { // return from the queue all requests that are ready - while (this.pendingResults.length && this.pendingResults[0].state !== QueryState.pending) { + while ( + this.pendingResults.length && + this.pendingResults[0].state !== QueryState.pending + ) { const result = this.pendingResults.shift()!; if (result.state === QueryState.ready) { result.fulfill(); } else if (result.state === QueryState.done) { - this._logger.warn('Found resolved query in pending results.'); + this._eventBus?.emit( + EventType.Warning, + new MosaicWarningEvent({ + message: 'Found resolved query in pending results.', + }), + ); } } if (request.type === 'exec') this.pendingExec = false; @@ -76,27 +108,49 @@ export class QueryManager { * @param result The query result. */ async submit(request: QueryRequest, result: QueryResult): Promise { + let queryId: number | undefined; + let queryText = ''; + let materialized = false; + try { const { query, type, cache = false, options } = request; - const sql = Array.isArray(query) ? query.filter(x => x).join(';\n') : query ? String(query) : null; + const sql = Array.isArray(query) + ? query.filter((x) => x).join(';\n') + : query + ? String(query) + : null; + queryId = this._nextQueryId++; + queryText = sql || ''; + materialized = cache; + + this._eventBus?.emit( + EventType.QueryStart, + new MosaicQueryStartEvent({ + queryId, + query: queryText, + materialized, + }), + ); // check query cache if (cache) { const cached = this.clientCache!.get(sql!); if (cached) { const data = await cached; - this._logger.debug('Cache'); result.ready(data); + this._eventBus?.emit( + EventType.QueryEnd, + new MosaicQueryEndEvent({ + queryId, + query: queryText, + materialized, + status: 'success', + }), + ); return; } } - // issue query, potentially cache result - const t0 = performance.now(); - if (this._logQueries) { - this._logger.debug('Query', { type, sql, ...options }); - } - // @ts-expect-error type may be exec | json | arrow const promise = this.db!.query({ type, sql: sql!, ...options }); if (cache) this.clientCache!.set(sql!, promise); @@ -105,9 +159,37 @@ export class QueryManager { if (cache) this.clientCache!.set(sql!, data); - this._logger.debug(`Request: ${(performance.now() - t0).toFixed(1)}`); result.ready(type === 'exec' ? null : data); + + this._eventBus?.emit( + EventType.QueryEnd, + new MosaicQueryEndEvent({ + queryId, + query: queryText, + materialized, + status: 'success', + }), + ); } catch (err) { + this._eventBus?.emit( + EventType.Error, + new MosaicErrorEvent({ + message: err instanceof Error ? err.message : String(err), + error: err, + queryId, + }), + ); + if (queryId != null) { + this._eventBus?.emit( + EventType.QueryEnd, + new MosaicQueryEndEvent({ + queryId, + query: queryText, + materialized, + status: 'error', + }), + ); + } result.reject(err); } } @@ -121,32 +203,10 @@ export class QueryManager { cache(value: Cache | boolean): Cache; cache(value?: Cache | boolean): Cache | null { return value !== undefined - ? (this.clientCache = value === true ? lruCache() : (value || voidCache())) + ? (this.clientCache = value === true ? lruCache() : value || voidCache()) : this.clientCache; } - /** - * Get or set the current logger. - * @param value Logger to set - * @returns Current logger - */ - logger(): Logger; - logger(value: Logger): Logger; - logger(value?: Logger): Logger { - return value ? (this._logger = value) : this._logger; - } - - /** - * Get or set if queries should be logged. - * @param value Whether to log queries - * @returns Current logging state - */ - logQueries(): boolean; - logQueries(value: boolean): boolean; - logQueries(value?: boolean): boolean { - return value !== undefined ? this._logQueries = !!value : this._logQueries; - } - /** * Get or set the database connector. * @param connector Connector to set @@ -164,7 +224,10 @@ export class QueryManager { */ consolidate(flag: boolean): void { if (flag && !this._consolidate) { - this._consolidate = consolidator(this.enqueue.bind(this), this.clientCache!); + this._consolidate = consolidator( + this.enqueue.bind(this), + this.clientCache!, + ); } else if (!flag && this._consolidate) { this._consolidate = null; } diff --git a/packages/mosaic/core/src/Selection.ts b/packages/mosaic/core/src/Selection.ts index c167ccee1..0704fdbce 100644 --- a/packages/mosaic/core/src/Selection.ts +++ b/packages/mosaic/core/src/Selection.ts @@ -244,7 +244,6 @@ export class Selection extends Param { * @returns A dispatch queue filter function. For non-value events, * returns a function that always returns null (no filtering). */ - // @ts-expect-error selection and param use differing value types emitQueueFilter( type: string, value: SelectionClauseArray diff --git a/packages/mosaic/core/src/index.ts b/packages/mosaic/core/src/index.ts index db1244cb6..fb88f0c8d 100644 --- a/packages/mosaic/core/src/index.ts +++ b/packages/mosaic/core/src/index.ts @@ -1,6 +1,14 @@ export { MosaicClient } from './MosaicClient.js'; export { makeClient } from './make-client.js'; export { Coordinator, coordinator } from './Coordinator.js'; +export { observeLogger } from './logger.js'; +export { + EventType, + MosaicErrorEvent, + MosaicQueryEndEvent, + MosaicQueryStartEvent, + MosaicWarningEvent, +} from './Events.js'; export { Selection, isSelection } from './Selection.js'; export { Param, isParam } from './Param.js'; export { Priority } from './QueryManager.js'; @@ -38,5 +46,12 @@ export type { QueryResult } from './util/query-result.js'; export * from './types.js'; export type * from './connectors/Connector.js'; +export type { + MosaicEvent, + MosaicErrorEventInit, + MosaicEventMap, + MosaicMessageEventInit, + QueryLifecycleEventInit +} from './Events.js'; export type * from './Selection.js'; export type * from './SelectionClause.js'; diff --git a/packages/mosaic/core/src/logger.ts b/packages/mosaic/core/src/logger.ts new file mode 100644 index 000000000..eb572e826 --- /dev/null +++ b/packages/mosaic/core/src/logger.ts @@ -0,0 +1,69 @@ +import type { Coordinator } from './Coordinator.js'; +import type { Logger } from './types.js'; +import { + EventType, + type MosaicErrorEvent, + type MosaicQueryEndEvent, + type MosaicQueryStartEvent, + type MosaicWarningEvent, +} from './Events.js'; + +/** + * Observe coordinator events and log them in a way that mirrors prior + * coordinator-internal logging behavior. + * + * @param coordinator The coordinator to observe. + * @param logger The logger to use. Pass `null` to disable logging. + * @returns Unsubscribe function. + */ +export function observeLogger( + coordinator: Coordinator, + logger: Logger | null = console, +): () => void { + if (!logger) return () => {}; + + const starts = new Map(); + + const onQueryStart = (event: MosaicQueryStartEvent): void => { + starts.set(event.queryId, event.timestamp); + + logger.groupCollapsed(`query ${event.query}`); + }; + + const onQueryEnd = (event: MosaicQueryEndEvent): void => { + const t0 = starts.get(event.queryId); + const openedGroup = t0 != null; + starts.delete(event.queryId); + const elapsed = t0 == null ? undefined : (event.timestamp - t0).toFixed(1); + + if (elapsed != null) { + logger.log(event.query, elapsed); + } else { + // Surface unmatched terminal events without closing a group this observer did not open. + logger.log(event.query); + } + + if (openedGroup) logger.groupEnd(); + }; + + const onWarning = (event: MosaicWarningEvent): void => { + logger.warn(event.message); + }; + + const onError = (event: MosaicErrorEvent): void => { + logger.error(event.error ?? event.message); + }; + + coordinator.eventBus.addEventListener(EventType.QueryStart, onQueryStart); + coordinator.eventBus.addEventListener(EventType.QueryEnd, onQueryEnd); + coordinator.eventBus.addEventListener(EventType.Warning, onWarning); + coordinator.eventBus.addEventListener(EventType.Error, onError); + + return () => { + coordinator.eventBus.removeEventListener(EventType.QueryStart, onQueryStart); + coordinator.eventBus.removeEventListener(EventType.QueryEnd, onQueryEnd); + coordinator.eventBus.removeEventListener(EventType.Warning, onWarning); + coordinator.eventBus.removeEventListener(EventType.Error, onError); + starts.clear(); + }; +} diff --git a/packages/mosaic/core/src/preagg/PreAggregator.ts b/packages/mosaic/core/src/preagg/PreAggregator.ts index 9aa5bb1f8..dd8e76a26 100644 --- a/packages/mosaic/core/src/preagg/PreAggregator.ts +++ b/packages/mosaic/core/src/preagg/PreAggregator.ts @@ -5,6 +5,7 @@ import type { Selection } from '../Selection.js'; import type { BinMethod, ClauseSource, IntervalMetadata, SelectionClause } from '../SelectionClause.js'; import { fnv_hash } from '../util/hash.js'; import { preaggColumns, PreAggColumnsResult } from './preagg-columns.js'; +import { EventType, MosaicErrorEvent } from '../Events.js'; const Skip = { skip: true, result: null }; @@ -61,10 +62,10 @@ export class PreAggregator { * @param coordinator A Mosaic coordinator. * @param options Pre-aggregation options. */ - constructor(coordinator: Coordinator, { - schema = 'mosaic', - enabled = true - }: PreAggregateOptions = {}) { + constructor( + coordinator: Coordinator, + { schema = "mosaic", enabled = true }: PreAggregateOptions = {}, + ) { this.entries = new Map(); this.active = null; this.mc = coordinator; @@ -215,7 +216,12 @@ export class PreAggregator { createSchema(schema), createTable(info.table, info.create, { temp: false }) ]); - info.result.catch((e: Error) => mc.logger().error(e)); + info.result.catch((e: Error) => + mc.eventBus.emit( + EventType.Error, + new MosaicErrorEvent({ message: e.message, error: e }), + ), + ); } entries.set(client, info); @@ -267,15 +273,19 @@ function activeColumns(clause: SelectionClause): ActiveColumnsResult { } else { // selection clause predicate has type AndNode // multiple interval selection - predicate = (p?: ExprNode) => p - ? and((p as AndNode).clauses.map( - (c, i) => isBetween(`active${i}`, c.extent?.map(bins[i]!)) - )) - : []; + predicate = (p?: ExprNode) => + p + ? and( + (p as AndNode).clauses.map((c, i) => + isBetween(`active${i}`, c.extent?.map(bins[i]!)), + ), + ) + : []; columns = Object.fromEntries( - (clausePred as AndNode).clauses.map( - (p, i) => [`active${i}`, bins[i]!(p.expr)] - ) + (clausePred as AndNode).clauses.map((p, i) => [ + `active${i}`, + bins[i]!(p.expr), + ]), ); } } @@ -298,24 +308,22 @@ const BIN: Record FunctionNode> = { ceil, round }; function binInterval( scale: ScaleOptions, pixelSize: number, - bin?: BinMethod + bin?: BinMethod, ): ((value: ExprValue) => ExprNode) | undefined { const { type, domain, range, apply, sqlApply } = scaleTransform(scale)!; if (!apply) return; // unsupported scale type const binFn = BIN[`${bin}`.toLowerCase()] || floor; - const dom = domain!.map(x => Number(x)); + const dom = domain!.map((x) => Number(x)); const lo = apply(Math.min(...dom)); const hi = apply(Math.max(...dom)); - const s = (type === 'identity' - ? 1 - : Math.abs(range![1] - range![0]) / (hi - lo)) / pixelSize; - const scalar = s === 1 - ? (x: ExprValue) => x - : (x: ExprValue) => mul(float64(s), x); - const diff = lo === 0 - ? (x: ExprValue) => x - : (x: ExprValue) => sub(x, float64(lo)); - return value => int32(binFn(scalar(diff(sqlApply(value))))); + const s = + (type === "identity" ? 1 : Math.abs(range![1] - range![0]) / (hi - lo)) / + pixelSize; + const scalar = + s === 1 ? (x: ExprValue) => x : (x: ExprValue) => mul(float64(s), x); + const diff = + lo === 0 ? (x: ExprValue) => x : (x: ExprValue) => sub(x, float64(lo)); + return (value) => int32(binFn(scalar(diff(sqlApply(value))))); } /** @@ -330,7 +338,7 @@ function preaggregateInfo( query: SelectQuery, active: ActiveColumnsResult, preaggCols: PreAggColumnsResult, - schema: string + schema: string, ): PreAggregateInfo { const { dims, groupby, output, preagg } = preaggCols; const { columns = {} } = active; @@ -419,9 +427,11 @@ function subqueryPushdown(query: Query, cols: string[]): void { // if an aggregation query, we need to push to groupby as well // we also deduplicate as the column may already be present const set = new Set( - q._groupby.flatMap(x => x instanceof ColumnNameRefNode ? [x.name] : []) + q._groupby.flatMap((x) => + x instanceof ColumnNameRefNode ? [x.name] : [], + ), ); - q.groupby(cols.filter(c => !set.has(c))); + q.groupby(cols.filter((c) => !set.has(c))); } } q.subqueries.forEach(pushdown); diff --git a/packages/mosaic/core/src/util/AsyncDispatch.ts b/packages/mosaic/core/src/util/AsyncDispatch.ts index 592dc3fdb..be44a49fd 100644 --- a/packages/mosaic/core/src/util/AsyncDispatch.ts +++ b/packages/mosaic/core/src/util/AsyncDispatch.ts @@ -1,7 +1,6 @@ -type EventCallback = (value: T) => void | Promise; +export type EventCallback = (value: T) => void | Promise; interface DispatchEntry { - callbacks: Set>; pending: Promise | null; queue: DispatchQueue; } @@ -12,16 +11,12 @@ interface QueueNode { } /** - * Event dispatcher supporting asynchronous updates. If an event handler - * callback returns a Promise, the dispatcher waits for all such Promises - * to settle before dispatching future events of the same type. + * Abstract base class for event dispatchers. + * Provides common event listener management and lifecycle methods. */ -export class AsyncDispatch { - _callbacks: Map>; +export abstract class Dispatch { + protected _callbacks: Map>>; - /** - * Create a new asynchronous dispatcher instance. - */ constructor() { this._callbacks = new Map(); } @@ -35,14 +30,10 @@ export class AsyncDispatch { */ addEventListener(type: string, callback: EventCallback): void { if (!this._callbacks.has(type)) { - this._callbacks.set(type, { - callbacks: new Set>(), - pending: null, - queue: new DispatchQueue() - }); + this._callbacks.set(type, new Set>()); } - const entry = this._callbacks.get(type)!; - entry.callbacks.add(callback); + const callbacks = this._callbacks.get(type)!; + callbacks.add(callback); } /** @@ -52,9 +43,9 @@ export class AsyncDispatch { * callback function to remove. */ removeEventListener(type: string, callback: EventCallback): void { - const entry = this._callbacks.get(type); - if (entry) { - entry.callbacks.delete(callback); + const callbacks = this._callbacks.get(type); + if (callbacks) { + callbacks.delete(callback); } } @@ -80,22 +71,63 @@ export class AsyncDispatch { * @param type The event type. * @param value The new event value that will be enqueued. * @returns A dispatch queue filter - * function, or null if all unemitted event values should be filtered. + * function, or null if all un-emitted event values should be filtered. */ emitQueueFilter( _type: string, // eslint-disable-line @typescript-eslint/no-unused-vars - _value: unknown // eslint-disable-line @typescript-eslint/no-unused-vars - ): ((value: unknown) => boolean | null) | null { + _value: T // eslint-disable-line @typescript-eslint/no-unused-vars + ): ((value: T) => boolean | null) | null { // removes all pending items return null; } /** - * Cancel all unemitted event values for the given event type. + * Emit an event value to listeners for the given event type. + * @param type The event type. + * @param value The event value. + */ + abstract emit(type: string, value: T): void; +} + +/** + * Event dispatcher supporting asynchronous updates. If an event handler + * callback returns a Promise, the dispatcher waits for all such Promises + * to settle before dispatching future events of the same type. + */ +export class AsyncDispatch extends Dispatch { + private _entries: Map>; + + /** + * Create a new asynchronous dispatcher instance. + */ + constructor() { + super(); + this._entries = new Map(); + } + + /** + * Add an event listener callback for the provided event type. + * @param type The event type. + * @param callback The event handler + * callback function to add. If the callback has already been + * added for the event type, this method has no effect. + */ + override addEventListener(type: string, callback: EventCallback): void { + super.addEventListener(type, callback); + if (!this._entries.has(type)) { + this._entries.set(type, { + pending: null, + queue: new DispatchQueue(), + }); + } + } + + /** + * Cancel all un-emitted event values for the given event type. * @param type The event type. */ cancel(type: string): void { - const entry = this._callbacks.get(type); + const entry = this._entries.get(type); entry?.queue.clear(); } @@ -107,7 +139,7 @@ export class AsyncDispatch { * @returns A pending event promise. */ async pending(type: string): Promise { - await this._callbacks.get(type)?.pending; + await this._entries.get(type)?.pending; } /** @@ -115,19 +147,26 @@ export class AsyncDispatch { * If a previous emit has not yet resolved, the event value * will be queued to be emitted later. * The actual event value given to listeners will be the result - * of passing the input value through the emitValue() method. + * of passing the input value through the willEmit() method. * @param type The event type. * @param value The event value. */ - emit(type: string, value: T): void { - const entry = this._callbacks.get(type) || {} as DispatchEntry; + override emit(type: string, value: T): void { + if (!this._entries.has(type)) { + this._entries.set(type, { + pending: null, + queue: new DispatchQueue(), + }); + } + const entry = this._entries.get(type)!; if (entry.pending) { // an earlier emit is still processing // enqueue the current update, possibly filtering other pending updates entry.queue.enqueue(value, this.emitQueueFilter(type, value)); } else { const event = this.willEmit(type, value); - const { callbacks, queue } = entry; + const callbacks = this._callbacks.get(type); + const { queue } = entry; if (callbacks?.size) { // broadcast update to callbacks, which may return promises // wait until promises resolve, then process pending updates @@ -144,7 +183,7 @@ export class AsyncDispatch { } /** - * Queue for managing unemitted event values. + * Queue for managing un-emitted event values. */ export class DispatchQueue { next: QueueNode | null = null; @@ -209,4 +248,4 @@ export class DispatchQueue { this.next = next?.next || null; return next?.value; } -} \ No newline at end of file +} diff --git a/packages/mosaic/core/src/util/ObserveDispatch.ts b/packages/mosaic/core/src/util/ObserveDispatch.ts new file mode 100644 index 000000000..1c4d5d072 --- /dev/null +++ b/packages/mosaic/core/src/util/ObserveDispatch.ts @@ -0,0 +1,60 @@ +import { Dispatch, EventCallback } from './AsyncDispatch.js'; + +type EventMap = Record; +type EventKey = Extract; + +/** + * Synchronous event dispatcher that pushes pre-built events directly + * to callbacks without queueing or Promise handling. + */ +export class ObserveDispatch extends Dispatch< + E[EventKey] +> { + /** + * Add an event listener callback for the provided event type. + * @param type The event type. + * @param callback The event handler callback function to add. + */ + addEventListener>( + type: K, + callback: EventCallback, + ): void; + override addEventListener( + type: string, + callback: EventCallback]>, + ): void { + super.addEventListener(type, callback); + } + + /** + * Remove an event listener callback for the provided event type. + * @param type The event type. + * @param callback The event handler callback function to remove. + */ + removeEventListener>( + type: K, + callback: EventCallback, + ): void; + override removeEventListener( + type: string, + callback: EventCallback]>, + ): void { + super.removeEventListener(type, callback); + } + + /** + * Emit an already-constructed event value to listeners for the given event type. + * @param type The event type. + * @param value The complete event object. + */ + emit>(type: K, value: E[K]): void; + override emit(type: string, value: E[EventKey]): void { + const callbacks = this._callbacks.get(type); + if (!callbacks?.size) return; + + const event = this.willEmit(type, value); + for (const callback of callbacks) { + callback(event); + } + } +} diff --git a/packages/mosaic/core/src/util/void-logger.ts b/packages/mosaic/core/src/util/void-logger.ts deleted file mode 100644 index 87bb341cd..000000000 --- a/packages/mosaic/core/src/util/void-logger.ts +++ /dev/null @@ -1,23 +0,0 @@ -interface Logger { - debug(...args: unknown[]): void; - info(...args: unknown[]): void; - log(...args: unknown[]): void; - warn(...args: unknown[]): void; - error(...args: unknown[]): void; - group(label?: string): void; - groupCollapsed(label?: string): void; - groupEnd(): void; -} - -export function voidLogger(): Logger { - return { - debug(): void {}, - info(): void {}, - log(): void {}, - warn(): void {}, - error(): void {}, - group(): void {}, - groupCollapsed(): void {}, - groupEnd(): void {} - }; -} \ No newline at end of file diff --git a/packages/mosaic/core/test/async-dispatch.test.ts b/packages/mosaic/core/test/async-dispatch.test.ts new file mode 100644 index 000000000..ba6df0da5 --- /dev/null +++ b/packages/mosaic/core/test/async-dispatch.test.ts @@ -0,0 +1,125 @@ +import { describe, expect, it } from "vitest"; +import { AsyncDispatch } from "../src/util/AsyncDispatch.js"; + +class FilteredDispatch extends AsyncDispatch { + override emitQueueFilter(): (value: number) => boolean { + return (value: number) => value % 2 === 0; + } +} + +function deferred(): { promise: Promise; resolve: () => void } { + let resolve = (): void => {}; + const promise = new Promise((r) => { + resolve = r; + }); + return { promise, resolve }; +} + +describe("AsyncDispatch", () => { + it("emits to listeners added after an initial listener-less emit", () => { + const dispatch = new AsyncDispatch(); + const values: number[] = []; + + dispatch.emit("value", 1); + dispatch.addEventListener("value", (value) => { + values.push(value); + }); + dispatch.emit("value", 2); + + expect(values).toEqual([2]); + }); + + it("deduplicates and removes listeners", () => { + const dispatch = new AsyncDispatch(); + const values: number[] = []; + const listener = (value: number): void => { + values.push(value); + }; + + dispatch.addEventListener("value", listener); + dispatch.addEventListener("value", listener); + dispatch.emit("value", 1); + + dispatch.removeEventListener("value", listener); + dispatch.emit("value", 2); + + expect(values).toEqual([1]); + }); + + it("replaces queued values by default while an emit is pending", async () => { + const dispatch = new AsyncDispatch(); + const values: number[] = []; + const gate = deferred(); + let block = true; + + dispatch.addEventListener("value", async (value) => { + values.push(value); + if (block) { + block = false; + await gate.promise; + } + }); + + dispatch.emit("value", 1); + dispatch.emit("value", 2); + dispatch.emit("value", 3); + + expect(values).toEqual([1]); + + gate.resolve(); + await dispatch.pending("value"); + await dispatch.pending("value"); + + expect(values).toEqual([1, 3]); + }); + + it("filters queued values using emitQueueFilter", async () => { + const dispatch = new FilteredDispatch(); + const values: number[] = []; + const gate = deferred(); + let block = true; + + dispatch.addEventListener("value", async (value) => { + values.push(value); + if (block) { + block = false; + await gate.promise; + } + }); + + dispatch.emit("value", 1); + dispatch.emit("value", 2); + dispatch.emit("value", 3); + dispatch.emit("value", 4); + + gate.resolve(); + await dispatch.pending("value"); + await dispatch.pending("value"); + + expect(values).toEqual([1, 2, 4]); + }); + + it("cancels queued values without interrupting a pending emit", async () => { + const dispatch = new AsyncDispatch(); + const values: number[] = []; + const gate = deferred(); + let block = true; + + dispatch.addEventListener("value", async (value) => { + values.push(value); + if (block) { + block = false; + await gate.promise; + } + }); + + dispatch.emit("value", 1); + dispatch.emit("value", 2); + dispatch.cancel("value"); + + gate.resolve(); + await dispatch.pending("value"); + + expect(values).toEqual([1]); + }); +}); diff --git a/packages/mosaic/core/test/coordinator.test.ts b/packages/mosaic/core/test/coordinator.test.ts index 4f4349933..53c28e7ce 100644 --- a/packages/mosaic/core/test/coordinator.test.ts +++ b/packages/mosaic/core/test/coordinator.test.ts @@ -1,12 +1,26 @@ import { Query } from '@uwdata/mosaic-sql'; -import { describe, it, expect } from 'vitest'; -import { clausePoint, type Connector, Coordinator, coordinator, type JSONQueryRequest, makeClient, Selection } from '../src/index.js'; +import { describe, it, expect, vi } from 'vitest'; +import { clausePoint, type Connector, Coordinator, coordinator, EventType, type JSONQueryRequest, type Logger, makeClient, MosaicQueryEndEvent, MosaicQueryStartEvent, observeLogger, Selection } from '../src/index.js'; +import { QueryManager } from '../src/QueryManager.js'; import { QueryResult, QueryState } from '../src/util/query-result.js'; async function wait() { return new Promise(resolve => setTimeout(resolve, 0)); } +function createLogger(): Logger { + return { + debug: vi.fn(), + info: vi.fn(), + log: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + group: vi.fn(), + groupCollapsed: vi.fn(), + groupEnd: vi.fn(), + }; +} + describe('coordinator', () => { it('has accessible singleton', () => { // Mock the connector, avoid instantiating default socket connector @@ -54,7 +68,6 @@ describe('coordinator', () => { expect(coord.manager.pendingResults).toHaveLength(4); // resolve promises in reverse order - promises.at(3)!.fulfill(0); await wait(); @@ -98,7 +111,7 @@ describe('coordinator', () => { // Mock the connector const connector = { async query(req: JSONQueryRequest) { - const index = req.sql.includes("WHERE") ? 1 : 0; + const index = req.sql.includes('WHERE') ? 1 : 0; events.push(`CONNECT ${index}`); return { index }; }, @@ -106,7 +119,6 @@ describe('coordinator', () => { // disable cache to ensure routing through connector const coord = new Coordinator(connector, { - logger: null, cache: false, preagg: { enabled: false } }); @@ -120,16 +132,16 @@ describe('coordinator', () => { async prepare() { await wait(); // force wait prepared = true; - events.push("PREPARE"); + events.push('PREPARE'); }, query(filter = []) { events.push(`QUERY ${prepared}`); - return Query.select("*").from("foo").where(filter); + return Query.select('*').from('foo').where(filter); } }); // fire selection update - filterBy.update(clausePoint("foo", 1, { source: {} })); + filterBy.update(clausePoint('foo', 1, { source: {} })); // await initial query, then selection update await client.pending; @@ -139,11 +151,263 @@ describe('coordinator', () => { // query calls should come post-initialization // all queries should include filter clause expect(events).toStrictEqual([ - "PREPARE", - "QUERY true", - "CONNECT 1", - "QUERY true", - "CONNECT 1", + 'PREPARE', + 'QUERY true', + 'CONNECT 1', + 'QUERY true', + 'CONNECT 1', ]); }); + + it('observeLogger reproduces old-style query logging and supports unsubscribe', async () => { + const connector = { + async query() { + return [{ value: 1 }]; + }, + } as unknown as Connector; + + const coord = new Coordinator(connector, { + cache: false, + consolidate: false, + preagg: { enabled: false }, + }); + + const logger = createLogger(); + + const unobserve = observeLogger(coord, logger); + + await coord.query('SELECT 1', { cache: false }); + + expect(logger.groupCollapsed).toHaveBeenCalledTimes(1); + expect(logger.groupCollapsed).toHaveBeenCalledWith('query SELECT 1'); + + expect(logger.log).toHaveBeenCalledTimes(1); + expect(logger.log).toHaveBeenCalledWith('SELECT 1', expect.any(String)); + + expect(logger.groupEnd).toHaveBeenCalledTimes(1); + + unobserve(); + + await coord.query('SELECT 1', { cache: false }); + + // no additional logger calls after unsubscribe + expect(logger.groupCollapsed).toHaveBeenCalledTimes(1); + expect(logger.log).toHaveBeenCalledTimes(1); + expect(logger.groupEnd).toHaveBeenCalledTimes(1); + }); + + it('observeLogger closes query groups for failed queries', async () => { + const error = new Error('boom'); + const connector = { + async query() { + throw error; + }, + } as unknown as Connector; + + const coord = new Coordinator(connector, { + cache: false, + consolidate: false, + preagg: { enabled: false }, + }); + + const logger = createLogger(); + observeLogger(coord, logger); + + await expect(coord.query('SELECT fail', { cache: false })).rejects.toThrow('boom'); + + expect(logger.groupCollapsed).toHaveBeenCalledTimes(1); + expect(logger.groupCollapsed).toHaveBeenCalledWith('query SELECT fail'); + expect(logger.error).toHaveBeenCalledTimes(1); + expect(logger.error).toHaveBeenCalledWith(error); + expect(logger.log).toHaveBeenCalledTimes(1); + expect(logger.log).toHaveBeenCalledWith('SELECT fail', expect.any(String)); + expect(logger.groupEnd).toHaveBeenCalledTimes(1); + }); + + it('observeLogger derives elapsed time from query event timestamps', () => { + const connector = { + async query() { + return [{ value: 1 }]; + }, + } as unknown as Connector; + + const coord = new Coordinator(connector, { + cache: false, + consolidate: false, + logger: null, + preagg: { enabled: false }, + }); + const logger = createLogger(); + observeLogger(coord, logger); + + coord.eventBus.emit( + EventType.QueryStart, + new MosaicQueryStartEvent({ + queryId: 1, + query: 'SELECT timed', + materialized: false, + timestamp: 100, + }), + ); + coord.eventBus.emit( + EventType.QueryEnd, + new MosaicQueryEndEvent({ + queryId: 1, + query: 'SELECT timed', + materialized: false, + status: 'success', + timestamp: 123.45, + }), + ); + + expect(logger.log).toHaveBeenCalledWith('SELECT timed', '23.5'); + expect(logger.groupEnd).toHaveBeenCalledTimes(1); + }); + + it('observeLogger does not close a group for unmatched query end events', () => { + const connector = { + async query() { + return [{ value: 1 }]; + }, + } as unknown as Connector; + + const coord = new Coordinator(connector, { + cache: false, + consolidate: false, + logger: null, + preagg: { enabled: false }, + }); + const logger = createLogger(); + observeLogger(coord, logger); + + coord.eventBus.emit( + EventType.QueryEnd, + new MosaicQueryEndEvent({ + queryId: 1, + query: 'SELECT unmatched', + materialized: false, + status: 'success', + }), + ); + + expect(logger.log).toHaveBeenCalledWith('SELECT unmatched'); + expect(logger.groupEnd).not.toHaveBeenCalled(); + }); + + it('observeLogger clears in-flight query state on unsubscribe', () => { + const connector = { + async query() { + return [{ value: 1 }]; + }, + } as unknown as Connector; + + const coord = new Coordinator(connector, { + cache: false, + consolidate: false, + logger: null, + preagg: { enabled: false }, + }); + const logger = createLogger(); + const unobserve = observeLogger(coord, logger); + + coord.eventBus.emit( + EventType.QueryStart, + new MosaicQueryStartEvent({ + queryId: 1, + query: 'SELECT in_flight', + materialized: false, + timestamp: 100, + }), + ); + + unobserve(); + observeLogger(coord, logger); + + coord.eventBus.emit( + EventType.QueryEnd, + new MosaicQueryEndEvent({ + queryId: 1, + query: 'SELECT in_flight', + materialized: false, + status: 'success', + timestamp: 200, + }), + ); + + expect(logger.groupCollapsed).toHaveBeenCalledTimes(1); + expect(logger.log).toHaveBeenCalledWith('SELECT in_flight'); + expect(logger.groupEnd).not.toHaveBeenCalled(); + }); + + it('wires custom query managers to the coordinator event bus', async () => { + const connector = { + async query() { + return [{ value: 1 }]; + }, + } as unknown as Connector; + + const manager = new QueryManager(); + const coord = new Coordinator(connector, { + cache: false, + consolidate: false, + logger: null, + manager, + preagg: { enabled: false }, + }); + const starts: MosaicQueryStartEvent[] = []; + + coord.eventBus.addEventListener(EventType.QueryStart, (event) => { + starts.push(event); + }); + + await coord.query('SELECT 1', { cache: false }); + + expect(coord.manager).toBe(manager); + expect(starts).toHaveLength(1); + expect(starts[0]?.query).toBe('SELECT 1'); + }); + + it('supports legacy coordinator logger configuration through the event bus', async () => { + const connector = { + async query() { + return [{ value: 1 }]; + }, + } as unknown as Connector; + + const logger = createLogger(); + const coord = new Coordinator(connector, { + cache: false, + consolidate: false, + logger, + preagg: { enabled: false }, + }); + + expect(coord.logger()).toBe(logger); + + await coord.query('SELECT 1', { cache: false }); + + expect(logger.groupCollapsed).toHaveBeenCalledTimes(1); + expect(logger.log).toHaveBeenCalledTimes(1); + expect(logger.groupEnd).toHaveBeenCalledTimes(1); + + const nextLogger = createLogger(); + expect(coord.logger(nextLogger)).toBe(nextLogger); + + await coord.query('SELECT 2', { cache: false }); + + expect(logger.groupCollapsed).toHaveBeenCalledTimes(1); + expect(logger.log).toHaveBeenCalledTimes(1); + expect(logger.groupEnd).toHaveBeenCalledTimes(1); + expect(nextLogger.groupCollapsed).toHaveBeenCalledTimes(1); + expect(nextLogger.log).toHaveBeenCalledTimes(1); + expect(nextLogger.groupEnd).toHaveBeenCalledTimes(1); + + expect(coord.logger(null)).toBeNull(); + + await coord.query('SELECT 3', { cache: false }); + + expect(nextLogger.groupCollapsed).toHaveBeenCalledTimes(1); + expect(nextLogger.log).toHaveBeenCalledTimes(1); + expect(nextLogger.groupEnd).toHaveBeenCalledTimes(1); + }); }); diff --git a/packages/mosaic/core/test/query-manager.test.ts b/packages/mosaic/core/test/query-manager.test.ts index c4f4be54b..e5206a0bd 100644 --- a/packages/mosaic/core/test/query-manager.test.ts +++ b/packages/mosaic/core/test/query-manager.test.ts @@ -1,7 +1,15 @@ import { describe, it, expect } from 'vitest'; import { QueryManager } from '../src/QueryManager.js'; import { QueryResult } from '../src/util/query-result.js'; -import { QueryRequest } from '../src/types.js'; +import type { QueryRequest } from '../src/types.js'; +import { ObserveDispatch } from '../src/util/ObserveDispatch.js'; +import { + EventType, + type MosaicEventMap, + MosaicErrorEvent, + MosaicQueryStartEvent, + MosaicQueryEndEvent, +} from '../src/Events.js'; describe('QueryManager', () => { it('should run a simple query', async () => { @@ -13,12 +21,12 @@ describe('QueryManager', () => { query: async ({ sql }) => { expect(sql).toBe('SELECT 1'); return [{ column: 1 }]; - } + }, }); const request: QueryRequest = { type: 'arrow', - query: 'SELECT 1' + query: 'SELECT 1', }; const result = queryManager.request(request); @@ -37,17 +45,17 @@ describe('QueryManager', () => { query: ({ sql }) => { expect(sql).toBe('CREATE TABLE test (id INT)'); return new Promise(() => {}); - } + }, }); const request1: QueryRequest = { type: 'exec', - query: 'CREATE TABLE test (id INT)' + query: 'CREATE TABLE test (id INT)', }; const request2: QueryRequest = { type: 'arrow', - query: 'SELECT * FROM test' + query: 'SELECT * FROM test', }; queryManager.request(request1); @@ -55,4 +63,177 @@ describe('QueryManager', () => { expect(queryManager.pendingResults).toHaveLength(1); }); + + it('emits typed QueryStart and QueryEnd events with type and timestamp', async () => { + const eventBus = new ObserveDispatch(); + const queryManager = new QueryManager(); + queryManager.attachEventBus(eventBus); + + const starts: MosaicQueryStartEvent[] = []; + const ends: MosaicQueryEndEvent[] = []; + + eventBus.addEventListener(EventType.QueryStart, (event) => { + starts.push(event); + }); + eventBus.addEventListener(EventType.QueryEnd, (event) => { + ends.push(event); + }); + + queryManager.connector({ + // @ts-expect-error assumes type value + query: async ({ sql }) => [{ sql }], + }); + + const result = queryManager.request({ + type: 'arrow', + query: 'SELECT 42', + cache: false, + }); + + await result; + + expect(starts).toHaveLength(1); + expect(ends).toHaveLength(1); + + const start = starts[0]; + const end = ends[0]; + + expect(start).toBeInstanceOf(MosaicQueryStartEvent); + expect(end).toBeInstanceOf(MosaicQueryEndEvent); + + expect(start.type).toBe(EventType.QueryStart); + expect(end.type).toBe(EventType.QueryEnd); + + expect(typeof start.timestamp).toBe('number'); + expect(typeof end.timestamp).toBe('number'); + + expect(start.query).toBe('SELECT 42'); + expect(end.query).toBe('SELECT 42'); + expect(start.queryId).toBe(1); + expect(end.queryId).toBe(start.queryId); + + expect(start.materialized).toBe(false); + expect(end.materialized).toBe(false); + expect(end.status).toBe('success'); + }); + + it('emits QueryStart and QueryEnd for cached requests', async () => { + const eventBus = new ObserveDispatch(); + const queryManager = new QueryManager(); + queryManager.attachEventBus(eventBus); + + const starts: MosaicQueryStartEvent[] = []; + const ends: MosaicQueryEndEvent[] = []; + let queryCalls = 0; + + eventBus.addEventListener(EventType.QueryStart, (event) => { + starts.push(event); + }); + eventBus.addEventListener(EventType.QueryEnd, (event) => { + ends.push(event); + }); + + queryManager.connector({ + // @ts-expect-error assumes type value + query: async () => { + queryCalls += 1; + return [{ value: 1 }]; + }, + }); + + queryManager.cache(true); + + await queryManager.request({ + type: 'arrow', + query: 'SELECT 1', + cache: true, + }); + + await queryManager.request({ + type: 'arrow', + query: 'SELECT 1', + cache: true, + }); + + expect(queryCalls).toBe(1); + expect(starts).toHaveLength(2); + expect(ends).toHaveLength(2); + + for (const event of starts) { + expect(event.type).toBe(EventType.QueryStart); + expect(typeof event.timestamp).toBe('number'); + expect(event.query).toBe('SELECT 1'); + expect(event.materialized).toBe(true); + } + + for (const event of ends) { + expect(event.type).toBe(EventType.QueryEnd); + expect(typeof event.timestamp).toBe('number'); + expect(event.query).toBe('SELECT 1'); + expect(event.materialized).toBe(true); + expect(event.status).toBe('success'); + } + + expect(starts.map((event) => event.queryId)).toEqual([1, 2]); + expect(ends.map((event) => event.queryId)).toEqual([1, 2]); + }); + + it('emits QueryEnd with error status when a query fails', async () => { + const eventBus = new ObserveDispatch(); + const queryManager = new QueryManager(); + queryManager.attachEventBus(eventBus); + + const starts: MosaicQueryStartEvent[] = []; + const errors: MosaicErrorEvent[] = []; + const ends: MosaicQueryEndEvent[] = []; + + eventBus.addEventListener(EventType.QueryStart, (event) => { + starts.push(event); + }); + eventBus.addEventListener(EventType.Error, (event) => { + errors.push(event); + }); + eventBus.addEventListener(EventType.QueryEnd, (event) => { + ends.push(event); + }); + + const error = new Error('boom'); + + queryManager.connector({ + query: async () => { + throw error; + }, + }); + + await expect(queryManager.request({ + type: 'arrow', + query: 'SELECT fail', + cache: false, + })).rejects.toThrow('boom'); + + expect(starts).toHaveLength(1); + expect(errors).toHaveLength(1); + expect(ends).toHaveLength(1); + + expect(errors[0].message).toBe('boom'); + expect(errors[0].error).toBe(error); + expect(errors[0].queryId).toBe(starts[0].queryId); + + expect(ends[0].query).toBe('SELECT fail'); + expect(ends[0].materialized).toBe(false); + expect(ends[0].queryId).toBe(starts[0].queryId); + expect(ends[0].status).toBe('error'); + }); + + it('treats event bus attachment as one-time coordinator wiring', () => { + const eventBus = new ObserveDispatch(); + const otherBus = new ObserveDispatch(); + const queryManager = new QueryManager(); + + queryManager.attachEventBus(eventBus); + expect(() => queryManager.attachEventBus(eventBus)).not.toThrow(); + expect(() => queryManager.attachEventBus(otherBus)).toThrow( + 'QueryManager event bus is already attached.', + ); + }); }); diff --git a/packages/vgplot/widget/src/index.js b/packages/vgplot/widget/src/index.js index da2de0f99..b6174fec7 100644 --- a/packages/vgplot/widget/src/index.js +++ b/packages/vgplot/widget/src/index.js @@ -61,7 +61,7 @@ export default { async function updateSpec() { const spec = getSpec(); reset(); - logger.log('Setting spec:', spec); + logger?.log('Setting spec:', spec); const dom = await instantiateSpec(spec); view.el.replaceChildren(dom.element); @@ -101,27 +101,27 @@ export default { view.model.on('change:preagg_schema', () => configureCoordinator()); view.model.on('msg:custom', (msg, buffers) => { - logger.group(`query ${msg.uuid}`); - logger.log('received message', msg, buffers); + logger?.group(`query ${msg.uuid}`); + logger?.log('received message', msg, buffers); const query = openQueries.get(msg.uuid); openQueries.delete(msg.uuid); - logger.log(query.query.sql, (performance.now() - query.startTime).toFixed(1)); + logger?.log(query.query.sql, (performance.now() - query.startTime).toFixed(1)); if (msg.error) { query.reject(msg.error); - logger.error(msg.error); + logger?.error(msg.error); } else { switch (msg.type) { case 'arrow': { const table = decodeIPC(buffers[0].buffer); - logger.log('table', table); + logger?.log('table', table); query.resolve(table); break; } case 'json': { - logger.log('json', msg.result); + logger?.log('json', msg.result); query.resolve(msg.result); break; } @@ -131,7 +131,7 @@ export default { } } } - logger.groupEnd(); + logger?.groupEnd(); }); coordinator().databaseConnector(connector);