Use ReplicatedMergeTree

PostHog · Dec 14, 2023 · 86e98f9 · 86e98f9
1 parent a0aaab4
commit 86e98f9
Show file tree

Hide file tree

Showing 3 changed files with 19 additions and 9 deletions.
diff --git a/posthog/clickhouse/migrations/0050_add_channel_definitions.py b/posthog/clickhouse/migrations/0050_add_channel_definitions.py
@@ -6,7 +6,7 @@
 )
 
 operations = [
-    run_sql_with_exceptions(CHANNEL_DEFINITION_TABLE_SQL),
+    run_sql_with_exceptions(CHANNEL_DEFINITION_TABLE_SQL()),
     run_sql_with_exceptions(CHANNEL_DEFINITION_DATA_SQL),
     run_sql_with_exceptions(CHANNEL_DEFINITION_DICTIONARY_SQL),
 ]
diff --git a/posthog/models/channel_type/sql.py b/posthog/models/channel_type/sql.py
@@ -1,18 +1,28 @@
 import json
 import os
 
+from posthog.clickhouse.table_engines import (
+    MergeTreeEngine,
+    ReplicationScheme,
+)
 from posthog.settings import CLICKHOUSE_CLUSTER
 
-CHANNEL_DEFINITION_TABLE_SQL = f"""
-CREATE TABLE IF NOT EXISTS channel_definition ON CLUSTER {CLICKHOUSE_CLUSTER} (
+CHANNEL_DEFINITION_TABLE_SQL = (
+    lambda: """
+CREATE TABLE IF NOT EXISTS channel_definition ON CLUSTER '{cluster}' (
     domain String NOT NULL,
     kind String NOT NULL,
     domain_type String NULL,
     type_if_paid String NULL,
-    type_if_organic String NULL,
-) ENGINE = MergeTree()
-ORDER BY domain, kind;
-"""
+    type_if_organic String NULL
+) ENGINE = {engine}
+ORDER BY (domain, kind);
+""".format(
+        engine=MergeTreeEngine("channel_definition", replication_scheme=ReplicationScheme.REPLICATED),
+        cluster=CLICKHOUSE_CLUSTER,
+    )
+)
+
 
 DROP_CHANNEL_DEFINITION_TABLE_SQL = f"DROP TABLE IF EXISTS channel_definition ON CLUSTER '{CLICKHOUSE_CLUSTER}'"
 

diff --git a/posthog/test/base.py b/posthog/test/base.py
@@ -860,7 +860,7 @@ def setUp(self):
                 PERSONS_TABLE_SQL(),
                 SESSION_RECORDING_EVENTS_TABLE_SQL(),
                 SESSION_REPLAY_EVENTS_TABLE_SQL(),
-                CHANNEL_DEFINITION_TABLE_SQL,
+                CHANNEL_DEFINITION_TABLE_SQL(),
                 CHANNEL_DEFINITION_DICTIONARY_SQL,
             ]
         )
@@ -894,7 +894,7 @@ def tearDown(self):
                 PERSONS_TABLE_SQL(),
                 SESSION_RECORDING_EVENTS_TABLE_SQL(),
                 SESSION_REPLAY_EVENTS_TABLE_SQL(),
-                CHANNEL_DEFINITION_TABLE_SQL,
+                CHANNEL_DEFINITION_TABLE_SQL(),
                 CHANNEL_DEFINITION_DICTIONARY_SQL,
             ]
         )