[FIXUP] pass abort signal deeper

Author: mcheshkov

packages/cubejs-clickhouse-driver/src/ClickHouseDriver.ts

@@ -170,7 +170,7 @@ export class ClickHouseDriver extends BaseDriver implements DriverInterface {
     this.client = this.createClient(maxPoolSize);
   }
 
-  protected withCancel<T>(fn: (con: ClickHouseClient, queryId: string) => Promise<T>): Promise<T> {
+  protected withCancel<T>(fn: (con: ClickHouseClient, queryId: string, signal: AbortSignal) => Promise<T>): Promise<T> {
     console.log("withCancel call");
     const queryId = uuidv4();
 
@@ -180,12 +180,10 @@ export class ClickHouseDriver extends BaseDriver implements DriverInterface {
     const promise = (async () => {
       await this.client.ping();
       signal.throwIfAborted();
-      // TODO pass signal deeper, new driver supports abort signal, but does not do autokill
-      const result = await fn(this.client, queryId);
+      const result = await fn(this.client, queryId, signal);
       signal.throwIfAborted();
       return result;
     })();
-    // TODO why do we need this?
     (promise as any).cancel = async () => {
       abortController.abort();
       // Use separate client for kill query, usual pool may be busy
@@ -269,6 +267,7 @@ export class ClickHouseDriver extends BaseDriver implements DriverInterface {
         query_id: queryId,
         format: 'JSON',
         clickhouse_settings: this.config.clickhouseSettings,
+        abort_signal: signal,
       });
       console.log("queryResponse resultSet", query, resultSet.query_id, resultSet.response_headers);
 
@@ -652,20 +651,24 @@ export class ClickHouseDriver extends BaseDriver implements DriverInterface {
 
   // This is not part of a driver interface, and marked public only for testing
   public async command(query: string): Promise<void> {
-    await this.withCancel(async (connection) => {
+    await this.withCancel(async (connection, queryId, signal) => {
       await connection.command({
         query,
+        query_id: queryId,
+        abort_signal: signal,
       });
     });
   }
 
   // This is not part of a driver interface, and marked public only for testing
   public async insert(table: string, values: Array<Array<unknown>>): Promise<void> {
-    await this.withCancel(async (connection) => {
+    await this.withCancel(async (connection, queryId, signal) => {
       await connection.insert({
         table,
         values,
         format: 'JSONCompactEachRow',
+        query_id: queryId,
+        abort_signal: signal,
       });
     });
   }