From 96dd103b0100e15578af4895985e71d76b96f4b3 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Thu, 4 Feb 2021 01:05:52 +0100 Subject: [PATCH 01/10] create postgres parity tests, fix some bugs --- src/db.ts | 36 +++++++---- src/types.ts | 5 ++ tests/clickhouse/postgres-parity.test.ts | 76 ++++++++++++++++++++++++ tests/helpers/sql.ts | 10 +++- tests/shared/process-event.ts | 10 +--- 5 files changed, 114 insertions(+), 23 deletions(-) create mode 100644 tests/clickhouse/postgres-parity.test.ts diff --git a/src/db.ts b/src/db.ts index 98f635dc..9aeb247a 100644 --- a/src/db.ts +++ b/src/db.ts @@ -1,5 +1,5 @@ import { Properties } from '@posthog/plugin-scaffold' -import { ClickHouse, QueryCursor } from 'clickhouse' +import { ClickHouse } from 'clickhouse' import { Producer } from 'kafkajs' import { DateTime } from 'luxon' import { Pool, QueryConfig, QueryResult, QueryResultRow } from 'pg' @@ -7,16 +7,18 @@ import { string } from 'yargs' import { KAFKA_PERSON, KAFKA_PERSON_UNIQUE_ID } from './ingestion/topics' import { chainToElements, hashElements, unparsePersonPartial } from './ingestion/utils' import { + ClickHouseEvent, + Database, + Element, + ElementGroup, + Event, Person, PersonDistinctId, - RawPerson, - RawOrganization, PostgresSessionRecordingEvent, - Event, - ClickHouseEvent, - Element, + RawOrganization, + RawPerson, SessionRecordingEvent, - ElementGroup, + TimestampFormat, } from './types' import { castTimestampOrNow, clickHouseTimestampToISO, sanitizeSqlIdentifier } from './utils' @@ -53,9 +55,15 @@ export class DB { // Person - public async fetchPersons(): Promise { - const result = await this.postgresQuery('SELECT * FROM posthog_person') - return result.rows as Person[] + public async fetchPersons(database: Database = Database.Postgres): Promise { + if (database === Database.ClickHouse) { + return (await this.clickhouseQuery('SELECT * FROM person')) as Person[] + } else if (database === Database.Postgres) { + const result = await this.postgresQuery('SELECT * FROM posthog_person') + return result.rows as Person[] + } else { + throw new Error(`Can't fetch persons for database: ${database}`) + } } public async fetchPerson(teamId: number, distinctId: string): Promise { @@ -95,7 +103,7 @@ export class DB { const personCreated = insertResult.rows[0] as Person if (this.kafkaProducer) { const data = { - created_at: castTimestampOrNow(createdAt), + created_at: castTimestampOrNow(createdAt, TimestampFormat.ClickHouse), properties: JSON.stringify(properties), team_id: teamId, is_identified: isIdentified, @@ -125,7 +133,7 @@ export class DB { ) if (this.kafkaProducer) { const data = { - created_at: castTimestampOrNow(updatedPerson.created_at), + created_at: castTimestampOrNow(updatedPerson.created_at, TimestampFormat.ClickHouse), properties: JSON.stringify(updatedPerson.properties), team_id: updatedPerson.team_id, is_identified: updatedPerson.is_identified, @@ -167,7 +175,9 @@ export class DB { if (this.kafkaProducer) { await this.kafkaProducer.send({ topic: KAFKA_PERSON_UNIQUE_ID, - messages: [{ value: Buffer.from(JSON.stringify(personDistinctIdCreated)) }], + messages: [ + { value: Buffer.from(JSON.stringify({ ...personDistinctIdCreated, person_id: person.uuid })) }, + ], }) } } diff --git a/src/types.ts b/src/types.ts index 97bfef79..5076f3fe 100644 --- a/src/types.ts +++ b/src/types.ts @@ -322,3 +322,8 @@ export enum TimestampFormat { ClickHouse = 'clickhouse', ISO = 'iso', } + +export enum Database { + ClickHouse = 'clickhouse', + Postgres = 'postgres', +} diff --git a/tests/clickhouse/postgres-parity.test.ts b/tests/clickhouse/postgres-parity.test.ts new file mode 100644 index 00000000..6bf4208d --- /dev/null +++ b/tests/clickhouse/postgres-parity.test.ts @@ -0,0 +1,76 @@ +import { Database, LogLevel, PluginsServer, PluginsServerConfig, Team } from '../../src/types' +import { getFirstTeam, resetTestDatabase } from '../helpers/sql' +import { startPluginsServer } from '../../src/server' +import { makePiscina } from '../../src/worker/piscina' +import { createPosthog, DummyPostHog } from '../../src/extensions/posthog' +import { pluginConfig39 } from '../helpers/plugins' +import { UUIDT } from '../../src/utils' +import { resetTestDatabaseClickhouse } from '../helpers/clickhouse' +import { resetKafka } from '../helpers/kafka' +import { delayUntilEventIngested } from '../shared/process-event' +import { DateTime } from 'luxon' + +jest.setTimeout(60000) // 60 sec timeout + +const extraServerConfig: Partial = { + KAFKA_ENABLED: true, + KAFKA_HOSTS: process.env.KAFKA_HOSTS || 'kafka:9092', + WORKER_CONCURRENCY: 2, + PLUGIN_SERVER_INGESTION: true, + LOG_LEVEL: LogLevel.Log, +} + +describe('postgres parity', () => { + let server: PluginsServer + let stopServer: () => Promise + let posthog: DummyPostHog + let team: Team + + beforeAll(async () => { + await resetKafka(extraServerConfig) + }) + + beforeEach(async () => { + await resetTestDatabase(` + async function processEvent (event) { + event.properties.processed = 'hell yes' + event.properties.upperUuid = event.properties.uuid?.toUpperCase() + return event + } + `) + await resetTestDatabaseClickhouse(extraServerConfig) + const startResponse = await startPluginsServer(extraServerConfig, makePiscina) + server = startResponse.server + stopServer = startResponse.stop + posthog = createPosthog(server, pluginConfig39) + team = await getFirstTeam(server) + }) + + afterEach(async () => { + await stopServer() + }) + + test('createPerson does the same in both databases', async () => { + const person = server.db.createPerson( + DateTime.utc(), + { userProp: 'propValue' }, + team.id, + null, + true, + new UUIDT().toString(), + ['distinct1', 'distinct2'] + ) + await delayUntilEventIngested(() => server.db.fetchPersons(Database.ClickHouse)) + + const clickHousePersons = await server.db.fetchPersons(Database.ClickHouse) + const postgresPersons = await server.db.fetchPersons(Database.Postgres) + + expect(clickHousePersons.length).toEqual(postgresPersons.length) + }) + + // test('createPerson', async () => {}) + // test('updatePerson', async () => {}) + // test('deletePerson', async () => {}) + // test('addDistinctId', async () => {}) + // test('updateDistinctId', async () => {}) +}) diff --git a/tests/helpers/sql.ts b/tests/helpers/sql.ts index 4c1aef36..35464c0e 100644 --- a/tests/helpers/sql.ts +++ b/tests/helpers/sql.ts @@ -2,7 +2,7 @@ import { makePluginObjects, commonOrganizationId, commonUserId, commonOrganizati import { defaultConfig } from '../../src/config' import { Pool } from 'pg' import { delay, UUIDT } from '../../src/utils' -import { PluginsServerConfig } from '../../src/types' +import { PluginsServer, PluginsServerConfig, Team } from '../../src/types' export async function resetTestDatabase( code: string, @@ -113,3 +113,11 @@ export async function createUserTeamAndOrganization( is_demo: false, }) } + +export async function getTeams(server: PluginsServer): Promise { + return (await server.db.postgresQuery('SELECT * FROM posthog_team ORDER BY id')).rows +} + +export async function getFirstTeam(server: PluginsServer): Promise { + return (await getTeams(server))[0] +} diff --git a/tests/shared/process-event.ts b/tests/shared/process-event.ts index b9a4c058..249b67ab 100644 --- a/tests/shared/process-event.ts +++ b/tests/shared/process-event.ts @@ -12,7 +12,7 @@ import { ClickHouseEvent, SessionRecordingEvent, } from '../../src/types' -import { createUserTeamAndOrganization, resetTestDatabase } from '../helpers/sql' +import { createUserTeamAndOrganization, getFirstTeam, getTeams, resetTestDatabase } from '../helpers/sql' import { EventsProcessor } from '../../src/ingestion/process-event' import { DateTime } from 'luxon' import { delay, UUIDT } from '../../src/utils' @@ -21,14 +21,6 @@ import { hashElements } from '../../src/ingestion/utils' jest.setTimeout(600000) // 600 sec timeout -async function getTeams(server: PluginsServer): Promise { - return (await server.db.postgresQuery('SELECT * FROM posthog_team ORDER BY id')).rows -} - -async function getFirstTeam(server: PluginsServer): Promise { - return (await getTeams(server))[0] -} - export async function delayUntilEventIngested(fetchEvents: () => Promise, minCount = 1): Promise { for (let i = 0; i < 30; i++) { if ((await fetchEvents()).length >= minCount) { From 24bad9b8b850470de0854a29259021cd0b661e78 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Thu, 4 Feb 2021 01:32:47 +0100 Subject: [PATCH 02/10] create all topics --- tests/helpers/kafka.ts | 40 +++++++++++++++++++++------------------- 1 file changed, 21 insertions(+), 19 deletions(-) diff --git a/tests/helpers/kafka.ts b/tests/helpers/kafka.ts index 5195e85c..d993b375 100644 --- a/tests/helpers/kafka.ts +++ b/tests/helpers/kafka.ts @@ -3,6 +3,9 @@ import { Kafka, Consumer, logLevel, EachMessagePayload, Producer } from 'kafkajs import { KAFKA_EVENTS, KAFKA_EVENTS_INGESTION_HANDOFF, + KAFKA_EVENTS_WAL, + KAFKA_PERSON, + KAFKA_PERSON_UNIQUE_ID, KAFKA_SESSION_RECORDING_EVENTS, } from '../../src/ingestion/topics' import { parseRawEventMessage } from '../../src/ingestion/utils' @@ -127,25 +130,14 @@ export async function resetKafka(extraServerConfig: Partial }) const messages = [] - async function createTopic(topic: string) { - try { - const admin = kafka.admin() - await admin.connect() - await admin.createTopics({ - waitForLeaders: true, - topics: [{ topic }], - }) - await admin.disconnect() - } catch (e) { - if (!e?.error?.includes('Topic with this name already exists')) { - console.error(`Error creating kafka topic "${topic}". This might be fine!`) - console.error(e) - } - } - } - - await createTopic(KAFKA_EVENTS_INGESTION_HANDOFF) - await createTopic(KAFKA_SESSION_RECORDING_EVENTS) + await createTopics(kafka, [ + KAFKA_EVENTS, + KAFKA_EVENTS_INGESTION_HANDOFF, + KAFKA_EVENTS_WAL, + KAFKA_SESSION_RECORDING_EVENTS, + KAFKA_PERSON, + KAFKA_PERSON_UNIQUE_ID, + ]) const connected = await new Promise(async (resolve, reject) => { console.info('setting group join and crash listeners') @@ -184,3 +176,13 @@ export async function resetKafka(extraServerConfig: Partial return true } + +async function createTopics(kafka: Kafka, topics: string[]) { + const admin = kafka.admin() + await admin.connect() + await admin.createTopics({ + waitForLeaders: true, + topics: topics.map((topic) => ({ topic })), + }) + await admin.disconnect() +} From a074093eb33a972019d7634ae07d9a546204e5b5 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Thu, 4 Feb 2021 01:48:58 +0100 Subject: [PATCH 03/10] person and distinct id parity tests --- src/db.ts | 31 ++++++--- src/ingestion/process-event.ts | 4 +- src/types.ts | 10 +++ tests/clickhouse/postgres-parity.test.ts | 82 ++++++++++++++++++++++-- 4 files changed, 110 insertions(+), 17 deletions(-) diff --git a/src/db.ts b/src/db.ts index 9aeb247a..695005ab 100644 --- a/src/db.ts +++ b/src/db.ts @@ -8,6 +8,7 @@ import { KAFKA_PERSON, KAFKA_PERSON_UNIQUE_ID } from './ingestion/topics' import { chainToElements, hashElements, unparsePersonPartial } from './ingestion/utils' import { ClickHouseEvent, + ClickHousePerson, Database, Element, ElementGroup, @@ -55,12 +56,14 @@ export class DB { // Person - public async fetchPersons(database: Database = Database.Postgres): Promise { + public async fetchPersons(database?: Database.Postgres): Promise + public async fetchPersons(database: Database.ClickHouse): Promise + public async fetchPersons(database: Database = Database.Postgres): Promise { if (database === Database.ClickHouse) { return (await this.clickhouseQuery('SELECT * FROM person')) as Person[] } else if (database === Database.Postgres) { const result = await this.postgresQuery('SELECT * FROM posthog_person') - return result.rows as Person[] + return result.rows as ClickHousePerson[] } else { throw new Error(`Can't fetch persons for database: ${database}`) } @@ -158,12 +161,24 @@ export class DB { // PersonDistinctId - public async fetchDistinctIdValues(person: Person): Promise { - const result = await this.postgresQuery( - 'SELECT * FROM posthog_persondistinctid WHERE person_id=$1 and team_id=$2 ORDER BY id', - [person.id, person.team_id] - ) - return (result.rows as PersonDistinctId[]).map((pdi) => pdi.distinct_id) + public async fetchDistinctIdValues(person: Person, database: Database = Database.Postgres): Promise { + if (database === Database.ClickHouse) { + return ( + await this.clickhouseQuery( + `SELECT * FROM person_distinct_id WHERE person_id='${sanitizeSqlIdentifier( + person.uuid + )}' and team_id='${person.team_id}' ORDER BY id` + ) + ).map((row: PersonDistinctId) => row.distinct_id) + } 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', + [person.id, person.team_id] + ) + return (result.rows as PersonDistinctId[]).map((pdi) => pdi.distinct_id) + } else { + throw new Error(`Can't fetch persons for database: ${database}`) + } } public async addDistinctId(person: Person, distinctId: string): Promise { diff --git a/src/ingestion/process-event.ts b/src/ingestion/process-event.ts index 304fd9d2..174e8c87 100644 --- a/src/ingestion/process-event.ts +++ b/src/ingestion/process-event.ts @@ -359,9 +359,9 @@ export class EventsProcessor { teamId, null, false, - personUuid.toString() + personUuid.toString(), + [distinctId] ) - await this.db.addDistinctId(personCreated, distinctId) } catch {} } diff --git a/src/types.ts b/src/types.ts index 5076f3fe..73eb4030 100644 --- a/src/types.ts +++ b/src/types.ts @@ -289,6 +289,16 @@ export interface Person extends BasePerson { created_at: DateTime } +/** Clickhouse Person model. */ +export interface ClickHousePerson { + id: string + created_at: string + team_id: number + properties: string + is_identified: number + timestamp: string +} + /** Usable PersonDistinctId model. */ export interface PersonDistinctId { id: number diff --git a/tests/clickhouse/postgres-parity.test.ts b/tests/clickhouse/postgres-parity.test.ts index 6bf4208d..46d46b07 100644 --- a/tests/clickhouse/postgres-parity.test.ts +++ b/tests/clickhouse/postgres-parity.test.ts @@ -50,27 +50,95 @@ describe('postgres parity', () => { await stopServer() }) - test('createPerson does the same in both databases', async () => { - const person = server.db.createPerson( + test('createPerson', async () => { + const uuid = new UUIDT().toString() + const person = await server.db.createPerson( DateTime.utc(), { userProp: 'propValue' }, team.id, null, true, - new UUIDT().toString(), + uuid, ['distinct1', 'distinct2'] ) await delayUntilEventIngested(() => server.db.fetchPersons(Database.ClickHouse)) const clickHousePersons = await server.db.fetchPersons(Database.ClickHouse) - const postgresPersons = await server.db.fetchPersons(Database.Postgres) + expect(clickHousePersons).toEqual([ + { + id: uuid, + created_at: expect.any(String), // '2021-02-04 00:18:26.472', + team_id: team.id, + properties: '{"userProp":"propValue"}', + is_identified: 1, + _timestamp: expect.any(String), + _offset: expect.any(Number), + }, + ]) + const clickHouseDistinctIds = await server.db.fetchDistinctIdValues(person, Database.ClickHouse) + expect(clickHouseDistinctIds).toEqual(['distinct1', 'distinct2']) - expect(clickHousePersons.length).toEqual(postgresPersons.length) + const postgresPersons = await server.db.fetchPersons(Database.Postgres) + expect(postgresPersons).toEqual([ + { + id: expect.any(Number), + created_at: expect.any(String), + properties: { + userProp: 'propValue', + }, + team_id: 2, + is_user_id: null, + is_identified: true, + uuid: uuid, + }, + ]) + const postgresDistinctIds = await server.db.fetchDistinctIdValues(person, Database.Postgres) + expect(postgresDistinctIds).toEqual(['distinct1', 'distinct2']) }) // test('createPerson', async () => {}) // test('updatePerson', async () => {}) // test('deletePerson', async () => {}) - // test('addDistinctId', async () => {}) - // test('updateDistinctId', async () => {}) + + test('addDistinctId & updateDistinctId', async () => { + const uuid = new UUIDT().toString() + const uuid2 = new UUIDT().toString() + const person = await server.db.createPerson( + DateTime.utc(), + { userProp: 'propValue' }, + team.id, + null, + true, + uuid, + ['distinct1'] + ) + const anotherPerson = await server.db.createPerson( + DateTime.utc(), + { userProp: 'propValue' }, + team.id, + null, + true, + uuid2, + ['another_distinct_id'] + ) + await delayUntilEventIngested(() => server.db.fetchPersons(Database.ClickHouse)) + + const [postgresPerson] = await server.db.fetchPersons(Database.Postgres) + + const clickHouseDistinctIds = await server.db.fetchDistinctIdValues(postgresPerson, Database.ClickHouse) + const postgresDistinctIds = await server.db.fetchDistinctIdValues(postgresPerson, Database.Postgres) + + expect(clickHouseDistinctIds).toEqual(['distinct1']) + expect(postgresDistinctIds).toEqual(['distinct1']) + + await server.db.addDistinctId(postgresPerson, 'anotherOne') + + await delayUntilEventIngested(() => server.db.fetchDistinctIdValues(postgresPerson, Database.ClickHouse), 2) + + const clickHouseDistinctIds2 = await server.db.fetchDistinctIdValues(postgresPerson, Database.ClickHouse) + const postgresDistinctIds2 = await server.db.fetchDistinctIdValues(postgresPerson, Database.Postgres) + + expect(clickHouseDistinctIds2).toEqual(['distinct1', 'distinct2', 'anotherOne']) + expect(postgresDistinctIds2).toEqual(['distinct1', 'distinct2', 'anotherOne']) + }) }) From 0a48369599e7cdf37a0a69f725153adcb019c67e Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Thu, 4 Feb 2021 01:51:33 +0100 Subject: [PATCH 04/10] add TODOs --- tests/clickhouse/postgres-parity.test.ts | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/tests/clickhouse/postgres-parity.test.ts b/tests/clickhouse/postgres-parity.test.ts index 46d46b07..83c6b7be 100644 --- a/tests/clickhouse/postgres-parity.test.ts +++ b/tests/clickhouse/postgres-parity.test.ts @@ -96,11 +96,15 @@ describe('postgres parity', () => { expect(postgresDistinctIds).toEqual(['distinct1', 'distinct2']) }) - // test('createPerson', async () => {}) - // test('updatePerson', async () => {}) - // test('deletePerson', async () => {}) + test.skip('updatePerson', async () => { + // TODO + }) + + test.skip('deletePerson', async () => { + // TODO + }) - test('addDistinctId & updateDistinctId', async () => { + test('addDistinctId', async () => { const uuid = new UUIDT().toString() const uuid2 = new UUIDT().toString() const person = await server.db.createPerson( @@ -141,4 +145,8 @@ describe('postgres parity', () => { expect(clickHouseDistinctIds2).toEqual(['distinct1', 'distinct2', 'anotherOne']) expect(postgresDistinctIds2).toEqual(['distinct1', 'distinct2', 'anotherOne']) }) + + test.skip('updateDistinctId', async () => { + // could be merged with the above one + }) }) From f1b78681ed8bb9e557007da788683795c6acaf3c Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Thu, 4 Feb 2021 09:49:38 +0100 Subject: [PATCH 05/10] fetch distinct ids from clickhouse --- src/db.ts | 27 ++++++++++++++++++--------- src/types.ts | 8 ++++++++ 2 files changed, 26 insertions(+), 9 deletions(-) diff --git a/src/db.ts b/src/db.ts index 695005ab..61659258 100644 --- a/src/db.ts +++ b/src/db.ts @@ -9,6 +9,7 @@ import { chainToElements, hashElements, unparsePersonPartial } from './ingestion import { ClickHouseEvent, ClickHousePerson, + ClickHousePersonDistinctId, Database, Element, ElementGroup, @@ -161,26 +162,34 @@ export class DB { // PersonDistinctId - public async fetchDistinctIdValues(person: Person, database: Database = Database.Postgres): Promise { + public async fetchDistinctIds(person: Person, database?: Database.Postgres): Promise + public async fetchDistinctIds(person: Person, database: Database.ClickHouse): Promise + public async fetchDistinctIds( + person: Person, + database: Database = Database.Postgres + ): Promise { if (database === Database.ClickHouse) { - return ( - await this.clickhouseQuery( - `SELECT * FROM person_distinct_id WHERE person_id='${sanitizeSqlIdentifier( - person.uuid - )}' and team_id='${person.team_id}' ORDER BY id` - ) - ).map((row: PersonDistinctId) => row.distinct_id) + return (await this.clickhouseQuery( + `SELECT * FROM person_distinct_id WHERE person_id='${sanitizeSqlIdentifier( + person.uuid + )}' and team_id='${person.team_id}' ORDER BY id` + )) 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', [person.id, person.team_id] ) - return (result.rows as PersonDistinctId[]).map((pdi) => pdi.distinct_id) + return result.rows as PersonDistinctId[] } else { throw new Error(`Can't fetch persons for database: ${database}`) } } + public async fetchDistinctIdValues(person: Person, database: Database = Database.Postgres): Promise { + const personDistinctIds = await this.fetchDistinctIds(person, database as any) + return personDistinctIds.map((pdi) => pdi.distinct_id) + } + public async addDistinctId(person: Person, distinctId: string): Promise { const insertResult = await this.postgresQuery( 'INSERT INTO posthog_persondistinctid (distinct_id, person_id, team_id) VALUES ($1, $2, $3) RETURNING *', diff --git a/src/types.ts b/src/types.ts index f1137890..4e70ef5c 100644 --- a/src/types.ts +++ b/src/types.ts @@ -308,6 +308,14 @@ export interface PersonDistinctId { distinct_id: string } +/** ClickHouse PersonDistinctId model. */ +export interface ClickHousePersonDistinctId { + id: number + team_id: number + person_id: string + distinct_id: string +} + /** Usable CohortPeople model. */ export interface CohortPeople { id: number From 010da64d172d724d4b5f5787c865d730eb0284fe Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Thu, 4 Feb 2021 09:56:04 +0100 Subject: [PATCH 06/10] create a specialised function for moving distinct ids and fix postgres/clickhouse person_id difference (number vs string) --- src/db.ts | 19 +++++++++---------- src/ingestion/process-event.ts | 2 +- 2 files changed, 10 insertions(+), 11 deletions(-) diff --git a/src/db.ts b/src/db.ts index 61659258..815bf5f6 100644 --- a/src/db.ts +++ b/src/db.ts @@ -206,21 +206,20 @@ export class DB { } } - public async updateDistinctId( + public async moveDistinctId( + person: Person, personDistinctId: PersonDistinctId, - update: Partial + moveToPerson: Person ): Promise { - const updatedPersonDistinctId: PersonDistinctId = { ...personDistinctId, ...update } - await this.postgresQuery( - `UPDATE posthog_persondistinctid SET ${Object.keys(update).map( - (field, index) => `"${sanitizeSqlIdentifier(field)}" = $${index + 1}` - )} WHERE id = $${Object.values(update).length + 1}`, - [...Object.values(update), personDistinctId.id] - ) + await this.postgresQuery(`UPDATE posthog_persondistinctid SET person_id = $1 WHERE id = $2`, [ + moveToPerson.id, + personDistinctId.id, + ]) if (this.kafkaProducer) { + const clickhouseModel: ClickHousePersonDistinctId = { ...personDistinctId, person_id: moveToPerson.uuid } await this.kafkaProducer.send({ topic: KAFKA_PERSON_UNIQUE_ID, - messages: [{ value: Buffer.from(JSON.stringify(updatedPersonDistinctId)) }], + messages: [{ value: Buffer.from(JSON.stringify(clickhouseModel)) }], }) } } diff --git a/src/ingestion/process-event.ts b/src/ingestion/process-event.ts index 174e8c87..8bd83b94 100644 --- a/src/ingestion/process-event.ts +++ b/src/ingestion/process-event.ts @@ -288,7 +288,7 @@ export class EventsProcessor { ) ).rows for (const personDistinctId of otherPersonDistinctIds) { - await this.db.updateDistinctId(personDistinctId, { person_id: mergeInto.id }) + await this.db.moveDistinctId(otherPerson, personDistinctId, mergeInto) } const otherCohortPeople: CohortPeople[] = ( From affd833d7f8ca973926abd143d9990f77e381899 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Thu, 4 Feb 2021 09:56:21 +0100 Subject: [PATCH 07/10] test for updating distinct ids --- tests/clickhouse/postgres-parity.test.ts | 90 ++++++++++++++++++++---- 1 file changed, 78 insertions(+), 12 deletions(-) diff --git a/tests/clickhouse/postgres-parity.test.ts b/tests/clickhouse/postgres-parity.test.ts index 83c6b7be..55e7d5ab 100644 --- a/tests/clickhouse/postgres-parity.test.ts +++ b/tests/clickhouse/postgres-parity.test.ts @@ -62,6 +62,7 @@ describe('postgres parity', () => { ['distinct1', 'distinct2'] ) await delayUntilEventIngested(() => server.db.fetchPersons(Database.ClickHouse)) + await delayUntilEventIngested(() => server.db.fetchDistinctIdValues(person, Database.ClickHouse), 2) const clickHousePersons = await server.db.fetchPersons(Database.ClickHouse) expect(clickHousePersons).toEqual([ @@ -94,6 +95,8 @@ describe('postgres parity', () => { ]) const postgresDistinctIds = await server.db.fetchDistinctIdValues(person, Database.Postgres) expect(postgresDistinctIds).toEqual(['distinct1', 'distinct2']) + + expect(person).toEqual(postgresPersons[0]) }) test.skip('updatePerson', async () => { @@ -126,27 +129,90 @@ describe('postgres parity', () => { ['another_distinct_id'] ) await delayUntilEventIngested(() => server.db.fetchPersons(Database.ClickHouse)) - const [postgresPerson] = await server.db.fetchPersons(Database.Postgres) - const clickHouseDistinctIds = await server.db.fetchDistinctIdValues(postgresPerson, Database.ClickHouse) - const postgresDistinctIds = await server.db.fetchDistinctIdValues(postgresPerson, Database.Postgres) + await delayUntilEventIngested(() => server.db.fetchDistinctIdValues(postgresPerson, Database.ClickHouse), 1) + const clickHouseDistinctIdValues = await server.db.fetchDistinctIdValues(postgresPerson, Database.ClickHouse) + const postgresDistinctIdValues = await server.db.fetchDistinctIdValues(postgresPerson, Database.Postgres) + + // check that all is in the right format + + expect(clickHouseDistinctIdValues).toEqual(['distinct1']) + expect(postgresDistinctIdValues).toEqual(['distinct1']) + + const clickHouseDistinctIds = await server.db.fetchDistinctIds(postgresPerson, Database.ClickHouse) + const postgresDistinctIds = await server.db.fetchDistinctIds(postgresPerson, Database.Postgres) + + expect(clickHouseDistinctIds).toEqual([ + { + id: expect.any(Number), + distinct_id: 'distinct1', + person_id: person.uuid, + team_id: team.id, + _timestamp: expect.any(String), + _offset: expect.any(Number), + }, + ]) + expect(postgresDistinctIds).toEqual([ + { + id: expect.any(Number), + distinct_id: 'distinct1', + person_id: person.id, + team_id: team.id, + }, + ]) + expect(clickHouseDistinctIds[0].id).toEqual(postgresDistinctIds[0].id) - expect(clickHouseDistinctIds).toEqual(['distinct1']) - expect(postgresDistinctIds).toEqual(['distinct1']) + // add 'anotherOne' to person await server.db.addDistinctId(postgresPerson, 'anotherOne') await delayUntilEventIngested(() => server.db.fetchDistinctIdValues(postgresPerson, Database.ClickHouse), 2) - const clickHouseDistinctIds2 = await server.db.fetchDistinctIdValues(postgresPerson, Database.ClickHouse) - const postgresDistinctIds2 = await server.db.fetchDistinctIdValues(postgresPerson, Database.Postgres) + const clickHouseDistinctIdValues2 = await server.db.fetchDistinctIdValues(postgresPerson, Database.ClickHouse) + const postgresDistinctIdValues2 = await server.db.fetchDistinctIdValues(postgresPerson, Database.Postgres) - expect(clickHouseDistinctIds2).toEqual(['distinct1', 'distinct2', 'anotherOne']) - expect(postgresDistinctIds2).toEqual(['distinct1', 'distinct2', 'anotherOne']) - }) + expect(clickHouseDistinctIdValues2).toEqual(['distinct1', 'anotherOne']) + expect(postgresDistinctIdValues2).toEqual(['distinct1', 'anotherOne']) + + // check anotherPerson for their initial distinct id + + const clickHouseDistinctIdValuesOther = await server.db.fetchDistinctIdValues( + anotherPerson, + Database.ClickHouse + ) + const postgresDistinctIdValuesOther = await server.db.fetchDistinctIdValues(anotherPerson, Database.Postgres) + + expect(clickHouseDistinctIdValuesOther).toEqual(['another_distinct_id']) + expect(postgresDistinctIdValuesOther).toEqual(['another_distinct_id']) + + // move 'distinct1' from person to to anotherPerson + + await server.db.moveDistinctId(postgresPerson, postgresDistinctIds[0], anotherPerson) + await delayUntilEventIngested(() => server.db.fetchDistinctIdValues(anotherPerson, Database.ClickHouse), 2) + + // it got added + + const clickHouseDistinctIdValuesMoved = await server.db.fetchDistinctIdValues( + anotherPerson, + Database.ClickHouse + ) + const postgresDistinctIdValuesMoved = await server.db.fetchDistinctIdValues(anotherPerson, Database.Postgres) + + expect(clickHouseDistinctIdValuesMoved).toEqual(['distinct1', 'another_distinct_id']) + expect(postgresDistinctIdValuesMoved).toEqual(['distinct1', 'another_distinct_id']) + + // it got removed + + const clickHouseDistinctIdValuesRemoved = await server.db.fetchDistinctIdValues( + postgresPerson, + Database.ClickHouse + ) + const postgresDistinctIdValuesRemoved = await server.db.fetchDistinctIdValues(postgresPerson, Database.Postgres) - test.skip('updateDistinctId', async () => { - // could be merged with the above one + // The `distinct1` key is still there in clickhouse, yet ALSO there for the new person. + // Eventually this should be compacted away but it's not right now. + expect(clickHouseDistinctIdValuesRemoved).toEqual(['distinct1', 'anotherOne']) + expect(postgresDistinctIdValuesRemoved).toEqual(['anotherOne']) }) }) From 9a42c7cb7f7ccd3a47a4edc268e96a52e31d0cb3 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Thu, 4 Feb 2021 11:35:00 +0100 Subject: [PATCH 08/10] createPerson, updatePerson and deletePerson --- src/db.ts | 73 +++++++++++------- src/ingestion/process-event.ts | 2 +- src/utils.ts | 7 ++ tests/clickhouse/postgres-parity.test.ts | 96 ++++++++++++++++++++++-- tests/utils.test.ts | 11 +++ 5 files changed, 153 insertions(+), 36 deletions(-) diff --git a/src/db.ts b/src/db.ts index 815bf5f6..cfab730c 100644 --- a/src/db.ts +++ b/src/db.ts @@ -22,7 +22,7 @@ import { SessionRecordingEvent, TimestampFormat, } from './types' -import { castTimestampOrNow, clickHouseTimestampToISO, sanitizeSqlIdentifier } from './utils' +import { castTimestampOrNow, clickHouseTimestampToISO, escapeClickHouseString, sanitizeSqlIdentifier } from './utils' /** The recommended way of accessing the database. */ export class DB { @@ -61,10 +61,15 @@ export class DB { public async fetchPersons(database: Database.ClickHouse): Promise public async fetchPersons(database: Database = Database.Postgres): Promise { if (database === Database.ClickHouse) { - return (await this.clickhouseQuery('SELECT * FROM person')) as Person[] + return (await this.clickhouseQuery('SELECT * FROM person')) as ClickHousePerson[] } else if (database === Database.Postgres) { - const result = await this.postgresQuery('SELECT * FROM posthog_person') - return result.rows as ClickHousePerson[] + return ((await this.postgresQuery('SELECT * FROM posthog_person')).rows as RawPerson[]).map( + (rawPerson: RawPerson) => + ({ + ...rawPerson, + created_at: DateTime.fromISO(rawPerson.created_at).toUTC(), + } as Person) + ) } else { throw new Error(`Can't fetch persons for database: ${database}`) } @@ -87,7 +92,7 @@ export class DB { ) if (selectResult.rows.length > 0) { const rawPerson: RawPerson = selectResult.rows[0] - return { ...rawPerson, created_at: DateTime.fromISO(rawPerson.created_at) } + return { ...rawPerson, created_at: DateTime.fromISO(rawPerson.created_at).toUTC() } } } @@ -104,7 +109,8 @@ export class DB { 'INSERT INTO posthog_person (created_at, properties, team_id, is_user_id, is_identified, uuid) VALUES ($1, $2, $3, $4, $5, $6) RETURNING *', [createdAt.toISO(), JSON.stringify(properties), teamId, isUserId, isIdentified, uuid] ) - const personCreated = insertResult.rows[0] as Person + const personCreated = insertResult.rows[0] as RawPerson + const person = { ...personCreated, created_at: DateTime.fromISO(personCreated.created_at).toUTC() } as Person if (this.kafkaProducer) { const data = { created_at: castTimestampOrNow(createdAt, TimestampFormat.ClickHouse), @@ -120,10 +126,9 @@ export class DB { } for (const distinctId of distinctIds || []) { - await this.addDistinctId(personCreated, distinctId) + await this.addDistinctId(person, distinctId) } - - return personCreated + return person } public async updatePerson(person: Person, update: Partial): Promise { @@ -135,28 +140,40 @@ export class DB { )} WHERE id = $${Object.values(update).length + 1}`, values ) - if (this.kafkaProducer) { - const data = { - created_at: castTimestampOrNow(updatedPerson.created_at, TimestampFormat.ClickHouse), - properties: JSON.stringify(updatedPerson.properties), - team_id: updatedPerson.team_id, - is_identified: updatedPerson.is_identified, - id: updatedPerson.uuid.toString(), + if (this.clickhouse) { + const { is_user_id, id, uuid, ...validUpdates } = update + const updateString = Object.entries(validUpdates) + .map(([key, value]) => { + let clickhouseValue: string + if (typeof value === 'string') { + clickhouseValue = value + } else if (typeof value === 'boolean') { + clickhouseValue = value ? '1' : '0' + } else if (DateTime.isDateTime(value)) { + clickhouseValue = castTimestampOrNow(value, TimestampFormat.ClickHouse) + } else { + clickhouseValue = JSON.stringify(value) + } + return `${sanitizeSqlIdentifier(key)} = '${escapeClickHouseString(clickhouseValue)}'` + }) + .join(', ') + if (updateString.length > 0) { + await this.clickhouseQuery( + `ALTER TABLE person UPDATE ${updateString} WHERE id = '${escapeClickHouseString(person.uuid)}'` + ) } - await this.kafkaProducer.send({ - topic: KAFKA_PERSON, - messages: [{ value: Buffer.from(JSON.stringify(data)) }], - }) } return updatedPerson } - public async deletePerson(personId: number): Promise { - await this.postgresQuery('DELETE FROM posthog_persondistinctid WHERE person_id = $1', [personId]) - await this.postgresQuery('DELETE FROM posthog_person WHERE id = $1', [personId]) + public async deletePerson(person: Person): Promise { + await this.postgresQuery('DELETE FROM posthog_persondistinctid WHERE person_id = $1', [person.id]) + await this.postgresQuery('DELETE FROM posthog_person WHERE id = $1', [person.id]) if (this.clickhouse) { - await this.clickhouseQuery(`ALTER TABLE person DELETE WHERE id = ${personId}`) - await this.clickhouseQuery(`ALTER TABLE person_distinct_id DELETE WHERE person_id = ${personId}`) + await this.clickhouseQuery(`ALTER TABLE person DELETE WHERE id = '${escapeClickHouseString(person.uuid)}'`) + await this.clickhouseQuery( + `ALTER TABLE person_distinct_id DELETE WHERE person_id = '${escapeClickHouseString(person.uuid)}'` + ) } } @@ -170,7 +187,7 @@ export class DB { ): Promise { if (database === Database.ClickHouse) { return (await this.clickhouseQuery( - `SELECT * FROM person_distinct_id WHERE person_id='${sanitizeSqlIdentifier( + `SELECT * FROM person_distinct_id WHERE person_id='${escapeClickHouseString( person.uuid )}' and team_id='${person.team_id}' ORDER BY id` )) as ClickHousePersonDistinctId[] @@ -216,6 +233,8 @@ export class DB { personDistinctId.id, ]) if (this.kafkaProducer) { + // The "ALTER TABLE" statement fails with "Cannot UPDATE key column `person_id`", so just add another row + // ... even though the django version does nothing in this case! const clickhouseModel: ClickHousePersonDistinctId = { ...personDistinctId, person_id: moveToPerson.uuid } await this.kafkaProducer.send({ topic: KAFKA_PERSON_UNIQUE_ID, @@ -281,7 +300,7 @@ export class DB { public async fetchElements(event?: Event): Promise { if (this.kafkaProducer) { const events = (await this.clickhouseQuery( - `SELECT elements_chain FROM events WHERE uuid='${sanitizeSqlIdentifier((event as any).uuid)}'` + `SELECT elements_chain FROM events WHERE uuid='${escapeClickHouseString((event as any).uuid)}'` )) as ClickHouseEvent[] const chain = events?.[0]?.elements_chain return chainToElements(chain) diff --git a/src/ingestion/process-event.ts b/src/ingestion/process-event.ts index 8bd83b94..acc584e6 100644 --- a/src/ingestion/process-event.ts +++ b/src/ingestion/process-event.ts @@ -301,7 +301,7 @@ export class EventsProcessor { ]) } - await this.db.deletePerson(otherPerson.id) + await this.db.deletePerson(otherPerson) } } diff --git a/src/utils.ts b/src/utils.ts index eccf8e5e..bd6227a4 100644 --- a/src/utils.ts +++ b/src/utils.ts @@ -325,3 +325,10 @@ export function delay(ms: number): Promise { export function sanitizeSqlIdentifier(unquotedIdentifier: string): string { return unquotedIdentifier.replace(/[^\w\d_]+/g, '') } + +/** Escape single quotes and slashes */ +export function escapeClickHouseString(string: string): string { + // In string literals, you need to escape at least `'` and `\`. + // https://clickhouse.tech/docs/en/sql-reference/syntax/ + return string.replace(/\\/g, '\\\\').replace(/'/g, "\\'") +} diff --git a/tests/clickhouse/postgres-parity.test.ts b/tests/clickhouse/postgres-parity.test.ts index 55e7d5ab..a964faf1 100644 --- a/tests/clickhouse/postgres-parity.test.ts +++ b/tests/clickhouse/postgres-parity.test.ts @@ -1,10 +1,10 @@ -import { Database, LogLevel, PluginsServer, PluginsServerConfig, Team } from '../../src/types' +import { Database, LogLevel, PluginsServer, PluginsServerConfig, Team, TimestampFormat } from '../../src/types' import { getFirstTeam, resetTestDatabase } from '../helpers/sql' import { startPluginsServer } from '../../src/server' import { makePiscina } from '../../src/worker/piscina' import { createPosthog, DummyPostHog } from '../../src/extensions/posthog' import { pluginConfig39 } from '../helpers/plugins' -import { UUIDT } from '../../src/utils' +import { castTimestampOrNow, UUIDT } from '../../src/utils' import { resetTestDatabaseClickhouse } from '../helpers/clickhouse' import { resetKafka } from '../helpers/kafka' import { delayUntilEventIngested } from '../shared/process-event' @@ -83,7 +83,7 @@ describe('postgres parity', () => { expect(postgresPersons).toEqual([ { id: expect.any(Number), - created_at: expect.any(String), + created_at: expect.any(DateTime), properties: { userProp: 'propValue', }, @@ -99,15 +99,95 @@ describe('postgres parity', () => { expect(person).toEqual(postgresPersons[0]) }) - test.skip('updatePerson', async () => { - // TODO + test('updatePerson', async () => { + const uuid = new UUIDT().toString() + const person = await server.db.createPerson( + DateTime.utc(), + { userProp: 'propValue' }, + team.id, + null, + false, + uuid, + ['distinct1', 'distinct2'] + ) + await delayUntilEventIngested(() => server.db.fetchPersons(Database.ClickHouse)) + await delayUntilEventIngested(() => server.db.fetchDistinctIdValues(person, Database.ClickHouse), 2) + + // update JSON and boolean to true + + await server.db.updatePerson(person, { properties: { replacedUserProp: 'propValue' }, is_identified: true }) + + await delayUntilEventIngested(async () => + (await server.db.fetchPersons(Database.ClickHouse)).filter((p) => p.is_identified) + ) + + const clickHousePersons = await server.db.fetchPersons(Database.ClickHouse) + const postgresPersons = await server.db.fetchPersons(Database.Postgres) + + expect(clickHousePersons.length).toEqual(1) + expect(postgresPersons.length).toEqual(1) + + expect(postgresPersons[0].is_identified).toEqual(true) + expect(postgresPersons[0].properties).toEqual({ replacedUserProp: 'propValue' }) + + expect(clickHousePersons[0].is_identified).toEqual(1) + expect(clickHousePersons[0].properties).toEqual('{"replacedUserProp":"propValue"}') + + // update date and boolean to false + + const randomDate = DateTime.utc().minus(100000).setZone('UTC') + await server.db.updatePerson(person, { created_at: randomDate, is_identified: false }) + + await delayUntilEventIngested(async () => + (await server.db.fetchPersons(Database.ClickHouse)).filter((p) => p.is_identified) + ) + + const clickHousePersons2 = await server.db.fetchPersons(Database.ClickHouse) + const postgresPersons2 = await server.db.fetchPersons(Database.Postgres) + + expect(clickHousePersons2.length).toEqual(1) + expect(postgresPersons2.length).toEqual(1) + + expect(postgresPersons2[0].is_identified).toEqual(false) + expect(postgresPersons2[0].created_at.toISO()).toEqual(randomDate.toISO()) + + expect(clickHousePersons2[0].is_identified).toEqual(0) + expect(clickHousePersons2[0].created_at).toEqual(castTimestampOrNow(randomDate, TimestampFormat.ClickHouse)) }) - test.skip('deletePerson', async () => { - // TODO + test('deletePerson', async () => { + const uuid = new UUIDT().toString() + const person = await server.db.createPerson( + DateTime.utc(), + { userProp: 'propValue' }, + team.id, + null, + false, + uuid, + ['distinct1', 'distinct2'] + ) + await delayUntilEventIngested(() => server.db.fetchPersons(Database.ClickHouse)) + await delayUntilEventIngested(() => server.db.fetchDistinctIdValues(person, Database.ClickHouse), 2) + + await server.db.deletePerson(person) + + await delayUntilEventIngested(async () => + (await server.db.fetchPersons(Database.ClickHouse)).length === 0 ? ['deleted!'] : [] + ) + + const clickHousePersons = await server.db.fetchPersons(Database.ClickHouse) + const postgresPersons = await server.db.fetchPersons(Database.Postgres) + + expect(clickHousePersons.length).toEqual(0) + expect(postgresPersons.length).toEqual(0) + + const clickHouseDistinctIdValues = await server.db.fetchDistinctIdValues(person, Database.ClickHouse) + const postgresDistinctIdValues = await server.db.fetchDistinctIdValues(person, Database.Postgres) + expect(clickHouseDistinctIdValues.length).toEqual(0) + expect(postgresDistinctIdValues.length).toEqual(0) }) - test('addDistinctId', async () => { + test('addDistinctId & moveDistinctId', async () => { const uuid = new UUIDT().toString() const uuid2 = new UUIDT().toString() const person = await server.db.createPerson( diff --git a/tests/utils.test.ts b/tests/utils.test.ts index 27e1de0a..6cdef39c 100644 --- a/tests/utils.test.ts +++ b/tests/utils.test.ts @@ -8,6 +8,7 @@ import { UUID, UUIDT, sanitizeSqlIdentifier, + escapeClickHouseString, } from '../src/utils' import { randomBytes } from 'crypto' import { LogLevel } from '../src/types' @@ -306,3 +307,13 @@ describe('sanitizeSqlIdentifier', () => { expect(sanitizedIdentifier).toStrictEqual('some_fieldDROPTABLEactually_an_injection9') }) }) + +describe('escapeClickHouseString', () => { + it('escapes single quotes and slashes', () => { + const rawString = "insert'escape \\" + + const sanitizedString = escapeClickHouseString(rawString) + + expect(sanitizedString).toStrictEqual("insert\\'escape \\\\") + }) +}) From 3543c8b98ad8700aa4373727b45bccf360b9f805 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Thu, 4 Feb 2021 11:48:18 +0100 Subject: [PATCH 09/10] add a debug line to help debug flaky github action --- tests/postgres/process-event.test.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/postgres/process-event.test.ts b/tests/postgres/process-event.test.ts index 2c12142b..3288aea1 100644 --- a/tests/postgres/process-event.test.ts +++ b/tests/postgres/process-event.test.ts @@ -13,6 +13,8 @@ describe('process event (postgresql)', () => { const elementsHash = server!.db.createElementGroup(elements, 2) const elementGroup = await server!.db.fetchElements() + console.log(elementGroup) + expect(elementGroup[0].tag_name).toEqual('button') expect(elementGroup[1].tag_name).toEqual('div') expect(elementGroup.length).toEqual(2) From 11ea92bd2ee7fecc70cb631a3fe479b238c42a92 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Thu, 4 Feb 2021 12:13:32 +0100 Subject: [PATCH 10/10] remove falsehood --- src/db.ts | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/db.ts b/src/db.ts index cfab730c..4458298b 100644 --- a/src/db.ts +++ b/src/db.ts @@ -233,8 +233,6 @@ export class DB { personDistinctId.id, ]) if (this.kafkaProducer) { - // The "ALTER TABLE" statement fails with "Cannot UPDATE key column `person_id`", so just add another row - // ... even though the django version does nothing in this case! const clickhouseModel: ClickHousePersonDistinctId = { ...personDistinctId, person_id: moveToPerson.uuid } await this.kafkaProducer.send({ topic: KAFKA_PERSON_UNIQUE_ID,