feat: queryStream method

jest.config.js

@@ -1,8 +1,10 @@
 module.exports = {
   preset: 'ts-jest',
   testEnvironment: 'node',
+  collectCoverage: true,
   coverageDirectory: 'coverage',
   collectCoverageFrom: ['packages/**/src/**/*.{ts,tsx,js,jsx}', 'packages/**/!src/**/*.d.ts'],
+  coverageReporters: ['text-summary', 'json', 'lcov'],
   globals: {
     'ts-jest': {
       tsConfig: 'tsconfig.json',

packages/clickhouse-adapter/.mdeprc

@@ -3,8 +3,8 @@
   "node": "12.14.0",
   "nycCoverage": false,
   "coverage": false,
-  "test_framework": "jest --verbose -i -b -c ../../jest.config.js --coverageDirectory <coverageDirectory>",
-  "tests": "../clickhouse-adapter/test/suites/**/*.ts",
+  "test_framework": "jest --verbose -i -b -c ../../jest.config.js --runInBand --coverageDirectory <coverageDirectory>",
+  "tests": "../clickhouse-adapter/test/suites",
   "auto_compose": true,
   "with_local_compose": true,
   "docker_compose": "./test/docker-compose.yml",

packages/clickhouse-adapter/src/client.ts

@@ -1,25 +1,36 @@
 import ClickHouse from '@apla/clickhouse'
 import { Promise } from 'bluebird'
 import { merge } from 'lodash'
+
 import {
   InsertData,
-  ClickhouseOptions,
+  Options,
   TableBuilder,
   ClickhouseClientInterface,
-  QueryOptions,
+  QueryStream,
+  ClientConfig,
+  Callback,
 } from './interfaces'
 
 export class ClickhouseClient implements ClickhouseClientInterface {
-  public static readonly defaultOpts: ClickhouseOptions = {
+  public static readonly defaultOpts: Options = {
     host: 'clickhouse',
   }
 
   public readonly connection: ClickHouse
 
-  public readonly queryAsync: (query: string, options: QueryOptions) => Promise<any>
+  public readonly queryAsync: (query: string, options?: Options) => Promise<any>
+
+  private readonly database: string
+  private readonly options: Options
+
+  constructor(config: ClientConfig) {
+    const { dbName, ...options } = config
+
+    this.options = merge({}, ClickhouseClient.defaultOpts, options)
+    this.database = dbName
 
-  constructor(options: ClickhouseOptions) {
-    this.connection = new ClickHouse(merge({}, ClickhouseClient.defaultOpts, options))
+    this.connection = new ClickHouse(this.options)
 
     this.queryAsync = Promise.promisify(this.query, { context: this })
   }
@@ -32,17 +43,20 @@ export class ClickhouseClient implements ClickhouseClientInterface {
   public insert(
     dbName: string,
     insertData: InsertData,
-    options: QueryOptions,
+    options: Options,
     cb: (err: any, result: any) => void
   ): void
   public insert(dbName: string, insertData: InsertData, arg1: any, arg2?: any): void {
-    const queryOptions: QueryOptions = typeof arg1 === 'object' ? arg1 : {}
+    const options: Options = typeof arg1 === 'object' ? arg1 : {}
     const stream = this.connection.query(
       insertData.query(),
       {
-        ...queryOptions,
-        format: queryOptions.format || 'JSONEachRow',
-        queryOptions: { database: dbName },
+        ...options,
+        format: options.format || 'JSONEachRow',
+        queryOptions: {
+          ...options.queryOptions,
+          database: dbName,
+        },
       },
       typeof arg1 === 'function' ? arg1 : arg2
     )
@@ -56,11 +70,41 @@ export class ClickhouseClient implements ClickhouseClientInterface {
     stream.end()
   }
 
-  public query(query: string, options: QueryOptions, cb: (err: any, result: any) => void): void {
-    this.connection.query(
-      query,
-      { syncParser: true, ...options, format: options.format || 'JSONCompact' },
-      cb
-    )
+  public query(query: string, options?: Options, cb?: Callback): void {
+    // promisify passes callback in 2nd argument if no arg provided
+    const callback = options.constructor === Function ? options : cb
+
+    const opts = {
+      ...this.getQueryOptions(cb ? options : {}),
+      syncParser: true,
+    }
+
+    this.connection.query(query, opts, callback)
+  }
+
+  public queryStream(query: string, options?: Options, cb?: Callback): QueryStream {
+    const opts = {
+      ...this.getQueryOptions(options),
+      syncParser: false,
+    }
+
+    return this.connection.query(query, opts, cb)
+  }
+
+  private getQueryOptions(options?: Options): Record<string, any> {
+    return {
+      ...options,
+      syncParser: true,
+      format: this.getFormat(options, 'JSONCompact'),
+      queryOptions: {
+        database: this.database,
+        ...options?.queryOptions,
+        ...this.options.queryOptions,
+      },
+    }
+  }
+
+  private getFormat(options: Options | undefined, defaultFormat: string): string {
+    return options?.format || this.options.format || defaultFormat
   }
 }

packages/clickhouse-adapter/src/interfaces/clickhouseClientInterface.ts

@@ -1,7 +1,9 @@
 import ClickHouse from '@apla/clickhouse'
 import { TableBuilder } from './tableBuilder'
 import { InsertData } from './insertData'
-import { QueryOptions } from './queryOptions'
+import { ClickhouseSettings } from './clickhouseSettings'
+
+export type Callback = (err: Error | undefined, res: any) => void
 
 export interface ClickhouseClientInterface {
   connection: ClickHouse
@@ -10,7 +12,7 @@ export interface ClickhouseClientInterface {
   insert(
     dbName: string,
     insertData: InsertData,
-    options: QueryOptions,
+    options: ClickhouseSettings,
     cb: (err: any, result: any) => void
   ): void
   insert(dbName: string, insertData: InsertData, arg1: any, arg2: any): void

packages/clickhouse-adapter/src/interfaces/clickhouseOptions.ts

@@ -1,17 +1,12 @@
 import { RequestOptions } from 'http'
 import { Formats } from '../types'
+import { ClickhouseSettings } from './clickhouseSettings'
 
-export interface ClickhouseOptions {
-  readonly host: string
-  readonly user?: string
-  readonly password?: string
-  readonly dbName?: string
-  readonly path?: string
-  readonly port?: number
-  readonly protocol?: 'https:' | 'http:'
+export interface Options extends RequestOptions {
   readonly dataObjects?: boolean
   readonly format?: Formats
-  readonly queryOptions?: any
+  readonly omitFormat?: boolean
+  readonly queryOptions?: ClickhouseSettings
   readonly readonly?: boolean
-  readonly requestOptions?: RequestOptions
+  readonly syncParser?: boolean
 }

packages/clickhouse-adapter/src/interfaces/clickhouseSettings.ts

@@ -0,0 +1,12 @@
+const allowDDlValue = [0, 1] as const
+
+/**
+ * See https://clickhouse.tech/docs/en/operations/settings/
+ */
+export interface ClickhouseSettings {
+  readonly database: string
+  readonly readonly?: boolean
+  readonly allow_ddl?: typeof allowDDlValue[number]
+  readonly profile?: string
+  readonly [key: string]: any
+}

packages/clickhouse-adapter/src/interfaces/clientConfig.ts

@@ -0,0 +1,11 @@
+import { Options } from './clickhouseOptions'
+
+export interface ClientConfig extends Options {
+  readonly host: string
+  readonly user?: string
+  readonly password?: string
+  readonly dbName?: string
+  readonly path?: string
+  readonly port?: number
+  readonly protocol?: 'https:' | 'http:'
+}

packages/clickhouse-adapter/src/interfaces/index.ts

@@ -6,4 +6,6 @@ export * from './table-spec'
 export * from './clickhouseOptions'
 export * from './tableBuilder'
 export * from './clickhouseClientInterface'
-export * from './queryOptions'
+export * from './clickhouseSettings'
+export * from './clientConfig'
+export * from './queryStream'

packages/clickhouse-adapter/src/interfaces/queryStream.ts

@@ -0,0 +1,17 @@
+import type { Duplex } from 'stream'
+
+export interface SupplementalInformation {
+  rows: number
+  statistics: {
+    elapsed: number
+    rows_read: number
+    bytes_read: number
+  }
+}
+
+export interface QueryStream extends Duplex {
+  /**
+   * Available only for SELECT queries with JSON* format
+   */
+  supplemental?: SupplementalInformation
+}

packages/clickhouse-adapter/src/migrators/system.ts

@@ -12,7 +12,7 @@ export class SystemMigrator {
   }
 
   public async up(dbName: string): Promise<void> {
-    await this.ch.queryAsync(createDb(dbName), {
+    await this.ch.connection.querying(createDb(dbName), {
       format: 'TabSeparated',
     })
 

packages/clickhouse-adapter/test/suites/01-create-table.test.ts

@@ -31,7 +31,7 @@ describe('create table', () => {
       host: 'ch1',
     })
 
-    await client.connection.querying('CREATE DATABASE IF NOT EXISTS db_test')
+    await client.connection.querying(`CREATE DATABASE IF NOT EXISTS db_test on cluster '{cluster}'`)
 
     await client.createTable(
       new TableMaker('db_test', 'test_replicated_table', `'{cluster}'`, {