Switch ClickHouse driver

package.json

@@ -37,10 +37,10 @@
     "license": "MIT",
     "dependencies": {
         "@google-cloud/bigquery": "^5.5.0",
+        "@posthog/clickhouse": "^1.7.0",
         "@sentry/node": "^5.29.0",
         "@sentry/tracing": "^5.29.0",
         "adm-zip": "^0.4.16",
-        "clickhouse": "^2.2.1",
         "fastify": "^3.8.0",
         "hot-shots": "^8.2.1",
         "ioredis": "^4.19.2",

src/db.ts

@@ -1,9 +1,8 @@
 import { Properties } from '@posthog/plugin-scaffold'
-import { ClickHouse } from 'clickhouse'
+import ClickHouse from '@posthog/clickhouse'
 import { Producer } from 'kafkajs'
 import { DateTime } from 'luxon'
 import { Pool, QueryConfig, QueryResult, QueryResultRow } from 'pg'
-import { string } from 'yargs'
 import { KAFKA_PERSON, KAFKA_PERSON_UNIQUE_ID } from './ingestion/topics'
 import { chainToElements, hashElements, unparsePersonPartial } from './ingestion/utils'
 import {
@@ -45,14 +44,17 @@ export class DB {
         queryTextOrConfig: string | QueryConfig<I>,
         values?: I
     ): Promise<QueryResult<R>> {
-        return this.postgres.query(queryTextOrConfig, values)
+        return await this.postgres.query(queryTextOrConfig, values)
     }
 
-    public async clickhouseQuery(query: string, reqParams?: Record<string, any>): Promise<Record<string, any>> {
+    public async clickhouseQuery(
+        query: string,
+        options?: ClickHouse.QueryOptions
+    ): Promise<ClickHouse.QueryResult<Record<string, any>>> {
         if (!this.clickhouse) {
             throw new Error('ClickHouse connection has not been provided to this DB instance!')
         }
-        return this.clickhouse.query(query, reqParams).toPromise()
+        return await this.clickhouse.querying(query, options)
     }
 
     // Person
@@ -61,7 +63,7 @@ export class DB {
     public async fetchPersons(database: Database.ClickHouse): Promise<ClickHousePerson[]>
     public async fetchPersons(database: Database = Database.Postgres): Promise<Person[] | ClickHousePerson[]> {
         if (database === Database.ClickHouse) {
-            return (await this.clickhouseQuery('SELECT * FROM person')) as ClickHousePerson[]
+            return (await this.clickhouseQuery('SELECT * FROM person')).data as ClickHousePerson[]
         } else if (database === Database.Postgres) {
             return ((await this.postgresQuery('SELECT * FROM posthog_person')).rows as RawPerson[]).map(
                 (rawPerson: RawPerson) =>
@@ -186,11 +188,13 @@ export class DB {
         database: Database = Database.Postgres
     ): Promise<PersonDistinctId[] | ClickHousePersonDistinctId[]> {
         if (database === Database.ClickHouse) {
-            return (await this.clickhouseQuery(
-                `SELECT * FROM person_distinct_id WHERE person_id='${escapeClickHouseString(
-                    person.uuid
-                )}' and team_id='${person.team_id}' ORDER BY id`
-            )) as ClickHousePersonDistinctId[]
+            return (
+                await this.clickhouseQuery(
+                    `SELECT * FROM person_distinct_id WHERE person_id='${escapeClickHouseString(
+                        person.uuid
+                    )}' and team_id='${person.team_id}' ORDER BY id`
+                )
+            ).data as ClickHousePersonDistinctId[]
         } else if (database === Database.Postgres) {
             const result = await this.postgresQuery(
                 'SELECT * FROM posthog_persondistinctid WHERE person_id=$1 and team_id=$2 ORDER BY id',
@@ -255,7 +259,7 @@ export class DB {
 
     public async fetchEvents(): Promise<Event[] | ClickHouseEvent[]> {
         if (this.kafkaProducer) {
-            const events = (await this.clickhouseQuery(`SELECT * FROM events`)) as ClickHouseEvent[]
+            const events = (await this.clickhouseQuery(`SELECT * FROM events`)).data as ClickHouseEvent[]
             return (
                 events?.map(
                     (event) =>
@@ -278,9 +282,8 @@ export class DB {
 
     public async fetchSessionRecordingEvents(): Promise<PostgresSessionRecordingEvent[] | SessionRecordingEvent[]> {
         if (this.kafkaProducer) {
-            const events = ((await this.clickhouseQuery(
-                `SELECT * FROM session_recording_events`
-            )) as SessionRecordingEvent[]).map((event) => {
+            const events = ((await this.clickhouseQuery(`SELECT * FROM session_recording_events`))
+                .data as SessionRecordingEvent[]).map((event) => {
                 return {
                     ...event,
                     snapshot_data: event.snapshot_data ? JSON.parse(event.snapshot_data) : null,
@@ -297,9 +300,11 @@ export class DB {
 
     public async fetchElements(event?: Event): Promise<Element[]> {
         if (this.kafkaProducer) {
-            const events = (await this.clickhouseQuery(
-                `SELECT elements_chain FROM events WHERE uuid='${escapeClickHouseString((event as any).uuid)}'`
-            )) as ClickHouseEvent[]
+            const events = (
+                await this.clickhouseQuery(
+                    `SELECT elements_chain FROM events WHERE uuid='${escapeClickHouseString((event as any).uuid)}'`
+                )
+            ).data as ClickHouseEvent[]
             const chain = events?.[0]?.elements_chain
             return chainToElements(chain)
         } else {

src/ingestion/process-event.ts

@@ -16,7 +16,7 @@ import { Event as EventProto, IEvent } from '../idl/protos'
 import { Producer } from 'kafkajs'
 import { KAFKA_EVENTS, KAFKA_SESSION_RECORDING_EVENTS } from './topics'
 import { elementsToString, sanitizeEventName } from './utils'
-import { ClickHouse } from 'clickhouse'
+import ClickHouse from '@posthog/clickhouse'
 import { DB } from '../db'
 import { status } from '../status'
 import * as Sentry from '@sentry/node'

src/server.ts

@@ -5,7 +5,7 @@ import { Kafka, logLevel, Producer } from 'kafkajs'
 import { FastifyInstance } from 'fastify'
 import { PluginsServer, PluginsServerConfig, Queue } from './types'
 import { startQueue } from './worker/queue'
-import { ClickHouse } from 'clickhouse'
+import ClickHouse from '@posthog/clickhouse'
 import { startFastifyInstance, stopFastifyInstance } from './web/server'
 import { version } from '../package.json'
 import { PluginEvent } from '@posthog/plugin-scaffold'
@@ -20,6 +20,7 @@ import { startSchedule } from './services/schedule'
 import { ConnectionOptions } from 'tls'
 import { DB } from './db'
 import { DateTime } from 'luxon'
+import * as fs from 'fs'
 import { KAFKA_EVENTS_PLUGIN_INGESTION, KAFKA_EVENTS_WAL } from './ingestion/topics'
 
 export async function createServer(
@@ -71,17 +72,19 @@ export async function createServer(
             throw new Error('You must set KAFKA_HOSTS to process events from Kafka!')
         }
         clickhouse = new ClickHouse({
-            url: `http${serverConfig.CLICKHOUSE_SECURE ? 's' : ''}://$${serverConfig.CLICKHOUSE_HOST}`,
+            host: serverConfig.CLICKHOUSE_HOST,
             port: serverConfig.CLICKHOUSE_SECURE ? 8443 : 8123,
-            basicAuth: {
-                username: serverConfig.CLICKHOUSE_USER,
-                password: serverConfig.CLICKHOUSE_PASSWORD,
-            },
-            config: {
+            protocol: serverConfig.CLICKHOUSE_SECURE ? 'https:' : 'http:',
+            user: serverConfig.CLICKHOUSE_USER,
+            password: serverConfig.CLICKHOUSE_PASSWORD || undefined,
+            dataObjects: true,
+            queryOptions: {
                 database: serverConfig.CLICKHOUSE_DATABASE,
+                output_format_json_quote_64bit_integers: false,
             },
+            ca: serverConfig.CLICKHOUSE_CA ? fs.readFileSync(serverConfig.CLICKHOUSE_CA).toString() : undefined,
         })
-        await clickhouse.query('SELECT 1') // test that the connection works
+        await clickhouse.querying('SELECT 1') // test that the connection works
 
         if (!serverConfig.KAFKA_CONSUMPTION_TOPIC) {
             // When ingesting events, listen to the "INGESTION_HANDOFF" topic, otherwise listen to the "WAL" and discard

src/types.ts

@@ -6,7 +6,7 @@ import { VM } from 'vm2'
 import { DateTime } from 'luxon'
 import { StatsD } from 'hot-shots'
 import { EventsProcessor } from 'ingestion/process-event'
-import { ClickHouse } from 'clickhouse'
+import ClickHouse from '@posthog/clickhouse'
 import { DB } from './db'
 
 export enum LogLevel {

tests/helpers/clickhouse.ts

@@ -1,22 +1,24 @@
 import { defaultConfig } from '../../src/config'
-import { ClickHouse } from 'clickhouse'
+import ClickHouse from '@posthog/clickhouse'
 import { PluginsServerConfig } from '../../src/types'
 
 export async function resetTestDatabaseClickhouse(extraServerConfig: Partial<PluginsServerConfig>): Promise<void> {
     const config = { ...defaultConfig, ...extraServerConfig }
     const clickhouse = new ClickHouse({
-        url: `http://$${config.CLICKHOUSE_HOST}`,
+        host: config.CLICKHOUSE_HOST,
         port: 8123,
-        config: {
+        dataObjects: true,
+        queryOptions: {
             database: config.CLICKHOUSE_DATABASE,
+            output_format_json_quote_64bit_integers: false,
         },
     })
-    await clickhouse.query('TRUNCATE events').toPromise()
-    await clickhouse.query('TRUNCATE events_mv').toPromise()
-    await clickhouse.query('TRUNCATE person').toPromise()
-    await clickhouse.query('TRUNCATE person_distinct_id').toPromise()
-    await clickhouse.query('TRUNCATE person_mv').toPromise()
-    await clickhouse.query('TRUNCATE person_static_cohort').toPromise()
-    await clickhouse.query('TRUNCATE session_recording_events').toPromise()
-    await clickhouse.query('TRUNCATE session_recording_events_mv').toPromise()
+    await clickhouse.querying('TRUNCATE events')
+    await clickhouse.querying('TRUNCATE events_mv')
+    await clickhouse.querying('TRUNCATE person')
+    await clickhouse.querying('TRUNCATE person_distinct_id')
+    await clickhouse.querying('TRUNCATE person_mv')
+    await clickhouse.querying('TRUNCATE person_static_cohort')
+    await clickhouse.querying('TRUNCATE session_recording_events')
+    await clickhouse.querying('TRUNCATE session_recording_events_mv')
 }