[WIP] feat(clickhouse): Use ClickHouse query with parameters via HTTP interface

Author: mcheshkov

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

@@ -230,6 +230,15 @@ export class ClickHouseDriver extends BaseDriver implements DriverInterface {
     });
   }
 
+  protected prepareParams(values: unknown[]): Record<string, string> {
+    return Object.fromEntries(values.map((value, idx) => {
+      const paramName = this.paramName(idx);
+      const paramKey = `param_${paramName}`;
+      const preparedValue = formatQueryParams(value);
+      return [paramKey, preparedValue];
+    }));
+  }
+
   protected queryResponse(query: string, values?: unknown[]) {
     // todo drop this
     console.log('queryResponse call', query, values);
@@ -251,12 +260,7 @@ export class ClickHouseDriver extends BaseDriver implements DriverInterface {
     // https://clickhouse.com/docs/en/interfaces/cli#cli-queries-with-parameters
     // https://clickhouse.com/docs/en/interfaces/http#cli-queries-with-parameters
 
-    const paramsValues = Object.fromEntries((values ?? []).map((value, idx) => {
-      const paramName = this.paramName(idx);
-      const paramKey = `param_${paramName}`;
-      const preparedValue = formatQueryParams(value);
-      return [paramKey, preparedValue];
-    }));
+    const paramsValues = this.prepareParams(values ?? []);
 
     // todo drop this
     console.log('queryResponse prepared', formattedQuery, paramsValues);
@@ -359,11 +363,19 @@ export class ClickHouseDriver extends BaseDriver implements DriverInterface {
     // eslint-disable-next-line @typescript-eslint/no-unused-vars
     { highWaterMark }: StreamOptions
   ): Promise<StreamTableDataWithTypes> {
+    // todo drop this
+    console.log('stream call', query, values);
+
     // eslint-disable-next-line no-underscore-dangle
     const conn = await (<any> this.pool)._factory.create();
 
     try {
-      const formattedQuery = sqlstring.format(query, values);
+      // const formattedQuery = sqlstring.format(query, values);
+      const formattedQuery = query;
+      const paramsValues = this.prepareParams(values ?? []);
+
+      // todo drop this
+      console.log('stream prepared', formattedQuery, paramsValues);
 
       return await new Promise((resolve, reject) => {
         const options = {
@@ -377,12 +389,15 @@ export class ClickHouseDriver extends BaseDriver implements DriverInterface {
             //
             //
             ...(this.readOnlyMode ? {} : { join_use_nulls: 1 }),
+
+            // Add parameter values to query string
+            ...paramsValues,
           }
         };
 
         const originalStream = conn.query(formattedQuery, options, (err: Error | null, result: any) => {
           if (err) {
-            reject(err);
+            reject(new Error(`Failed during stream createion; query: ${query}; values: ${values}`, { cause: err }));
           } else {
             const rowStream = new HydrationStream(result.meta);
             originalStream.pipe(rowStream);

packages/cubejs-schema-compiler/package.json

@@ -57,6 +57,7 @@
     "uuid": "^8.3.2"
   },
   "devDependencies": {
+    "@clickhouse/client-common": "^1.8.0",
     "@cubejs-backend/apla-clickhouse": "^1.7.0",
     "@cubejs-backend/linter": "^1.0.0",
     "@cubejs-backend/query-orchestrator": "1.1.3",

packages/cubejs-schema-compiler/test/integration/clickhouse/ClickHouseDbRunner.js

@@ -1,3 +1,4 @@
+import { formatQueryParams } from '@clickhouse/client-common';
 /* eslint-disable */
 import ClickHouse from '@cubejs-backend/apla-clickhouse';
 import { GenericContainer } from 'testcontainers';
@@ -136,13 +137,27 @@ export class ClickHouseDbRunner extends BaseDbRunner {
     } : {};
 
     for (const [query, params] of queries) {
-      requests.push(clickHouse.querying(formatSql(query, params), {
+      // TODO get rid of sqlstring
+      // TODO migrate to upstream clickhouse client
+      const preparedQ = query;
+      const preparedV = Object.fromEntries(params.map((value, idx) => {
+        const paramName = `p${idx}`;
+        const paramKey = `param_${paramName}`;
+        const preparedValue = formatQueryParams(value);
+        return [paramKey, preparedValue];
+      }));
+      console.log("ClickHOuse testQueries prepared", preparedQ, preparedV);
+      requests.push(clickHouse.querying(preparedQ, {
         dataObjects: true,
         queryOptions: {
           session_id: clickHouse.sessionId,
           join_use_nulls: '1',
-          ...extendedDateTimeResultsOptions
+          ...extendedDateTimeResultsOptions,
+          // Add parameter values to query string
+          ...preparedV,
         }
+      }).catch((err) => {
+        throw new Error(`Failed during query; query: ${query}; params: ${params}`, { cause: err });
       }));
     }