Add a distributed engine table to back migrations

This will work with clickhouse-operator

src/infi/clickhouse_orm/database.py

@@ -355,12 +355,17 @@ class Database(object):
                 break
 
     def _get_applied_migrations(self, migrations_package_name, replicated):
-        from .migrations import MigrationHistory, MigrationHistoryReplicated
+        from .migrations import MigrationHistory, MigrationHistoryReplicated, MigrationHistoryDistributed
 
-        Model = MigrationHistoryReplicated if replicated else MigrationHistory
-        self.create_table(Model)
-        query = "SELECT module_name from $table WHERE package_name = '%s'" % migrations_package_name
-        query = self._substitute(query, Model)
+        query = "SELECT DISTINCT module_name FROM $table WHERE package_name = '%s'" % migrations_package_name
+
+        if replicated:
+            self.create_table(MigrationHistoryReplicated)
+            self.create_table(MigrationHistoryDistributed)
+            query = self._substitute(query, MigrationHistoryDistributed)
+        else:
+            self.create_table(MigrationHistory)
+            query = self._substitute(query, MigrationHistory)
         return set(obj.module_name for obj in self.select(query))
 
     def _send(self, data, settings=None, stream=False):

src/infi/clickhouse_orm/migrations.py

@@ -1,6 +1,6 @@
 import logging
 
-from .engines import MergeTree
+from .engines import Distributed, MergeTree
 from .fields import DateField, StringField
 from .models import BufferModel, Model
 from .utils import escape, get_subclass_names
@@ -294,18 +294,33 @@ class MigrationHistoryReplicated(Model):
     package_name = StringField()
     module_name = StringField()
     applied = DateField()
-    
+
     engine = MergeTree(
         "applied",
         ("package_name", "module_name"),
         replica_table_path="/clickhouse/prod/tables/noshard/posthog.infi_clickhouse_orm_migrations/{uuid}",
         replica_name="{replica}-{shard}",
     )
-    
+
     @classmethod
     def table_name(cls):
         return "infi_clickhouse_orm_migrations"
 
+class MigrationHistoryDistributed(Model):
+    """
+    Distributed table for storing which migrations are applied to the containing database
+    """
+
+    package_name = StringField()
+    module_name = StringField()
+    applied = DateField()
+
+    engine = Distributed(table="infi_clickhouse_orm_migrations", sharding_key="rand()")
+
+    @classmethod
+    def table_name(cls):
+        return "infi_clickhouse_orm_migrations_distributed"
+
 
 # Expose only relevant classes in import *
 __all__ = get_subclass_names(locals(), Operation)