[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,31 @@ 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}`;
+
+    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 +160,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 +179,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 +193,41 @@ 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>) {
     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) => {
       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();
 
@@ -229,10 +251,10 @@ export class ClickHouseDriver extends BaseDriver implements DriverInterface {
   protected queryResponse(query: string, values: unknown[]) {
     const formattedQuery = sqlstring.format(query, values);
 
-    return this.withConnection((connection, queryId) => connection.querying(formattedQuery, {
-      dataObjects: true,
-      queryOptions: {
-        query_id: queryId,
+    return this.withConnection((connection, queryId) => connection.query({
+      query: formattedQuery,
+      query_id: queryId,
+      clickhouse_settings: {
         //
         //
         // If ClickHouse user's permissions are restricted with "readonly = 1",
@@ -241,7 +263,7 @@ export class ClickHouseDriver extends BaseDriver implements DriverInterface {
         //
         //
         ...(this.readOnlyMode ? {} : { join_use_nulls: 1 }),
-      }
+      },
     }));
   }
 

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==