[WIP] feat(clickhouse-driver): Switch from apla-clickhouse to @clickhouse/client

Author: mcheshkov

packages/cubejs-clickhouse-driver/package.json

@@ -27,7 +27,7 @@
     "integration:clickhouse": "jest dist/test"
   },
   "dependencies": {
-    "@cubejs-backend/apla-clickhouse": "^1.7",
+    "@clickhouse/client": "^1.7.0",
     "@cubejs-backend/base-driver": "1.1.2",
     "@cubejs-backend/shared": "1.1.2",
     "generic-pool": "^3.6.0",

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

@@ -21,14 +21,14 @@ import {
   TableStructure,
   UnloadOptions,
 } from '@cubejs-backend/base-driver';
+
+import { ClickHouseClient, createClient } from '@clickhouse/client';
 import genericPool, { Pool } from 'generic-pool';
 import { v4 as uuidv4 } from 'uuid';
 import sqlstring from 'sqlstring';
 
 import { HydrationStream, transformRow } from './HydrationStream';
 
-const ClickHouse = require('@cubejs-backend/apla-clickhouse');
-
 const ClickhouseTypeToGeneric: Record<string, string> = {
   enum: 'text',
   string: 'text',
@@ -86,7 +86,7 @@ export class ClickHouseDriver extends BaseDriver implements DriverInterface {
     return 5;
   }
 
-  protected readonly pool: Pool<any>;
+  protected readonly pool: Pool<ClickHouseClient>;
 
   protected readonly readOnlyMode: boolean;
 
@@ -122,19 +122,33 @@ export class ClickHouseDriver extends BaseDriver implements DriverInterface {
       config.dataSource ||
       assertDataSource('default');
 
+    // TODO recheck everything in config for new driver
+    const host = getEnv('dbHost', { dataSource });
+    const port = getEnv('dbPort', { dataSource });
+    const protocol = getEnv('dbSsl', { dataSource }) ? 'https:' : 'http:';
+    // TODO proper value here, with proper back compat, and treating protocol
+    const url = `${protocol}//${host}:${port}`;
+    // TODO drop this
+    console.log('ClickHouseDriver will use url', url);
+
+    const username = getEnv('dbUser', { dataSource });
+    const password = getEnv('dbPass', { dataSource });
     this.config = {
-      host: getEnv('dbHost', { dataSource }),
-      port: getEnv('dbPort', { dataSource }),
-      auth:
-        getEnv('dbUser', { dataSource }) ||
-        getEnv('dbPass', { dataSource })
-          ? `${
-            getEnv('dbUser', { dataSource })
-          }:${
-            getEnv('dbPass', { dataSource })
-          }`
-          : '',
-      protocol: getEnv('dbSsl', { dataSource }) ? 'https:' : 'http:',
+      // host: getEnv('dbHost', { dataSource }),
+      // port: getEnv('dbPort', { dataSource }),
+      url,
+      // auth:
+      //   getEnv('dbUser', { dataSource }) ||
+      //   getEnv('dbPass', { dataSource })
+      //     ? `${
+      //       getEnv('dbUser', { dataSource })
+      //     }:${
+      //       getEnv('dbPass', { dataSource })
+      //     }`
+      //     : '',
+      username,
+      password,
+      // protocol: getEnv('dbSsl', { dataSource }) ? 'https:' : 'http:',
       queryOptions: {
         database:
           getEnv('dbName', { dataSource }) ||
@@ -148,10 +162,17 @@ export class ClickHouseDriver extends BaseDriver implements DriverInterface {
     this.readOnlyMode =
       getEnv('clickhouseReadOnly', { dataSource }) === 'true';
 
+    // TODO @clickhouse/client have internal pool, that does NOT guarantee same connection, and can break with temp tables. Disable it?
     this.pool = genericPool.createPool({
-      create: async () => new ClickHouse({
+      create: async () => createClient({
         ...this.config,
-        queryOptions: {
+
+        username: getEnv('dbUser', { dataSource }),
+        password: getEnv('dbPass', { dataSource }),
+
+        database: this.config.queryOptions.database,
+        session_id: uuidv4(),
+        clickhouse_settings: {
           //
           //
           // If ClickHouse user's permissions are restricted with "readonly = 1",
@@ -160,9 +181,7 @@ export class ClickHouseDriver extends BaseDriver implements DriverInterface {
           //
           //
           ...(this.readOnlyMode ? {} : { join_use_nulls: 1 }),
-          session_id: uuidv4(),
-          ...this.config.queryOptions,
-        }
+        },
       }),
       destroy: () => Promise.resolve()
     }, {
@@ -176,36 +195,43 @@ export class ClickHouseDriver extends BaseDriver implements DriverInterface {
       idleTimeoutMillis: 30000,
       acquireTimeoutMillis: 20000
     });
+
+    // https://github.com/coopernurse/node-pool/blob/ee5db9ddb54ce3a142fde3500116b393d4f2f755/README.md#L220-L226
+    this.pool.on('factoryCreateError', (err) => {
+      this.databasePoolError(err);
+    });
+    this.pool.on('factoryDestroyError', (err) => {
+      this.databasePoolError(err);
+    });
   }
 
-  protected withConnection(fn: (con: any, queryId: string) => Promise<any>) {
+  protected withConnection(fn: (con: ClickHouseClient, queryId: string) => Promise<any>) {
+    console.log("withConnection call");
     const self = this;
     const connectionPromise = this.pool.acquire();
     const queryId = uuidv4();
 
     let cancelled = false;
     const cancelObj: any = {};
 
-    const promise: any = connectionPromise.then((connection: any) => {
+    const promise: any = connectionPromise.then((connection: ClickHouseClient) => {
+      console.log("withConnection got connection");
       cancelObj.cancel = async () => {
         cancelled = true;
         await self.withConnection(async conn => {
-          await conn.querying(`KILL QUERY WHERE query_id = '${queryId}'`);
+          await conn.command({
+            query: `KILL QUERY WHERE query_id = '${queryId}'`,
+          });
         });
       };
       return fn(connection, queryId)
-        .then(res => this.pool.release(connection).then(() => {
+        .finally(() => this.pool.release(connection))
+        .then((res) => {
           if (cancelled) {
             throw new Error('Query cancelled');
           }
           return res;
-        }))
-        .catch((err) => this.pool.release(connection).then(() => {
-          if (cancelled) {
-            throw new Error('Query cancelled');
-          }
-          throw err;
-        }));
+        });
     });
     promise.cancel = () => cancelObj.cancel();
 
@@ -227,22 +253,48 @@ export class ClickHouseDriver extends BaseDriver implements DriverInterface {
   }
 
   protected queryResponse(query: string, values: unknown[]) {
+    console.log("queryResponse call", query);
+
     const formattedQuery = sqlstring.format(query, values);
 
-    return this.withConnection((connection, queryId) => connection.querying(formattedQuery, {
-      dataObjects: true,
-      queryOptions: {
+    return this.withConnection(async (connection, queryId) => {
+      // if (formattedQuery.startsWith("CREATE TABLE")) {
+      //
+      // }
+
+      const resultSet = await connection.query({
+        query: formattedQuery,
         query_id: queryId,
-        //
-        //
-        // If ClickHouse user's permissions are restricted with "readonly = 1",
-        // change settings queries are not allowed. Thus, "join_use_nulls" setting
-        // can not be changed
-        //
-        //
-        ...(this.readOnlyMode ? {} : { join_use_nulls: 1 }),
+        clickhouse_settings: {
+          //
+          //
+          // If ClickHouse user's permissions are restricted with "readonly = 1",
+          // change settings queries are not allowed. Thus, "join_use_nulls" setting
+          // can not be changed
+          //
+          //
+          ...(this.readOnlyMode ? {} : { join_use_nulls: 1 }),
+        },
+      });
+      console.log("queryResponse resultSet", query, resultSet.query_id, resultSet.response_headers);
+
+      if (resultSet.response_headers['x-clickhouse-format'] !== 'JSON') {
+        const results = await resultSet.text();
+        console.log("queryResponse text results", query, results);
+        console.log("queryResponse text results JSON", JSON.stringify(results));
+        return [];
+      } else {
+        const results = await resultSet.json();
+        console.log("queryResponse json results", query, results);
+        console.log("queryResponse json results JSON", JSON.stringify(results));
+        return results;
       }
-    }));
+
+      //       'content-type': 'text/plain; charset=UTF-8',
+      // vs
+      //       'content-type': 'application/json; charset=UTF-8',
+      //       'x-clickhouse-format': 'JSON',
+    });
   }
 
   protected normaliseResponse(res: any) {

yarn.lock

@@ -4222,6 +4222,18 @@
   resolved "https://registry.yarnpkg.com/@bcoe/v8-coverage/-/v8-coverage-0.2.3.tgz#75a2e8b51cb758a7553d6804a5932d7aace75c39"
   integrity sha512-0hYQ8SB4Db5zvZB4axdMHGwEaQjkZzFjQiN9LVYvIFB2nSUHW9tYpxWriPrWDASIxiaXax83REcLxuSdnGPZtw==
 
+"@clickhouse/[email protected]":
+  version "1.7.0"
+  resolved "https://registry.yarnpkg.com/@clickhouse/client-common/-/client-common-1.7.0.tgz#4d0315158d275ea8d55ed8e04d69871832f4d8ba"
+  integrity sha512-RkHYf23/wyv/6C0KcVD4nRX4JAn/Y+9AZBQPlrSId2JwXsmAnjDkkKpuPLwZPNVH6J3BkW+y8bQCEk3VHQzArw==
+
+"@clickhouse/client@^1.7.0":
+  version "1.7.0"
+  resolved "https://registry.yarnpkg.com/@clickhouse/client/-/client-1.7.0.tgz#a6b7b72db825162b1f54c2fe383f349dbf437fbd"
+  integrity sha512-2aESIFRbSPWEZIU++sXt1RYWgEKZH75C3jyXLcRBeafMDjq7bKV2AX1X9n9xscN+Y4VvnkBzkjFxcbuqFSBk6w==
+  dependencies:
+    "@clickhouse/client-common" "1.7.0"
+
 "@codemirror/highlight@^0.19.0":
   version "0.19.6"
   resolved "https://registry.yarnpkg.com/@codemirror/highlight/-/highlight-0.19.6.tgz#7f2e066f83f5649e8e0748a3abe0aaeaf64b8ac2"
@@ -4353,7 +4365,7 @@
     tiny-invariant "^1.3.3"
     valid-url "^1.0.9"
 
-"@cubejs-backend/apla-clickhouse@^1.7", "@cubejs-backend/apla-clickhouse@^1.7.0":
+"@cubejs-backend/apla-clickhouse@^1.7.0":
   version "1.7.0"
   resolved "https://registry.yarnpkg.com/@cubejs-backend/apla-clickhouse/-/apla-clickhouse-1.7.0.tgz#6359f46c56492d1704d18be0210c7546fdac5f5e"
   integrity sha512-qwXapTC/qosA6RprElRjnl8gmlDQaxtJPtbgcdjyNvkmiyao1HI+w5QkjHWCiVm6aTzE0gjFr6/2y87TZ9fojg==