diff --git a/packages/comms/README.md b/packages/comms/README.md index 9bd6d47514..b082296392 100644 --- a/packages/comms/README.md +++ b/packages/comms/README.md @@ -1,6 +1,72 @@ # @hpcc-js/comms This package is part of the mono repository "@hpcc-js" (aka Visualization Framework), for more information including [Quick Start](https://github.com/hpcc-systems/Visualization/wiki/Quick-Start), [Gallery](https://raw.githack.com/hpcc-systems/Visualization/trunk/demos/gallery/gallery.html) and [Tutorials](https://github.com/hpcc-systems/Visualization/wiki/Tutorials), please visit the main page on GitHub: [hpcc-systems/Visualization](https://github.com/hpcc-systems/Visualization). +## WsSQL2 + +`SQL2Service` exposes the WsSQL2 operations (`ExecuteQuery`, `ExplainQuery`, `FetchResultSet`, `CloseResultSet`, `LogicalFileMeta`, `SplitQuery`, and `Ping`) with request and response types in `WsSQL2`. + +```ts +import { SQL2Service } from "@hpcc-js/comms"; + +const sql = new SQL2Service({ baseUrl: "http://localhost:8510" }); +const { CursorId, ResultSetMetadata } = await sql.ExecuteQuery({ + SqlText: "SELECT 1", IncludeResultSetMetadata: true +}); +try { + const { Rows, Complete } = await sql.FetchResultSet({ CursorId, FetchSize: 100 }); + console.log(ResultSetMetadata, Rows, Complete); +} finally { + await sql.CloseResultSet({ CursorId }); +} +``` + +To regenerate the WsSQL2 TypeScript bindings from the local WSDL, run `npm run wsdl-sql2` in this package. +`ExplainQuery` returns `Plan` / `PlanJson`; a fetch may include `RuntimeGraph` / `RuntimeGraphJson` when `ExecuteQuery` requests `IncludeRuntimeGraph`. + +For small results, `queryRows()` collects rows as objects keyed by column label and closes its statement automatically: + +```ts +import { SQLConnection } from "@hpcc-js/comms"; + +const connection = new SQLConnection({ baseUrl: "http://localhost:8510" }); +try { + const rows = await connection.queryRows("SELECT id FROM hpcc_scan('example::file')", 100); + console.log(rows[0]?.id); + const { Plan } = await connection.explainQuery("SELECT 1"); + const file = await connection.logicalFileMeta("example::file"); + const available = await connection.ping(); +} finally { + await connection.close(); +} +``` + +`queryRows()` preserves native JSON values (including `null`) and rejects missing or duplicate column labels, ignoring case. It holds all rows in memory; use `createStatement()` and `next()` below for large results or early cancellation. `explainQuery()` returns the typed service response, `logicalFileMeta()` returns the file metadata or `undefined`, and `ping()` returns `true` on success (or rejects on failure). + +For a JDBC-style, forward-only client, use the async `SQLConnection` helper: + +```ts +import { SQLConnection } from "@hpcc-js/comms"; + +const connection = new SQLConnection({ baseUrl: "http://localhost:8510" }); +try { + const statement = connection.createStatement(); + statement.setFetchSize(100); + const statements = await connection.splitQuery("SELECT 1; SELECT 2;"); + const result = await statement.executeQuery("SELECT id FROM hpcc_scan('example::file')"); + while (await result.next()) { + console.log(result.getObject("id")); + } +} finally { + await connection.close(); +} +``` + +`executeQuery` prepares SQL and returns metadata and a cursor without executing it. The first `next()` initiates execution; subsequent calls fetch pages on demand until `Complete` is true, including empty intermediate pages. `getObject` accepts a 1-based column index or a case-insensitive label, and `getMetaData()` exposes the WsSQL2 column descriptors. Always `await` `close()` on the result, statement, or connection when finished: it releases any active server cursor. All three support `Symbol.asyncDispose`, so `await using` also closes their owned cursors on scope exit; synchronous disposal cannot wait for the server. `statement.cancel()` or `result.cancel()` closes the cursor immediately, even while a fetch is in progress, and ends that result's iteration. Executing another query on the same statement closes its previous result first. This helper does not implement transactions, parameter binding, or JDBC type conversions. +`connection.splitQuery(sql)` delegates statement splitting to the server and returns the individual SQL strings (or `[]` for a comment-only script); it does not execute them. Pass each statement to `executeQuery()` separately when running a multi-statement script. +If `CloseResultSet` fails, the failure is reported and the cursor ID is retained; retry `close()` on the result, statement, or connection to attempt cleanup again. + +The integration tests in `tests/wsSQL2.spec.ts` run in both the Node and browser test projects. They require ESP at `http://127.0.0.1:8010/` and WsSQL2 at `http://127.0.0.1:8510/`; service-backed tests are skipped when the WsSQL2 ping fails. One test creates a unique 1,000-row logical file, verifies paging and early cursor close, then removes the file and its workunit. + ## Exported Widgets * [Connection](https://raw.githack.com/hpcc-systems/Visualization/trunk/demos/gallery/playground.html?./samples/comms/Connection.js) diff --git a/packages/comms/package.json b/packages/comms/package.json index 2358766188..229de9ba78 100644 --- a/packages/comms/package.json +++ b/packages/comms/package.json @@ -69,6 +69,7 @@ "wsdl-packageprocess": "node ./lib-esm/index.js -k --url=http://localhost:8010/WsPackageProcess?wsdl --outDir=./src/services/wsdl", "wsdl-sasha": "node ./lib-esm/index.js -k --url=http://localhost:8010/WsSasha?wsdl --outDir=./src/services/wsdl", "wsdl-smc": "node ./lib-esm/index.js -k --url=http://localhost:8010/WsSMC?wsdl --outDir=./src/services/wsdl", + "wsdl-sql2": "node ./lib-esm/index.js -k --url=http://localhost:8510/WsSQL2?wsdl --outDir=./src/services/wsdl", "wsdl-resources": "node ./lib-esm/index.js -k --url=http://localhost:8010/WsResources?wsdl --outDir=./src/services/wsdl", "wsdl-store": "node ./lib-esm/index.js -k --url=http://localhost:8010/WsStore?wsdl --outDir=./src/services/wsdl", "wsdl-topology": "node ./lib-esm/index.js -k --url=http://localhost:8010/WsTopology?wsdl --outDir=./src/services/wsdl", diff --git a/packages/comms/src/connection.ts b/packages/comms/src/connection.ts index 625ee4661d..771c46db6a 100644 --- a/packages/comms/src/connection.ts +++ b/packages/comms/src/connection.ts @@ -92,7 +92,12 @@ export function serializeRequest(obj: any, encodeRequest: boolean = true, prefix } export function deserializeResponse(body: string) { - return JSON.parse(body); + try { + return JSON.parse(body); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + throw Object.assign(new Error(`Error parsing JSON response '${message}'`), { cause: error }); + } } export function jsonp(opts: IOptions, action: string, request: any = {}, responseType: ResponseType = "json", header?: any): Promise { @@ -106,7 +111,11 @@ export function jsonp(opts: IOptions, action: string, request: any = {}, respons (window as any)[callbackName] = function (response: any) { respondedTimeout = 0; doCallback(); - resolve(responseType === "json" && typeof response === "string" ? deserializeResponse(response) : response); + try { + resolve(responseType === "json" && typeof response === "string" ? deserializeResponse(response) : response); + } catch (error) { + reject(error); + } }; const script = document.createElement("script"); let url = join(opts.baseUrl, action); diff --git a/packages/comms/src/ecl/sql2.ts b/packages/comms/src/ecl/sql2.ts new file mode 100644 index 0000000000..a4546228c7 --- /dev/null +++ b/packages/comms/src/ecl/sql2.ts @@ -0,0 +1,376 @@ +import { IConnection, IOptions } from "../connection.ts"; +import { SQL2Service, WsSQL2 } from "../services/wsSQL2.ts"; + +export type SQLValue = string | number | boolean | null | SQLValue[] | { [key: string]: SQLValue }; +export type SQLRow = Record; + +export class SQLCleanupError extends Error { + constructor(message: string, public readonly cause: unknown, public readonly cleanupErrors: unknown[]) { + super(message); + } +} + +function checkExceptions(response: { Exceptions?: WsSQL2.Exceptions }): void { + if (response.Exceptions) { + throw new Error(response.Exceptions.Exception?.map(e => e.Message ?? e.Code).join("; ") || "WsSQL2 request failed"); + } +} + +function rowsFrom(response: { Rows?: string }): SQLValue[][] { + if (typeof response.Rows !== "string") { + throw new Error("WsSQL2 response is missing Rows"); + } + let rows: unknown; + try { + rows = JSON.parse(response.Rows); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + throw Object.assign(new Error(`Error parsing WsSQL2 Rows JSON '${message}'`), { cause: error }); + } + if (!Array.isArray(rows) || !rows.every(row => Array.isArray(row))) { + throw new Error("WsSQL2 Rows must be an array of arrays"); + } + return rows as SQLValue[][]; +} + +export class SQLConnection { + protected readonly service: SQL2Service; + protected readonly statements = new Set(); + protected closed = false; + protected closing?: Promise; + + constructor(options: IOptions | IConnection | SQL2Service) { + this.service = options instanceof SQL2Service ? options : new SQL2Service(options); + } + + createStatement(): SQLStatement { + this.checkOpen(); + const statement = new SQLStatement(this.service, () => this.statements.delete(statement)); + this.statements.add(statement); + return statement; + } + + async queryRows(sql: string, fetchSize?: number): Promise { + const statement = this.createStatement(); + const rows: SQLRow[] = []; + try { + if (fetchSize !== undefined) statement.setFetchSize(fetchSize); + const result = await statement.executeQuery(sql); + while (await result.next()) rows.push(result.getRow()); + } catch (error) { + try { + await statement.close(); + } catch (closeError) { + throw new SQLCleanupError("Failed to query rows and close statement", error, [closeError]); + } + throw error; + } + await statement.close(); + return rows; + } + + async explainQuery(sql: string): Promise { + this.checkOpen(); + if (!sql.trim()) throw new Error("SQL text is required"); + const response = await this.service.ExplainQuery({ SqlText: sql }); + checkExceptions(response); + return response; + } + + async logicalFileMeta(logicalName: string): Promise { + this.checkOpen(); + if (!logicalName.trim()) throw new Error("Logical file name is required"); + const response = await this.service.LogicalFileMeta({ LogicalName: logicalName }); + checkExceptions(response); + return response.LogicalFileMeta; + } + + async ping(): Promise { + this.checkOpen(); + await this.service.Ping({}); + return true; + } + + async splitQuery(sql: string): Promise { + this.checkOpen(); + if (!sql.trim()) return []; + const response = await this.service.SplitQuery({ SqlText: sql }); + checkExceptions(response); + const statements = response.Statements?.Statement ?? []; + if (!Array.isArray(statements) || !statements.every(statement => typeof statement === "string")) { + throw new Error("WsSQL2 response has invalid Statements"); + } + return statements; + } + + close(): Promise { + if (this.closing) return this.closing; + this.closed = true; + this.closing = (async () => { + const results = await Promise.allSettled([...this.statements].map(statement => statement.close())); + const errors = results.filter((result): result is PromiseRejectedResult => result.status === "rejected").map(result => result.reason); + if (errors.length) throw new SQLCleanupError("Failed to close SQL statements", errors[0], errors); + })(); + void this.closing.then( + () => { this.closing = undefined; }, + () => { this.closing = undefined; } + ); + return this.closing; + } + + [Symbol.asyncDispose](): Promise { + return this.close(); + } + + checkOpen(): void { + if (this.closed) throw new Error("SQL connection is closed"); + } +} + +export class SQLStatement { + protected result?: SQLResultSet; + protected closed = false; + protected fetchSize = 0; + protected executing?: Promise; + protected closing?: Promise; + + constructor(protected readonly service: SQL2Service, protected readonly onClose: () => void) { + } + + setFetchSize(size: number): void { + this.checkOpen(); + if (!Number.isSafeInteger(size) || size < 0) throw new RangeError("Fetch size must be a non-negative integer"); + this.fetchSize = size; + } + + cancel(): Promise { + this.checkOpen(); + if (this.executing) return this.executing.then(result => result.cancel()); + return this.result?.cancel() ?? Promise.resolve(); + } + + executeQuery(sql: string): Promise { + this.checkOpen(); + if (this.executing) throw new Error("SQL statement is already executing"); + if (!sql.trim()) throw new Error("SQL text is required"); + return this.execute(sql); + } + + protected async execute(sql: string): Promise { + const operation = this.runQuery(sql); + this.executing = operation; + try { + return await operation; + } finally { + this.executing = undefined; + } + } + + protected async runQuery(sql: string): Promise { + await this.result?.close(); + this.result = undefined; + if (this.closed) throw new Error("SQL statement is closed"); + const response = await this.service.ExecuteQuery({ + SqlText: sql, IncludeResultSetMetadata: true + }); + const columns = response.ResultSetMetadata?.Column; + const result = response.CursorId ? new SQLResultSet(this.service, columns ?? [], response.CursorId, this.fetchSize) : undefined; + this.result = result; + try { + checkExceptions(response); + if (!response.CursorId) throw new Error("WsSQL2 response is missing CursorId"); + if (!columns) throw new Error("WsSQL2 response is missing result-set metadata"); + if (this.closed) throw new Error("SQL statement is closed"); + return result; + } catch (error) { + if (result) { + try { + await result.close(); + } catch (closeError) { + throw new SQLCleanupError("Failed to parse result and close cursor", error, [closeError]); + } + } + throw error; + } + } + + close(): Promise { + if (this.closing) return this.closing; + this.closed = true; + this.closing = (async () => { + if (this.executing) { + try { + await this.executing; + } catch { + // The query caller receives its own error; close still releases its cursor. + } + } + await this.result?.close(); + this.onClose(); + })(); + void this.closing.then( + () => { this.closing = undefined; }, + () => { this.closing = undefined; } + ); + return this.closing; + } + + [Symbol.asyncDispose](): Promise { + return this.close(); + } + + protected checkOpen(): void { + if (this.closed) throw new Error("SQL statement is closed"); + } +} + +export class SQLResultSet { + protected index = -1; + protected row?: SQLValue[]; + protected wasNullValue = false; + protected closed = false; + protected advancing?: Promise; + protected closing?: Promise; + + constructor( + protected readonly service: SQL2Service, + protected readonly columns: WsSQL2.Column[], + protected cursor: string | undefined, + protected readonly fetchSize: number + ) { + } + protected rows: SQLValue[][] = []; + protected complete = false; + + getMetaData(): readonly WsSQL2.Column[] { + this.checkOpen(); + return this.columns; + } + + findColumn(label: string): number { + this.checkOpen(); + const index = this.columns.findIndex(column => column.ColumnLabel?.toLowerCase() === label.toLowerCase()); + if (index < 0) throw new Error(`Unknown SQL column: ${label}`); + return index + 1; + } + + next(): Promise { + this.checkOpen(); + if (this.advancing) throw new Error("SQL result set is already fetching"); + return this.advance(); + } + + private async advance(): Promise { + const operation = this.advancePage(); + this.advancing = operation; + try { + return await operation; + } finally { + this.advancing = undefined; + } + } + + private async advancePage(): Promise { + this.row = undefined; + try { + while (++this.index >= this.rows.length) { + this.checkOpen(); + if (this.complete) return false; + const response = await this.service.FetchResultSet({ CursorId: this.cursor, FetchSize: this.fetchSize }); + checkExceptions(response); + if (typeof response.Complete !== "boolean") throw new Error("WsSQL2 fetch response is missing Complete"); + const rows = rowsFrom(response); + this.checkOpen(); + this.complete = response.Complete; + if (this.complete) this.cursor = undefined; + this.rows = rows; + this.index = -1; + } + this.checkOpen(); + const row = this.rows[this.index]; + if (row.length !== this.columns.length) throw new Error("WsSQL2 row width does not match result metadata"); + this.row = row; + return true; + } catch (error) { + this.closed = true; + try { + await this.close(); + } catch (closeError) { + throw new SQLCleanupError("Failed to read result and close cursor", error, [closeError]); + } + throw error; + } + } + + getObject(column: number | string): SQLValue { + this.checkOpen(); + if (!this.row) throw new Error("SQL result set is not positioned on a row"); + const index = typeof column === "string" ? this.findColumn(column) : column; + if (!Number.isInteger(index) || index < 1 || index > this.columns.length) { + throw new RangeError(`SQL column index out of range: ${index}`); + } + const value = this.row[index - 1]; + this.wasNullValue = value === null; + return value; + } + + getString(column: number | string): string | null { + const value = this.getObject(column); + return value === null ? null : String(value); + } + + getRow(): SQLRow { + this.checkOpen(); + const row = this.row; + if (!row) throw new Error("SQL result set is not positioned on a row"); + const result: SQLRow = Object.create(null); + const labels = new Set(); + this.columns.forEach((column, index) => { + const label = column.ColumnLabel; + if (!label) throw new Error("WsSQL2 column is missing a label"); + if (labels.has(label.toLowerCase())) throw new Error(`Duplicate SQL column label: ${label}`); + labels.add(label.toLowerCase()); + result[label] = row[index]; + }); + return result; + } + + wasNull(): boolean { + this.checkOpen(); + return this.wasNullValue; + } + + close(): Promise { + if (this.closing) return this.closing; + this.closed = true; + this.closing = (async () => { + await this.closeCursor(); + })(); + void this.closing.then( + () => { this.closing = undefined; }, + () => { this.closing = undefined; } + ); + return this.closing; + } + + [Symbol.asyncDispose](): Promise { + return this.close(); + } + + cancel(): Promise { + return this.close(); + } + + protected async closeCursor(): Promise { + if (this.cursor) { + const cursor = this.cursor; + const response = await this.service.CloseResultSet({ CursorId: cursor }); + checkExceptions(response); + this.cursor = undefined; + } + } + + protected checkOpen(): void { + if (this.closed) throw new Error("SQL result set is closed"); + } +} diff --git a/packages/comms/src/index.common.ts b/packages/comms/src/index.common.ts index 340977de85..0078dc7eda 100644 --- a/packages/comms/src/index.common.ts +++ b/packages/comms/src/index.common.ts @@ -15,6 +15,7 @@ export * from "./services/wsPackageProcess.ts"; export * from "./services/wsResources.ts"; export * from "./services/wsSasha.ts"; export * from "./services/wsSMC.ts"; +export * from "./services/wsSQL2.ts"; export * from "./services/wsStore.ts"; export * from "./services/wsTopology.ts"; export * from "./services/wsWorkunits.ts"; @@ -27,6 +28,7 @@ export * from "./ecl/query.ts"; export * from "./ecl/resource.ts"; export * from "./ecl/result.ts"; export * from "./ecl/scope.ts"; +export * from "./ecl/sql2.ts"; export * from "./ecl/sourceFile.ts"; export * from "./ecl/store.ts"; export * from "./ecl/timer.ts"; diff --git a/packages/comms/src/services/wsCloud.ts b/packages/comms/src/services/wsCloud.ts index ba3e7ef689..1035c75741 100644 --- a/packages/comms/src/services/wsCloud.ts +++ b/packages/comms/src/services/wsCloud.ts @@ -65,8 +65,10 @@ export class CloudService extends CloudServiceBase { const obj = typeof response.Result === "string" ? JSON.parse(response.Result) : response.Result; return mapPods(obj?.items ?? []); } catch (error) { - logger.error(`Error parsing V1Pods json '${(error instanceof Error ? error.message : String(error))}'`); - return []; + const message = error instanceof Error ? error.message : String(error); + const parseError = Object.assign(new Error(`Error parsing V1Pods json '${message}'`), { cause: error }); + logger.error(parseError.message); + throw parseError; } }); } diff --git a/packages/comms/src/services/wsEcl.ts b/packages/comms/src/services/wsEcl.ts index 5a9b8f1550..447a63e721 100644 --- a/packages/comms/src/services/wsEcl.ts +++ b/packages/comms/src/services/wsEcl.ts @@ -82,7 +82,13 @@ export class EclService extends Service { requestJson(querySet: string, queryId: string): Promise { // http://192.168.3.22:8002/WsEcl/example/request/query/roxie/peopleaccounts/json?display return this._connection.send(`example/request/query/${querySet}/${queryId}/json`, {}, "text").then(response => { - const requestSchema = JSON.parse(response); + let requestSchema; + try { + requestSchema = JSON.parse(response); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + throw Object.assign(new Error(`Error parsing WsEcl request JSON '${message}'`), { cause: error }); + } for (const key in requestSchema) { return requestSchema[key]; } @@ -93,7 +99,13 @@ export class EclService extends Service { responseJson(querySet: string, queryId: string): Promise { // http://192.168.3.22:8002/WsEcl/example/response/query/roxie/peopleaccounts/json?display return this._connection.send(`example/response/query/${querySet}/${queryId}/json`, {}, "text").then(response => { - const responseSchema = JSON.parse(response); + let responseSchema; + try { + responseSchema = JSON.parse(response); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + throw Object.assign(new Error(`Error parsing WsEcl response JSON '${message}'`), { cause: error }); + } for (const key in responseSchema) { return responseSchema[key].Results; } diff --git a/packages/comms/src/services/wsLogaccess.ts b/packages/comms/src/services/wsLogaccess.ts index 85615d6353..6f21eaf1d3 100644 --- a/packages/comms/src/services/wsLogaccess.ts +++ b/packages/comms/src/services/wsLogaccess.ts @@ -249,21 +249,22 @@ export class LogaccessService extends LogaccessServiceBase { }; return this.GetLogs(getLogsRequest).then(response => { + let logLines; try { - const logLines = JSON.parse(response.LogLines); - const lines = knownLogManagerTypes.has(logInfo.RemoteLogManagerType) - ? (logLines.lines?.map((line: any) => this.convertLogLine(columnMap, line)) ?? []) - : (logger.warning(`Unknown RemoteLogManagerType: ${logInfo.RemoteLogManagerType}`), []); - return { - lines, - total: response.TotalLogLinesAvailable ?? 10000 - }; - } catch (e: any) { - logger.error(e.message ?? e); + logLines = JSON.parse(response.LogLines); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + const parseError = Object.assign(new Error(`Error parsing log lines json '${message}'`), { cause: error }); + logger.error(parseError.message); + throw parseError; } + + const lines = knownLogManagerTypes.has(logInfo.RemoteLogManagerType) + ? (logLines.lines?.map((line: any) => this.convertLogLine(columnMap, line)) ?? []) + : (logger.warning(`Unknown RemoteLogManagerType: ${logInfo.RemoteLogManagerType}`), []); return { - lines: [], - total: 0 + lines, + total: response.TotalLogLinesAvailable ?? 10000 }; }); } diff --git a/packages/comms/src/services/wsSQL2.ts b/packages/comms/src/services/wsSQL2.ts new file mode 100644 index 0000000000..5bf283b5a7 --- /dev/null +++ b/packages/comms/src/services/wsSQL2.ts @@ -0,0 +1,8 @@ +import { SQL2ServiceBase, WsSQL2 } from "./wsdl/WsSQL2/v1.00/WsSQL2.ts"; + +export { + type WsSQL2 +}; + +export class SQL2Service extends SQL2ServiceBase { +} diff --git a/packages/comms/src/services/wsdl/WsSQL2/current/WsSQL2.ts b/packages/comms/src/services/wsdl/WsSQL2/current/WsSQL2.ts new file mode 100644 index 0000000000..96946a5b99 --- /dev/null +++ b/packages/comms/src/services/wsdl/WsSQL2/current/WsSQL2.ts @@ -0,0 +1,222 @@ +import { IConnection, IOptions } from "../../../../connection.ts"; +import { Service } from "../../../../espConnection.ts"; + +export namespace WsSQL2 { + + export type double = number; + export type unsignedInt = number; + export type long = number; + export type uint64 = number; + + export interface CloseResultSetRequest { + CursorId?: string; + } + + export interface Exception { + Code?: string; + Audience?: string; + Source?: string; + Message?: string; + } + + export interface Exceptions { + Source?: string; + Exception?: Exception[]; + } + + export interface CloseResultSetResponse { + Exceptions?: Exceptions; + Closed?: boolean; + } + + export interface ExecuteQueryRequest { + SqlText?: string; + IncludeRuntimeGraph?: boolean; + IncludeResultSetMetadata?: boolean; + IncludeDebug?: boolean; + } + + export interface Column { + ColumnLabel?: string; + ColumnTypeName?: string; + Precision?: unsignedInt; + Scale?: unsignedInt; + } + + export interface ResultSetMetadata { + Column?: Column[]; + } + + export interface ExecuteQueryResponse { + Exceptions?: Exceptions; + CursorId?: string; + InitMs?: double; + PrepareMs?: double; + ResultSetMetadata?: ResultSetMetadata; + } + + export interface ExplainQueryRequest { + SqlText?: string; + } + + export interface ExplainQueryResponse { + Exceptions?: Exceptions; + Plan?: string; + PlanJson?: string; + ResultSetMetadata?: ResultSetMetadata; + } + + export interface FetchResultSetRequest { + CursorId?: string; + FetchSize?: unsignedInt; + } + + export interface Start { + DuckDBUsedBytes?: long; + DuckDBTempBytes?: long; + HeapInUseBytes?: long; + ResidentBytes?: long; + PeakResidentBytes?: long; + } + + export interface Current { + DuckDBUsedBytes?: long; + DuckDBTempBytes?: long; + HeapInUseBytes?: long; + ResidentBytes?: long; + PeakResidentBytes?: long; + } + + export interface Released { + DuckDBUsedBytes?: long; + DuckDBTempBytes?: long; + HeapInUseBytes?: long; + ResidentBytes?: long; + PeakResidentBytes?: long; + } + + export interface Retained { + DuckDBUsedBytes?: long; + DuckDBTempBytes?: long; + HeapInUseBytes?: long; + ResidentBytes?: long; + PeakResidentBytes?: long; + } + + export interface ScanObjectsCreated { + BindData?: long; + GlobalStates?: long; + LocalStates?: long; + } + + export interface ScanObjectsLive { + BindData?: long; + GlobalStates?: long; + LocalStates?: long; + } + + export interface DebugInfo { + Start?: Start; + Current?: Current; + Released?: Released; + Retained?: Retained; + ScanObjectsCreated?: ScanObjectsCreated; + ScanObjectsLive?: ScanObjectsLive; + LeakCheck?: string; + LeakCheckDetail?: string; + } + + export interface FetchResultSetResponse { + Exceptions?: Exceptions; + Rows?: string; + Complete?: boolean; + ExecuteMs?: double; + RuntimeGraph?: string; + RuntimeGraphJson?: string; + DebugInfo?: DebugInfo; + } + + export interface LogicalFileMetaRequest { + LogicalName?: string; + } + + export interface LogicalFileMeta { + Version?: unsignedInt; + LogicalName?: string; + Kind?: string; + Format?: string; + Grouped?: boolean; + FormatCrc?: unsignedInt; + RecordTranslationMode?: unsignedInt; + Modified?: long; + RecordCount?: uint64; + FormatOptions?: string; + ProviderOptions?: string; + EclRecord?: string; + FileParts?: string; + SampleSQL?: string; + } + + export interface LogicalFileMetaResponse { + Exceptions?: Exceptions; + LogicalFileMeta?: LogicalFileMeta; + } + + export interface WsSQL2PingRequest { + + } + + export interface WsSQL2PingResponse { + + } + + export interface SplitQueryRequest { + SqlText?: string; + } + + export interface Statements { + Statement?: string[]; + } + + export interface SplitQueryResponse { + Exceptions?: Exceptions; + Statements?: Statements; + } + +} + +export class SQL2ServiceBase extends Service { + + constructor(optsConnection: IOptions | IConnection) { + super(optsConnection, "WsSQL2", "1.00"); + } + + CloseResultSet(request: WsSQL2.CloseResultSetRequest, abortSignal?: AbortSignal): Promise { + return this._connection.send("CloseResultSet", request, "json", false, abortSignal, "CloseResultSetResponse"); + } + + ExecuteQuery(request: WsSQL2.ExecuteQueryRequest, abortSignal?: AbortSignal): Promise { + return this._connection.send("ExecuteQuery", request, "json", false, abortSignal, "ExecuteQueryResponse"); + } + + ExplainQuery(request: WsSQL2.ExplainQueryRequest, abortSignal?: AbortSignal): Promise { + return this._connection.send("ExplainQuery", request, "json", false, abortSignal, "ExplainQueryResponse"); + } + + FetchResultSet(request: WsSQL2.FetchResultSetRequest, abortSignal?: AbortSignal): Promise { + return this._connection.send("FetchResultSet", request, "json", false, abortSignal, "FetchResultSetResponse"); + } + + LogicalFileMeta(request: WsSQL2.LogicalFileMetaRequest, abortSignal?: AbortSignal): Promise { + return this._connection.send("LogicalFileMeta", request, "json", false, abortSignal, "LogicalFileMetaResponse"); + } + + Ping(request: WsSQL2.WsSQL2PingRequest, abortSignal?: AbortSignal): Promise { + return this._connection.send("Ping", request, "json", false, abortSignal, "WsSQL2PingResponse"); + } + + SplitQuery(request: WsSQL2.SplitQueryRequest, abortSignal?: AbortSignal): Promise { + return this._connection.send("SplitQuery", request, "json", false, abortSignal, "SplitQueryResponse"); + } + +} diff --git a/packages/comms/src/services/wsdl/WsSQL2/v1.00/WsSQL2.ts b/packages/comms/src/services/wsdl/WsSQL2/v1.00/WsSQL2.ts new file mode 100644 index 0000000000..96946a5b99 --- /dev/null +++ b/packages/comms/src/services/wsdl/WsSQL2/v1.00/WsSQL2.ts @@ -0,0 +1,222 @@ +import { IConnection, IOptions } from "../../../../connection.ts"; +import { Service } from "../../../../espConnection.ts"; + +export namespace WsSQL2 { + + export type double = number; + export type unsignedInt = number; + export type long = number; + export type uint64 = number; + + export interface CloseResultSetRequest { + CursorId?: string; + } + + export interface Exception { + Code?: string; + Audience?: string; + Source?: string; + Message?: string; + } + + export interface Exceptions { + Source?: string; + Exception?: Exception[]; + } + + export interface CloseResultSetResponse { + Exceptions?: Exceptions; + Closed?: boolean; + } + + export interface ExecuteQueryRequest { + SqlText?: string; + IncludeRuntimeGraph?: boolean; + IncludeResultSetMetadata?: boolean; + IncludeDebug?: boolean; + } + + export interface Column { + ColumnLabel?: string; + ColumnTypeName?: string; + Precision?: unsignedInt; + Scale?: unsignedInt; + } + + export interface ResultSetMetadata { + Column?: Column[]; + } + + export interface ExecuteQueryResponse { + Exceptions?: Exceptions; + CursorId?: string; + InitMs?: double; + PrepareMs?: double; + ResultSetMetadata?: ResultSetMetadata; + } + + export interface ExplainQueryRequest { + SqlText?: string; + } + + export interface ExplainQueryResponse { + Exceptions?: Exceptions; + Plan?: string; + PlanJson?: string; + ResultSetMetadata?: ResultSetMetadata; + } + + export interface FetchResultSetRequest { + CursorId?: string; + FetchSize?: unsignedInt; + } + + export interface Start { + DuckDBUsedBytes?: long; + DuckDBTempBytes?: long; + HeapInUseBytes?: long; + ResidentBytes?: long; + PeakResidentBytes?: long; + } + + export interface Current { + DuckDBUsedBytes?: long; + DuckDBTempBytes?: long; + HeapInUseBytes?: long; + ResidentBytes?: long; + PeakResidentBytes?: long; + } + + export interface Released { + DuckDBUsedBytes?: long; + DuckDBTempBytes?: long; + HeapInUseBytes?: long; + ResidentBytes?: long; + PeakResidentBytes?: long; + } + + export interface Retained { + DuckDBUsedBytes?: long; + DuckDBTempBytes?: long; + HeapInUseBytes?: long; + ResidentBytes?: long; + PeakResidentBytes?: long; + } + + export interface ScanObjectsCreated { + BindData?: long; + GlobalStates?: long; + LocalStates?: long; + } + + export interface ScanObjectsLive { + BindData?: long; + GlobalStates?: long; + LocalStates?: long; + } + + export interface DebugInfo { + Start?: Start; + Current?: Current; + Released?: Released; + Retained?: Retained; + ScanObjectsCreated?: ScanObjectsCreated; + ScanObjectsLive?: ScanObjectsLive; + LeakCheck?: string; + LeakCheckDetail?: string; + } + + export interface FetchResultSetResponse { + Exceptions?: Exceptions; + Rows?: string; + Complete?: boolean; + ExecuteMs?: double; + RuntimeGraph?: string; + RuntimeGraphJson?: string; + DebugInfo?: DebugInfo; + } + + export interface LogicalFileMetaRequest { + LogicalName?: string; + } + + export interface LogicalFileMeta { + Version?: unsignedInt; + LogicalName?: string; + Kind?: string; + Format?: string; + Grouped?: boolean; + FormatCrc?: unsignedInt; + RecordTranslationMode?: unsignedInt; + Modified?: long; + RecordCount?: uint64; + FormatOptions?: string; + ProviderOptions?: string; + EclRecord?: string; + FileParts?: string; + SampleSQL?: string; + } + + export interface LogicalFileMetaResponse { + Exceptions?: Exceptions; + LogicalFileMeta?: LogicalFileMeta; + } + + export interface WsSQL2PingRequest { + + } + + export interface WsSQL2PingResponse { + + } + + export interface SplitQueryRequest { + SqlText?: string; + } + + export interface Statements { + Statement?: string[]; + } + + export interface SplitQueryResponse { + Exceptions?: Exceptions; + Statements?: Statements; + } + +} + +export class SQL2ServiceBase extends Service { + + constructor(optsConnection: IOptions | IConnection) { + super(optsConnection, "WsSQL2", "1.00"); + } + + CloseResultSet(request: WsSQL2.CloseResultSetRequest, abortSignal?: AbortSignal): Promise { + return this._connection.send("CloseResultSet", request, "json", false, abortSignal, "CloseResultSetResponse"); + } + + ExecuteQuery(request: WsSQL2.ExecuteQueryRequest, abortSignal?: AbortSignal): Promise { + return this._connection.send("ExecuteQuery", request, "json", false, abortSignal, "ExecuteQueryResponse"); + } + + ExplainQuery(request: WsSQL2.ExplainQueryRequest, abortSignal?: AbortSignal): Promise { + return this._connection.send("ExplainQuery", request, "json", false, abortSignal, "ExplainQueryResponse"); + } + + FetchResultSet(request: WsSQL2.FetchResultSetRequest, abortSignal?: AbortSignal): Promise { + return this._connection.send("FetchResultSet", request, "json", false, abortSignal, "FetchResultSetResponse"); + } + + LogicalFileMeta(request: WsSQL2.LogicalFileMetaRequest, abortSignal?: AbortSignal): Promise { + return this._connection.send("LogicalFileMeta", request, "json", false, abortSignal, "LogicalFileMetaResponse"); + } + + Ping(request: WsSQL2.WsSQL2PingRequest, abortSignal?: AbortSignal): Promise { + return this._connection.send("Ping", request, "json", false, abortSignal, "WsSQL2PingResponse"); + } + + SplitQuery(request: WsSQL2.SplitQueryRequest, abortSignal?: AbortSignal): Promise { + return this._connection.send("SplitQuery", request, "json", false, abortSignal, "SplitQueryResponse"); + } + +} diff --git a/packages/comms/tests/sql2.spec.ts b/packages/comms/tests/sql2.spec.ts new file mode 100644 index 0000000000..2f43db6fd0 --- /dev/null +++ b/packages/comms/tests/sql2.spec.ts @@ -0,0 +1,553 @@ +import { describe, expect, it, vi } from "vitest"; + +import { SQLCleanupError, SQLConnection } from "@hpcc-js/comms"; + +const metadata = { Column: [{ ColumnLabel: "id", ColumnTypeName: "INTEGER" }] }; +const execute = { ExecuteQueryResponse: { CursorId: "cursor-1", ResultSetMetadata: metadata } }; + +describe("SQLConnection", () => { + it("collects labeled rows across pages and keeps native values", async () => { + const columns = { + Column: [ + { ColumnLabel: "ID" }, { ColumnLabel: "active" }, { ColumnLabel: "data" }, { ColumnLabel: "__proto__" } + ] + }; + const actions: string[] = []; + let fetches = 0; + const conn = new SQLConnection({ + baseUrl: "http://localhost:8510", + hookSend: async (_options, action, request: { FetchSize?: number }) => { + actions.push(action); + if (action.endsWith("ExecuteQuery.json")) return { + ExecuteQueryResponse: { CursorId: "cursor-1", ResultSetMetadata: columns } + }; + if (action.endsWith("FetchResultSet.json")) { + expect(request.FetchSize).toBe(1); + return { + FetchResultSetResponse: ++fetches === 1 + ? { Rows: '[[1,true,{"nested":2},"first"]]', Complete: false } + : { Rows: '[[2,null,[3],"second"]]', Complete: true } + }; + } + throw new Error("Completed cursors must not be closed again"); + } + }); + const rows = await conn.queryRows("SELECT * FROM table", 1); + expect(rows[0]).toMatchObject({ ID: 1, active: true, data: { nested: 2 } }); + expect(rows[1]).toMatchObject({ ID: 2, active: null, data: [3] }); + expect(Object.getPrototypeOf(rows[0])).toBe(null); + expect(rows[0]["__proto__"]).toBe("first"); + expect(rows[1]["__proto__"]).toBe("second"); + expect(actions).toEqual(["WsSQL2/ExecuteQuery.json", "WsSQL2/FetchResultSet.json", "WsSQL2/FetchResultSet.json"]); + await conn.close(); + }); + + it("returns no rows for an empty result and rejects unlabeled or duplicate columns", async () => { + const actions: string[] = []; + let sql = ""; + const conn = new SQLConnection({ + baseUrl: "http://localhost:8510", + hookSend: async (_options, action, request: { SqlText?: string }) => { + actions.push(action); + if (action.endsWith("ExecuteQuery.json")) { + sql = request.SqlText!; + return { + ExecuteQueryResponse: { + CursorId: "cursor-1", ResultSetMetadata: { + Column: sql === "missing" ? [{}] + : sql === "duplicate" ? [{ ColumnLabel: "Id" }, { ColumnLabel: "id" }] + : [{ ColumnLabel: "id" }] + } + } + }; + } + if (action.endsWith("FetchResultSet.json")) return { + FetchResultSetResponse: { + Rows: sql === "missing" ? "[[1]]" : sql === "duplicate" ? "[[1,2]]" : "[]", + Complete: sql === "empty" + } + }; + return { CloseResultSetResponse: { Closed: true } }; + } + }); + expect(await conn.queryRows("empty")).toEqual([]); + await expect(conn.queryRows("missing")).rejects.toThrow("missing a label"); + await expect(conn.queryRows("duplicate")).rejects.toThrow("Duplicate SQL column label"); + expect(actions.filter(action => action.endsWith("CloseResultSet.json"))).toHaveLength(2); + await conn.close(); + }); + + it("closes a cursor after a failed fetch and keeps cleanup errors retryable", async () => { + let closes = 0; + const conn = new SQLConnection({ + baseUrl: "http://localhost:8510", + hookSend: async (_options, action) => { + if (action.endsWith("ExecuteQuery.json")) return execute; + if (action.endsWith("FetchResultSet.json")) throw new Error("Fetch failed"); + if (++closes <= 2) throw new Error(`Close failed ${closes}`); + return { CloseResultSetResponse: { Closed: true } }; + } + }); + const failure = await conn.queryRows("SELECT 1").catch(error => error); + expect(failure).toBeInstanceOf(SQLCleanupError); + expect(failure.cause).toBeInstanceOf(SQLCleanupError); + expect(failure.cause.cause).toHaveProperty("message", "Fetch failed"); + expect(failure.cleanupErrors[0]).toHaveProperty("message", "Close failed 2"); + await conn.close(); + expect(closes).toBe(3); + }); + + it("validates query input and closes its temporary statement after execution errors", async () => { + const actions: string[] = []; + const conn = new SQLConnection({ + baseUrl: "http://localhost:8510", + hookSend: async (_options, action) => { + actions.push(action); + throw new Error("Execution failed"); + } + }); + await expect(conn.queryRows("SELECT 1", -1)).rejects.toThrow(RangeError); + await expect(conn.queryRows(" ")).rejects.toThrow("SQL text is required"); + await expect(conn.queryRows("SELECT 1")).rejects.toThrow("Execution failed"); + expect(actions).toEqual(["WsSQL2/ExecuteQuery.json"]); + await conn.close(); + await expect(conn.queryRows("SELECT 1")).rejects.toThrow("closed"); + }); + + it("wraps explain, file metadata and ping while enforcing connection state", async () => { + const hookSend = vi.fn(async (_options, action: string) => { + if (action.endsWith("ExplainQuery.json")) return { ExplainQueryResponse: { Plan: "plan", PlanJson: "{}" } }; + if (action.endsWith("LogicalFileMeta.json")) return { + LogicalFileMetaResponse: { + LogicalFileMeta: { LogicalName: "example::file", RecordCount: 2 } + } + }; + return { WsSQL2PingResponse: {} }; + }); + const conn = new SQLConnection({ baseUrl: "http://localhost:8510", hookSend }); + expect(await conn.explainQuery("SELECT 1")).toMatchObject({ Plan: "plan", PlanJson: "{}" }); + expect(await conn.logicalFileMeta("example::file")).toEqual({ LogicalName: "example::file", RecordCount: 2 }); + expect(await conn.ping()).toBe(true); + expect(hookSend.mock.calls.map(call => call[1])).toEqual([ + "WsSQL2/ExplainQuery.json", "WsSQL2/LogicalFileMeta.json", "WsSQL2/Ping.json" + ]); + expect(hookSend.mock.calls[0][2]).toMatchObject({ SqlText: "SELECT 1" }); + expect(hookSend.mock.calls[1][2]).toMatchObject({ LogicalName: "example::file" }); + await expect(conn.explainQuery(" ")).rejects.toThrow("SQL text is required"); + await expect(conn.logicalFileMeta(" ")).rejects.toThrow("Logical file name is required"); + await conn.close(); + await expect(conn.explainQuery("SELECT 1")).rejects.toThrow("closed"); + await expect(conn.logicalFileMeta("example::file")).rejects.toThrow("closed"); + await expect(conn.ping()).rejects.toThrow("closed"); + }); + + it("propagates explain and metadata service exceptions", async () => { + const conn = new SQLConnection({ + baseUrl: "http://localhost:8510", + hookSend: async (_options, action) => action.endsWith("ExplainQuery.json") + ? { ExplainQueryResponse: { Exceptions: { Exception: [{ Message: "Bad SQL" }] } } } + : { LogicalFileMetaResponse: { Exceptions: { Exception: [{ Message: "Not found" }] } } } + }); + await expect(conn.explainQuery("bad")).rejects.toThrow("Bad SQL"); + await expect(conn.logicalFileMeta("missing")).rejects.toThrow("Not found"); + await conn.close(); + }); + + it("splits SQL using the server parser without executing it", async () => { + const hookSend = vi.fn(async () => ({ + SplitQueryResponse: { Statements: { Statement: ["SELECT 1", "SELECT 'a;b'"] } } + })); + const conn = new SQLConnection({ baseUrl: "http://localhost:8510", hookSend }); + expect(await conn.splitQuery("SELECT 1; SELECT 'a;b';")).toEqual(["SELECT 1", "SELECT 'a;b'"]); + expect(await conn.splitQuery(" ")).toEqual([]); + expect(hookSend).toHaveBeenCalledTimes(1); + expect(hookSend.mock.calls[0][1]).toBe("WsSQL2/SplitQuery.json"); + expect(hookSend.mock.calls[0][2]).toMatchObject({ SqlText: "SELECT 1; SELECT 'a;b';" }); + await conn.close(); + await expect(conn.splitQuery("SELECT 1")).rejects.toThrow("closed"); + }); + + it("reports SplitQuery service failures and malformed responses", async () => { + const conn = new SQLConnection({ + baseUrl: "http://localhost:8510", + hookSend: async () => ({ + SplitQueryResponse: { Exceptions: { Exception: [{ Message: "Invalid SQL" }] } } + }) + }); + await expect(conn.splitQuery("INVALID")).rejects.toThrow("Invalid SQL"); + await conn.close(); + + const malformed = new SQLConnection({ + baseUrl: "http://localhost:8510", + hookSend: async () => ({ SplitQueryResponse: { Statements: { Statement: "SELECT 1" } } }) + }); + await expect(malformed.splitQuery("SELECT 1")).rejects.toThrow("invalid Statements"); + await malformed.close(); + }); + + it("returns no statements for server-recognized comment-only SQL", async () => { + const conn = new SQLConnection({ + baseUrl: "http://localhost:8510", + hookSend: async () => ({ SplitQueryResponse: {} }) + }); + expect(await conn.splitQuery("-- comment")).toEqual([]); + await conn.close(); + }); + + it("lazily fetches pages using Complete and closes an unfinished cursor", async () => { + const hookSend = vi.fn(async (_options, action: string) => { + if (action.endsWith("ExecuteQuery.json")) return execute; + if (action.endsWith("FetchResultSet.json")) return { + FetchResultSetResponse: { Rows: "[[1],[2]]", Complete: false } + }; + return { CloseResultSetResponse: { Closed: true } }; + }); + const conn = new SQLConnection({ baseUrl: "http://localhost:8510", hookSend }); + const stmt = conn.createStatement(); + stmt.setFetchSize(2); + const rows = await stmt.executeQuery("SELECT id FROM some_table"); + expect(rows.getMetaData()).toEqual(metadata.Column); + expect(rows.findColumn("ID")).toBe(1); + expect(hookSend.mock.calls.map(call => call[1])).toEqual(["WsSQL2/ExecuteQuery.json"]); + expect(hookSend.mock.calls[0][2]).toEqual({ + SqlText: "SELECT id FROM some_table", IncludeResultSetMetadata: true, ver_: "1.00" + }); + expect(await rows.next()).toBe(true); + expect(rows.getObject("id")).toBe(1); + expect(await rows.next()).toBe(true); + expect(rows.getString(1)).toBe("2"); + expect(hookSend.mock.calls[1][2]).toMatchObject({ CursorId: "cursor-1", FetchSize: 2 }); + await conn.close(); + expect(hookSend.mock.calls[2][1]).toBe("WsSQL2/CloseResultSet.json"); + expect(hookSend.mock.calls[2][2]).toMatchObject({ CursorId: "cursor-1" }); + expect(() => rows.getObject(1)).toThrow("closed"); + }); + + it("async-disposes result sets, statements, and connections through their cursor owners", async () => { + const actions: string[] = []; + const conn = new SQLConnection({ + baseUrl: "http://localhost:8510", + hookSend: async (_options, action) => { + actions.push(action); + return action.endsWith("ExecuteQuery.json") ? execute : { CloseResultSetResponse: { Closed: true } }; + } + }); + const stmt = conn.createStatement(); + const result = await stmt.executeQuery("SELECT 1"); + await result[Symbol.asyncDispose](); + expect(() => result.next()).toThrow("closed"); + expect(actions.filter(action => action.endsWith("CloseResultSet.json"))).toHaveLength(1); + + const nextResult = await stmt.executeQuery("SELECT 2"); + await stmt[Symbol.asyncDispose](); + expect(() => nextResult.next()).toThrow("closed"); + expect(actions.filter(action => action.endsWith("CloseResultSet.json"))).toHaveLength(2); + + const lastResult = await conn.createStatement().executeQuery("SELECT 3"); + await conn[Symbol.asyncDispose](); + expect(() => lastResult.next()).toThrow("closed"); + await conn[Symbol.asyncDispose](); + expect(actions.filter(action => action.endsWith("CloseResultSet.json"))).toHaveLength(3); + }); + + it("retries a failed cursor disposal without losing ownership", async () => { + let closes = 0; + const conn = new SQLConnection({ + baseUrl: "http://localhost:8510", + hookSend: async (_options, action) => { + if (action.endsWith("ExecuteQuery.json")) return execute; + if (++closes === 1) throw new Error("Close failed"); + return { CloseResultSetResponse: { Closed: true } }; + } + }); + const result = await conn.createStatement().executeQuery("SELECT 1"); + await expect(result[Symbol.asyncDispose]()).rejects.toThrow("Close failed"); + await conn[Symbol.asyncDispose](); + expect(closes).toBe(2); + }); + + it("closes a prepared cursor on await using scope exit", async () => { + const actions: string[] = []; + { + await using conn = new SQLConnection({ + baseUrl: "http://localhost:8510", + hookSend: async (_options, action) => { + actions.push(action); + return action.endsWith("ExecuteQuery.json") ? execute : { CloseResultSetResponse: { Closed: true } }; + } + }); + await conn.createStatement().executeQuery("SELECT 1"); + } + expect(actions).toEqual(["WsSQL2/ExecuteQuery.json", "WsSQL2/CloseResultSet.json"]); + }); + + it("consumes final-page rows without another fetch or close request", async () => { + const actions: string[] = []; + const conn = new SQLConnection({ + baseUrl: "http://localhost:8510", + hookSend: async (_options, action) => { + actions.push(action); + if (action.endsWith("ExecuteQuery.json")) return execute; + return { FetchResultSetResponse: { Rows: "[[1],[2]]", Complete: true } }; + } + }); + const rows = await conn.createStatement().executeQuery("SELECT 1"); + expect(await rows.next()).toBe(true); + expect(rows.getObject(1)).toBe(1); + expect(await rows.next()).toBe(true); + expect(rows.getObject(1)).toBe(2); + expect(await rows.next()).toBe(false); + await conn.close(); + expect(actions).toEqual(["WsSQL2/ExecuteQuery.json", "WsSQL2/FetchResultSet.json"]); + }); + + it("fetches through empty intermediate pages", async () => { + let fetches = 0; + const conn = new SQLConnection({ + baseUrl: "http://localhost:8510", + hookSend: async (_options, action) => action.endsWith("ExecuteQuery.json") + ? execute + : { + FetchResultSetResponse: ++fetches === 1 + ? { Rows: "[]", Complete: false } + : { Rows: "[[42]]", Complete: true } + } + }); + const rows = await conn.createStatement().executeQuery("SELECT 42"); + expect(await rows.next()).toBe(true); + expect(rows.getObject(1)).toBe(42); + expect(await rows.next()).toBe(false); + expect(fetches).toBe(2); + await conn.close(); + }); + + it("closes the cursor if a fetch fails or returns malformed data", async () => { + for (const invalid of [new Error("Fetch failed"), { FetchResultSetResponse: { Rows: "bad json", Complete: false } }]) { + const actions: string[] = []; + const conn = new SQLConnection({ + baseUrl: "http://localhost:8510", + hookSend: async (_options, action) => { + actions.push(action); + if (action.endsWith("ExecuteQuery.json")) return execute; + if (action.endsWith("FetchResultSet.json")) { + if (invalid instanceof Error) throw invalid; + return invalid; + } + return { CloseResultSetResponse: { Closed: true } }; + } + }); + const rows = await conn.createStatement().executeQuery("SELECT 1"); + await expect(rows.next()).rejects.toThrow(); + expect(actions).toEqual([ + "WsSQL2/ExecuteQuery.json", "WsSQL2/FetchResultSet.json", "WsSQL2/CloseResultSet.json" + ]); + await conn.close(); + } + }); + + it("closes the cursor when the execute metadata is invalid", async () => { + const actions: string[] = []; + const conn = new SQLConnection({ + baseUrl: "http://localhost:8510", + hookSend: async (_options, action) => { + actions.push(action); + return action.endsWith("ExecuteQuery.json") + ? { ExecuteQueryResponse: { CursorId: "cursor-1" } } + : { CloseResultSetResponse: { Closed: true } }; + } + }); + await expect(conn.createStatement().executeQuery("SELECT 1")).rejects.toThrow("metadata"); + expect(actions).toEqual(["WsSQL2/ExecuteQuery.json", "WsSQL2/CloseResultSet.json"]); + await conn.close(); + }); + + it("closes a previous result before executing again", async () => { + const actions: string[] = []; + const conn = new SQLConnection({ + baseUrl: "http://localhost:8510", + hookSend: async (_options, action) => { + actions.push(action); + return action.endsWith("ExecuteQuery.json") ? execute : { CloseResultSetResponse: { Closed: true } }; + } + }); + const stmt = conn.createStatement(); + const first = await stmt.executeQuery("SELECT 1"); + await stmt.executeQuery("SELECT 2"); + expect(() => first.getMetaData()).toThrow("closed"); + expect(actions).toEqual([ + "WsSQL2/ExecuteQuery.json", "WsSQL2/CloseResultSet.json", "WsSQL2/ExecuteQuery.json" + ]); + await conn.close(); + expect(actions.at(-1)).toBe("WsSQL2/CloseResultSet.json"); + }); + + it("retries a failed result close without losing the cursor", async () => { + const cursorIds: string[] = []; + let releaseClose: (response: object) => void; + const pendingClose = new Promise(resolve => { releaseClose = resolve; }); + const conn = new SQLConnection({ + baseUrl: "http://localhost:8510", + hookSend: async (_options, action, request: { CursorId?: string }) => { + if (action.endsWith("ExecuteQuery.json")) return execute; + cursorIds.push(request.CursorId!); + return cursorIds.length === 1 ? pendingClose : { CloseResultSetResponse: { Closed: true } }; + } + }); + const rows = await conn.createStatement().executeQuery("SELECT 1"); + const closing = rows.close(); + expect(rows.close()).toBe(closing); + releaseClose!({ CloseResultSetResponse: { Exceptions: { Exception: [{ Message: "Temporary failure" }] } } }); + await expect(closing).rejects.toThrow("Temporary failure"); + await rows.close(); + expect(cursorIds).toEqual(["cursor-1", "cursor-1"]); + await conn.close(); + expect(cursorIds).toHaveLength(2); + }); + + it("retries failed statement and connection cleanup without losing ownership", async () => { + const closeAttempts = new Map(); + const conn = new SQLConnection({ + baseUrl: "http://localhost:8510", + hookSend: async (_options, action, request: { CursorId?: string; SqlText?: string }) => { + if (action.endsWith("ExecuteQuery.json")) return { + ExecuteQueryResponse: { CursorId: request.SqlText, ResultSetMetadata: metadata } + }; + const id = request.CursorId!; + closeAttempts.set(id, (closeAttempts.get(id) ?? 0) + 1); + if (id === "SELECT 1" && closeAttempts.get(id) === 1) throw new Error("Network error"); + return { CloseResultSetResponse: { Closed: true } }; + } + }); + const first = conn.createStatement(); + const second = conn.createStatement(); + await first.executeQuery("SELECT 1"); + await second.executeQuery("SELECT 2"); + await expect(conn.close()).rejects.toThrow("Failed to close SQL statements"); + await conn.close(); + expect([...closeAttempts]).toEqual([["SELECT 1", 2], ["SELECT 2", 1]]); + await conn.close(); + expect(closeAttempts.get("SELECT 1")).toBe(2); + }); + + it("retains a cursor after cleanup fails for invalid execute metadata", async () => { + let closeAttempts = 0; + const conn = new SQLConnection({ + baseUrl: "http://localhost:8510", + hookSend: async (_options, action) => { + if (action.endsWith("ExecuteQuery.json")) return { ExecuteQueryResponse: { CursorId: "cursor-1" } }; + ++closeAttempts; + if (closeAttempts === 1) throw new Error("Close failed"); + return { CloseResultSetResponse: { Closed: true } }; + } + }); + const stmt = conn.createStatement(); + await expect(stmt.executeQuery("SELECT 1")).rejects.toThrow("Failed to parse result and close cursor"); + await stmt.close(); + expect(closeAttempts).toBe(2); + await conn.close(); + }); + + it("closes immediately during a pending fetch to cancel server execution", async () => { + let releaseFetch: (response: object) => void; + const fetch = new Promise(resolve => { releaseFetch = resolve; }); + const actions: string[] = []; + const conn = new SQLConnection({ + baseUrl: "http://localhost:8510", + hookSend: async (_options, action) => { + actions.push(action); + if (action.endsWith("ExecuteQuery.json")) return execute; + if (action.endsWith("FetchResultSet.json")) return fetch; + return { CloseResultSetResponse: { Closed: true } }; + } + }); + const rows = await conn.createStatement().executeQuery("SELECT 1"); + const pending = rows.next(); + await conn.close(); + expect(actions).toEqual([ + "WsSQL2/ExecuteQuery.json", "WsSQL2/FetchResultSet.json", "WsSQL2/CloseResultSet.json" + ]); + releaseFetch!({ FetchResultSetResponse: { Rows: "[[1]]", Complete: false } }); + await expect(pending).rejects.toThrow("closed"); + expect(actions).toHaveLength(3); + }); + + it("cancels a prepared cursor before the first fetch", async () => { + const actions: string[] = []; + const conn = new SQLConnection({ + baseUrl: "http://localhost:8510", + hookSend: async (_options, action) => { + actions.push(action); + return action.endsWith("ExecuteQuery.json") ? execute : { CloseResultSetResponse: { Closed: true } }; + } + }); + const rows = await conn.createStatement().executeQuery("SELECT 1"); + await rows.cancel(); + expect(actions).toEqual(["WsSQL2/ExecuteQuery.json", "WsSQL2/CloseResultSet.json"]); + expect(() => rows.next()).toThrow("closed"); + await conn.close(); + }); + + it("cancels via the statement without closing the statement", async () => { + const actions: string[] = []; + const conn = new SQLConnection({ + baseUrl: "http://localhost:8510", + hookSend: async (_options, action) => { + actions.push(action); + return action.endsWith("ExecuteQuery.json") ? execute : { CloseResultSetResponse: { Closed: true } }; + } + }); + const stmt = conn.createStatement(); + const rows = await stmt.executeQuery("SELECT 1"); + await stmt.cancel(); + expect(() => rows.next()).toThrow("closed"); + await stmt.executeQuery("SELECT 2"); + await conn.close(); + expect(actions).toEqual([ + "WsSQL2/ExecuteQuery.json", "WsSQL2/CloseResultSet.json", + "WsSQL2/ExecuteQuery.json", "WsSQL2/CloseResultSet.json" + ]); + }); + + it("waits for an in-flight execute before connection close", async () => { + let release: (response: object) => void; + const pendingResponse = new Promise(resolve => { release = resolve; }); + let started: () => void; + const executing = new Promise(resolve => { started = resolve; }); + const actions: string[] = []; + const conn = new SQLConnection({ + baseUrl: "http://localhost:8510", + hookSend: async (_options, action) => { + actions.push(action); + if (action.endsWith("ExecuteQuery.json")) { + started(); + return pendingResponse; + } + return { CloseResultSetResponse: { Closed: true } }; + } + }); + const query = conn.createStatement().executeQuery("SELECT 1"); + await executing; + const closing = conn.close(); + release!(execute); + await expect(query).rejects.toThrow("closed"); + await closing; + expect(actions).toEqual(["WsSQL2/ExecuteQuery.json", "WsSQL2/CloseResultSet.json"]); + }); + + it("rejects invalid fetch sizes and column lookups", async () => { + const conn = new SQLConnection({ + baseUrl: "http://localhost:8510", + hookSend: async (_options, action) => action.endsWith("ExecuteQuery.json") + ? execute + : { FetchResultSetResponse: { Rows: "[[1]]", Complete: true } } + }); + const stmt = conn.createStatement(); + expect(() => stmt.setFetchSize(-1)).toThrow(RangeError); + const rows = await stmt.executeQuery("SELECT 1"); + expect(() => rows.getObject(1)).toThrow("not positioned"); + expect(await rows.next()).toBe(true); + expect(() => rows.getObject(2)).toThrow(RangeError); + expect(() => rows.getObject("missing")).toThrow("Unknown SQL column"); + await conn.close(); + }); +}); diff --git a/packages/comms/tests/wsLogaccess.spec.ts b/packages/comms/tests/wsLogaccess.spec.ts index d2ea4af62a..d5b5a76d83 100644 --- a/packages/comms/tests/wsLogaccess.spec.ts +++ b/packages/comms/tests/wsLogaccess.spec.ts @@ -45,13 +45,14 @@ describe("LogaccessService.GetLogsEx (ELK stack)", () => { expect(result.lines).toHaveLength(0); }); - it("returns empty lines and zero total when LogLines JSON is malformed", async () => { + it("rejects with a contextual error when LogLines JSON is malformed", async () => { const svc = new LogaccessService({ baseUrl: "http://localhost" }); vi.spyOn(svc, "GetLogAccessInfo").mockResolvedValue(elkInfo as WsLogaccess.GetLogAccessInfoResponse); vi.spyOn(svc, "GetLogs").mockResolvedValue({ LogLines: "NOT VALID JSON", TotalLogLinesAvailable: 50 } as any); - const result = await svc.GetLogsEx({ LogLineStartFrom: 0, LogLineLimit: 100 }); - expect(result.lines).toHaveLength(0); - expect(result.total).toBe(0); + const error: any = await svc.GetLogsEx({ LogLineStartFrom: 0, LogLineLimit: 100 }).then(() => undefined, e => e); + expect(error).toBeInstanceOf(Error); + expect(error.message).toContain("Error parsing log lines json"); + expect(error.cause).toBeInstanceOf(SyntaxError); }); }); diff --git a/packages/comms/tests/wsSQL2.spec.ts b/packages/comms/tests/wsSQL2.spec.ts new file mode 100644 index 0000000000..0598b5981d --- /dev/null +++ b/packages/comms/tests/wsSQL2.spec.ts @@ -0,0 +1,184 @@ +import { describe, expect, it, vi } from "vitest"; + +import { DFUArrayActions, DFUService, SQL2Service, SQLConnection, type WsSQL2, Workunit } from "@hpcc-js/comms"; +import { ESP_URL } from "./testLib.ts"; + +const SQL_URL = "http://127.0.0.1:8510/"; + +async function supportsSQL2(): Promise { + const connection = new SQLConnection({ baseUrl: SQL_URL }); + try { + return await connection.ping(); + } catch { + return false; + } finally { + await connection.close(); + } +} + +const hasSQL2 = await supportsSQL2(); + +describe.skipIf(!hasSQL2)("WsSQL2", () => { + it("routes typed requests and unwraps responses", async () => { + const hookSend = vi.fn(async (_options, action: string) => action.endsWith("ExecuteQuery.json") + ? { ExecuteQueryResponse: { CursorId: "cursor-1", ResultSetMetadata: { Column: [] } } } + : { FetchResultSetResponse: { Rows: "[]", Complete: true } }); + const service = new SQL2Service({ baseUrl: "http://localhost:8510", hookSend }); + const query: WsSQL2.ExecuteQueryRequest = { SqlText: "SELECT 1", IncludeResultSetMetadata: true }; + + expect(service.connection().service()).toBe("WsSQL2"); + expect(service.connection().version()).toBe("1.00"); + expect(await service.ExecuteQuery(query)).toEqual({ CursorId: "cursor-1", ResultSetMetadata: { Column: [] } }); + expect(hookSend).toHaveBeenCalledWith( + expect.anything(), "WsSQL2/ExecuteQuery.json", { ...query, ver_: "1.00" }, "json", expect.anything(), undefined + ); + + expect(await service.FetchResultSet({ CursorId: "cursor-1", FetchSize: 10 })).toEqual({ Rows: "[]", Complete: true }); + expect(hookSend).toHaveBeenLastCalledWith( + expect.anything(), "WsSQL2/FetchResultSet.json", { CursorId: "cursor-1", FetchSize: 10, ver_: "1.00" }, "json", expect.anything(), undefined + ); + }); + + it("exposes the renamed plan and runtime graph response fields", async () => { + const service = new SQL2Service({ + baseUrl: "http://localhost:8510", + hookSend: async (_options, action) => action.endsWith("ExplainQuery.json") + ? { ExplainQueryResponse: { Plan: "plan", PlanJson: "{}" } } + : { FetchResultSetResponse: { Rows: "[]", Complete: true, RuntimeGraph: "graph", RuntimeGraphJson: "{}" } } + }); + expect(await service.ExplainQuery({ SqlText: "SELECT 1" })).toMatchObject({ + Plan: "plan", PlanJson: "{}" + }); + expect(await service.FetchResultSet({ CursorId: "cursor-1" })).toMatchObject({ + Complete: true, RuntimeGraph: "graph", RuntimeGraphJson: "{}" + }); + }); + + it("routes SplitQuery and preserves its statement response", async () => { + const hookSend = vi.fn(async () => ({ + SplitQueryResponse: { Statements: { Statement: ["SELECT 1", "SELECT 'a;b'"] } } + })); + const service = new SQL2Service({ baseUrl: "http://localhost:8510", hookSend }); + const request: WsSQL2.SplitQueryRequest = { SqlText: "SELECT 1; SELECT 'a;b';" }; + expect(await service.SplitQuery(request)).toEqual({ + Statements: { Statement: ["SELECT 1", "SELECT 'a;b'"] } + }); + expect(hookSend).toHaveBeenCalledWith( + expect.anything(), "WsSQL2/SplitQuery.json", { ...request, ver_: "1.00" }, "json", expect.anything(), undefined + ); + }); + + describe("integration", () => { + it("splits a multi-statement script without splitting quoted semicolons", async () => { + const connection = new SQLConnection({ baseUrl: SQL_URL }); + try { + expect(await connection.splitQuery("SELECT 1; SELECT 'a;b';")).toEqual([ + "SELECT 1", "SELECT 'a;b'" + ]); + expect(await connection.splitQuery("-- comment")).toEqual([]); + } finally { + await connection.close(); + } + }); + + it("pages through a 1000-row logical file and removes its workunit and file", async () => { + const fileName = `hpcc-js::test::sql2::${Date.now().toString(36)}${Math.random().toString(36).slice(2)}`; + const dfu = new DFUService({ baseUrl: ESP_URL }); + let workunit: Workunit | undefined; + let connection: SQLConnection | undefined; + let fetches = 0; + let closedCursors = 0; + let failure: unknown; + try { + workunit = await Workunit.create({ baseUrl: ESP_URL }); + await workunit.update({ + QueryText: `rowLayout := RECORD + UNSIGNED4 id; + STRING label; +END; +testRows := DATASET(1000, TRANSFORM(rowLayout, + SELF.id := COUNTER; + SELF.label := (STRING) COUNTER; +)); +OUTPUT(testRows,, '~${fileName}', OVERWRITE);` + }); + await workunit.submit("hthor"); + await workunit.watchUntilComplete(); + if (workunit.isFailed()) { + await workunit.fetchInfo({ IncludeExceptions: true }); + throw new Error(`Workunit ${workunit.Wuid} failed: ${JSON.stringify(workunit.Exceptions)}`); + } + const createdFiles = await dfu.DFUQuery({ LogicalName: fileName }); + expect(createdFiles.DFULogicalFiles?.DFULogicalFile?.some(file => file.Name === fileName)).toBe(true); + + connection = new SQLConnection({ + baseUrl: SQL_URL, + hookSend: (options, action, request, responseType, defaultSend, header) => { + if (action === "WsSQL2/FetchResultSet.json") ++fetches; + if (action === "WsSQL2/CloseResultSet.json") ++closedCursors; + return defaultSend(options, action, request, responseType, header); + } + }); + const statement = connection.createStatement(); + statement.setFetchSize(73); + const result = await statement.executeQuery(`SELECT id, label FROM hpcc_scan('${fileName}') ORDER BY id`); + expect(fetches).toBe(0); + let count = 0; + while (await result.next()) { + ++count; + expect(result.getObject("id")).toBe(count); + expect(result.getString("label")).toBe(String(count)); + } + expect(count).toBe(1000); + expect(fetches).toBeGreaterThan(1); + await result.close(); + const earlyResult = await statement.executeQuery(`SELECT id FROM hpcc_scan('${fileName}') ORDER BY id`); + expect(await earlyResult.next()).toBe(true); + await earlyResult.close(); + expect(closedCursors).toBe(1); + const pendingResult = await statement.executeQuery(`SELECT id FROM hpcc_scan('${fileName}') ORDER BY id`); + const fetchesBeforeCancel = fetches; + await pendingResult.cancel(); + expect(fetches).toBe(fetchesBeforeCancel); + expect(closedCursors).toBe(2); + } catch (error) { + failure = error; + } finally { + const cleanup = await Promise.allSettled([ + connection?.close(), + workunit ? (async () => { + const files = await dfu.DFUQuery({ LogicalName: fileName }); + if (files.DFULogicalFiles?.DFULogicalFile?.some(file => file.Name === fileName)) { + const response = await dfu.DFUArrayAction({ + Type: DFUArrayActions.Delete, + LogicalFiles: { Item: [fileName] } + }); + const failures = response.ActionResults?.DFUActionInfo?.filter(action => action.Failed) ?? []; + if (failures.length) throw new Error(`Failed to remove logical file ${fileName}: ${JSON.stringify(failures)}`); + const remaining = await dfu.DFUQuery({ LogicalName: fileName }); + if (remaining.DFULogicalFiles?.DFULogicalFile?.some(file => file.Name === fileName)) { + throw new Error(`Logical file ${fileName} still exists after deletion`); + } + } + })() : undefined + ].filter((operation): operation is Promise => operation !== undefined)); + const errors = cleanup.filter((result): result is PromiseRejectedResult => result.status === "rejected"); + if (workunit) { + try { + await workunit.delete(); + expect(workunit.isDeleted()).toBe(true); + } catch (error) { + errors.push({ status: "rejected", reason: error }); + } + } + if (errors.length) { + failure = new AggregateError( + [...(failure === undefined ? [] : [failure]), ...errors.map(error => error.reason)], + "SQL2 integration test or cleanup failed" + ); + } + } + if (failure !== undefined) throw failure; + }, 120000); + }); +}); diff --git a/packages/other/src/Comms.ts b/packages/other/src/Comms.ts index d32df742d5..dc8a6a81c0 100644 --- a/packages/other/src/Comms.ts +++ b/packages/other/src/Comms.ts @@ -318,7 +318,12 @@ export class Comms extends ESPUrl { const xhr: any = new XMLHttpRequest(); xhr.onload = function (e) { if (this.status >= 200 && this.status < 300) { - resolve(JSON.parse(this.response)); + try { + resolve(JSON.parse(this.response)); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + reject(Object.assign(new Error(`Error parsing response JSON '${message}'`), { cause: error })); + } } else { reject(Error(this.statusText)); }