docs(brand): add hero banner

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.github/hero.svg

@@ -0,0 +1,9 @@
+<svg xmlns="http://www.w3.org/2000/svg" width="1280" height="640" viewBox="0 0 1280 640" role="img" aria-label="infi.clickhouse_orm">
+<rect width="1280" height="640" fill="#0A0A0A"/>
+<svg x="96" y="215" width="210" height="210" viewBox="0 0 67 67"><path d="M22.21 67V44.6369H0V67H22.21Z" fill="#fff"/><path d="M66.7038 22.3184H22.2534L0.0878906 44.6367H44.4634L66.7038 22.3184Z" fill="#fff"/><path d="M22.21 0H0V22.3184H22.21V0Z" fill="#fff"/><path d="M66.7198 0H44.5098V22.3184H66.7198V0Z" fill="#fff"/><path d="M66.7198 67V44.6369H44.5098V67H66.7198Z" fill="#fff"/></svg>
+<text x="378" y="276" font-family="Inter,system-ui,-apple-system,sans-serif" font-size="78" font-weight="800" letter-spacing="-2" fill="#ffffff">infi.clickhouse_orm</text>
+<text x="378" y="322" font-family="Inter,system-ui,sans-serif" font-size="30" fill="#ffffff" opacity=".66">A Python library for working with the ClickHouse database…</text>
+<rect x="378" y="338" width="806" height="3" rx="1.5" fill="#ffffff" opacity=".9"/>
+<text x="378" y="390" font-family="Inter,system-ui,sans-serif" font-size="24" font-weight="600" fill="#ffffff" opacity=".5">github.com/hanzoai</text>
+<text x="1184" y="390" text-anchor="end" font-family="Inter,system-ui,sans-serif" font-size="24" font-weight="600" fill="#ffffff" opacity=".5">hanzo.ai</text>
+</svg>

.gitignore

@@ -18,7 +18,6 @@
 bin/
 build/
 dist/
-setup.py
 
 # buildout
 buildout.in.cfg

CLAUDE.md

@@ -0,0 +1 @@
+LLM.md

LLM.md

@@ -0,0 +1,19 @@
+# infi.clickhouse_orm
+
+**Org:** hanzoai  ·  **Ecosystem:** hanzo  ·  **Path:** `/Users/a/work/hanzo/hanzoai/infi.clickhouse_orm`
+**Origin:** https://github.com/hanzoai/infi.clickhouse_orm.git
+
+## Discovery
+
+This file (`CLAUDE.md`) is the canonical agent-facing readme; `LLM.md` is a symlink to it. Update either name and both stay in sync.
+
+## Where to look first
+
+- `README.md` — human-facing overview (if present)
+- `package.json` / `Cargo.toml` / `pyproject.toml` / `go.mod` — language & deps
+- `.github/workflows/` — CI surface
+- `docs/` — extended docs (if present)
+
+## Sibling repos
+
+See the org-level `LLM.md` at `/Users/a/work/hanzo/hanzoai/LLM.md` for the full inventory of sibling repos and inter-repo dependencies.

README.md

@@ -1,3 +1,5 @@
+<p align="center"><img src=".github/hero.svg" alt="infi.clickhouse_orm" width="880"></p>
+
 Introduction
 ============
 

setup.py

@@ -0,0 +1,54 @@
+
+SETUP_INFO = dict(
+    name = 'infi.clickhouse_orm',
+    version = '2.1.0.post20',
+    author = 'James Greenhill',
+    author_email = 'fuziontech@gmail.com',
+
+    url = 'https://github.com/Infinidat/infi.clickhouse_orm',
+    license = 'BSD',
+    description = """A Python library for working with the ClickHouse database""",
+
+    # http://pypi.python.org/pypi?%3Aaction=list_classifiers
+    classifiers = [
+        "Intended Audience :: Developers",
+        "Intended Audience :: System Administrators",
+        "License :: OSI Approved :: BSD License",
+        "Operating System :: OS Independent",
+        "Programming Language :: Python",
+        "Programming Language :: Python :: 2.7",
+        "Programming Language :: Python :: 3.4",
+        "Topic :: Software Development :: Libraries :: Python Modules",
+        "Topic :: Database"
+    ],
+
+    install_requires = [
+'iso8601 >= 0.1.12',
+'pytz',
+'requests',
+'setuptools'
+],
+    namespace_packages = ['infi'],
+
+    package_dir = {'': 'src'},
+    package_data = {'': []},
+    include_package_data = True,
+    zip_safe = False,
+
+    entry_points = dict(
+        console_scripts = [],
+        gui_scripts = [],
+        ),
+)
+
+if SETUP_INFO['url'] is None:
+    _ = SETUP_INFO.pop('url')
+
+def setup():
+    from setuptools import setup as _setup
+    from setuptools import find_packages
+    SETUP_INFO['packages'] = find_packages('src')
+    _setup(**SETUP_INFO)
+
+if __name__ == '__main__':
+    setup()

src/infi/clickhouse_orm/database.py

@@ -81,8 +81,10 @@ class Database(object):
     '''
 
     def __init__(self, db_name, db_url='http://localhost:8123/',
-                 username=None, password=None, readonly=False, autocreate=True,
-                 timeout=60, verify_ssl_cert=True, log_statements=False):
+                 username=None, password=None, cluster=None,
+                 readonly=False, autocreate=True,
+                 timeout=60, verify_ssl_cert=True, log_statements=False, *, randomize_replica_paths=False,
+                 trust_env=True):
         '''
         Initializes a database instance. Unless it's readonly, the database will be
         created on the ClickHouse server if it does not already exist.
@@ -91,21 +93,29 @@ class Database(object):
         - `db_url`: URL of the ClickHouse server.
         - `username`: optional connection credentials.
         - `password`: optional connection credentials.
+        - `cluster`: optional cluster to create tables on
         - `readonly`: use a read-only connection.
         - `autocreate`: automatically create the database if it does not exist (unless in readonly mode).
         - `timeout`: the connection timeout in seconds.
         - `verify_ssl_cert`: whether to verify the server's certificate when connecting via HTTPS.
         - `log_statements`: when True, all database statements are logged.
+        - `randomize_replica_paths`: when True, a random integer is appended to table replica paths.
+          This way replicated tables (such as `MigrationHistoryReplicated`) can be dropped and recreated without causing
+          a conflict in Zookeeper. This shouldn't be used in production though.
+        - `trust_env`: when True, the request session will use environment variables for proxy configuration etc.
         '''
         self.db_name = db_name
         self.db_url = db_url
+        self.cluster = cluster
         self.readonly = False
         self.timeout = timeout
         self.request_session = requests.Session()
         self.request_session.verify = verify_ssl_cert
+        self.request_session.trust_env = trust_env
         if username:
             self.request_session.auth = (username, password or '')
         self.log_statements = log_statements
+        self.randomize_replica_paths = randomize_replica_paths
         self.settings = {}
         self.db_exists = False # this is required before running _is_existing_database
         self.db_exists = self._is_existing_database()
@@ -330,7 +340,7 @@ class Database(object):
             page_size=page_size
         )
 
-    def migrate(self, migrations_package_name, up_to=9999):
+    def migrate(self, migrations_package_name, up_to=9999, replicated=False):
         '''
         Executes schema migrations.
 
@@ -340,7 +350,7 @@ class Database(object):
         '''
         from .migrations import MigrationHistory
         logger = logging.getLogger('migrations')
-        applied_migrations = self._get_applied_migrations(migrations_package_name)
+        applied_migrations = self._get_applied_migrations_and_create_tables(migrations_package_name, replicated=replicated)
         modules = import_submodules(migrations_package_name)
         unapplied_migrations = set(modules.keys()) - applied_migrations
         for name in sorted(unapplied_migrations):
@@ -351,11 +361,28 @@ class Database(object):
             if int(name[:4]) >= up_to:
                 break
 
-    def _get_applied_migrations(self, migrations_package_name):
-        from .migrations import MigrationHistory
-        self.create_table(MigrationHistory)
-        query = "SELECT module_name from $table WHERE package_name = '%s'" % migrations_package_name
-        query = self._substitute(query, MigrationHistory)
+    def _get_applied_migrations_and_create_tables(self, migrations_package_name, replicated, allow_missing_tables=True):
+        if not allow_missing_tables:
+            return self._get_applied_migrations(migrations_package_name, replicated)
+
+        try:
+            return self._get_applied_migrations(migrations_package_name, replicated)
+        except ServerError:
+            from .migrations import MigrationHistory, MigrationHistoryReplicated, MigrationHistoryDistributed
+            if replicated:
+                self.create_table(MigrationHistoryReplicated)
+                self.create_table(MigrationHistoryDistributed)
+            else:
+                self.create_table(MigrationHistory)
+
+            return self._get_applied_migrations_and_create_tables(migrations_package_name, replicated, allow_missing_tables=False)
+
+
+    def _get_applied_migrations(self, migrations_package_name, replicated):
+        from .migrations import MigrationHistory, MigrationHistoryDistributed
+        query = "SELECT DISTINCT module_name FROM $table WHERE package_name = '%s'" % migrations_package_name
+        query = self._substitute(query, MigrationHistoryDistributed if replicated else MigrationHistory)
+
         return set(obj.module_name for obj in self.select(query))
 
     def _send(self, data, settings=None, stream=False):
@@ -408,6 +435,8 @@ class Database(object):
         except ServerError as e:
             logger.exception('Cannot determine server version (%s), assuming 1.1.0', e)
             ver = '1.1.0'
+        # :TRICKY: Altinity cloud uses a non-numeric suffix for the version, which this removes.
+        ver = re.sub(r"[.\D]+$", '', ver)
         return tuple(int(n) for n in ver.split('.')) if as_tuple else ver
 
     def _is_existing_database(self):

src/infi/clickhouse_orm/engines.py

@@ -1,6 +1,7 @@
 from __future__ import unicode_literals
 
 import logging
+import random
 
 from .utils import comma_join, get_subclass_names
 
@@ -104,7 +105,10 @@ class MergeTree(Engine):
     def _build_sql_params(self, db):
         params = []
         if self.replica_name:
-            params += ["'%s'" % self.replica_table_path, "'%s'" % self.replica_name]
+            final_replica_table_path = self.replica_table_path
+            if db.randomize_replica_paths:
+                final_replica_table_path += f'/{random.randint(0, 100000000)}'
+            params += ["'%s'" % final_replica_table_path, "'%s'" % self.replica_name]
 
         # In ClickHouse 1.1.54310 custom partitioning key was introduced
         # https://clickhouse.tech/docs/en/table_engines/custom_partitioning_key/
@@ -224,9 +228,9 @@ class Distributed(Engine):
     See full documentation here
     https://clickhouse.tech/docs/en/engines/table-engines/special/distributed/
     """
-    def __init__(self, cluster, table=None, sharding_key=None):
+    def __init__(self, cluster=None, table=None, sharding_key=None):
         """
-        - `cluster`: what cluster to access data from
+        - `cluster`: what cluster to access data from. Defaults to db.cluster
         - `table`: underlying table that actually stores data.
         If you are not specifying any table here, ensure that it can be inferred
         from your model's superclass (see models.DistributedModel.fix_engine_table)
@@ -259,7 +263,11 @@ class Distributed(Engine):
             raise ValueError("Cannot create {} engine: specify an underlying table".format(
                 self.__class__.__name__))
 
-        params = ["`%s`" % p for p in [self.cluster, db.db_name, self.table_name]]
+        if self.cluster is None and db.cluster is None:
+            raise ValueError("Cannot create engine: specify a cluster")
+
+        cluster = self.cluster if self.cluster is not None else db.cluster
+        params = ["`%s`" % p for p in [cluster, db.db_name, self.table_name]]
         if self.sharding_key:
             params.append(self.sharding_key)
         return params

src/infi/clickhouse_orm/fields.py

@@ -468,9 +468,16 @@ class BaseEnumField(Field):
             return value
         try:
             if isinstance(value, str):
-                return self.enum_cls[value]
+                try:
+                    return self.enum_cls[value]
+                except Exception:
+                    return self.enum_cls(value)
             if isinstance(value, bytes):
-                return self.enum_cls[value.decode('UTF-8')]
+                decoded = value.decode('UTF-8')
+                try:
+                    return self.enum_cls[decoded]
+                except Exception:
+                    return self.enum_cls(decoded)
             if isinstance(value, int):
                 return self.enum_cls(value)
         except (KeyError, ValueError):
@@ -665,4 +672,3 @@ class LowCardinalityField(Field):
 
 # Expose only relevant classes in import *
 __all__ = get_subclass_names(locals(), Field)
-

src/infi/clickhouse_orm/migrations.py

@@ -1,70 +1,71 @@
-from .models import Model, BufferModel
+import logging
+
+from .engines import Distributed, MergeTree
 from .fields import DateField, StringField
-from .engines import MergeTree
+from .models import BufferModel, Model
 from .utils import escape, get_subclass_names
 
-import logging
-logger = logging.getLogger('migrations')
+logger = logging.getLogger("migrations")
 
 
-class Operation():
-    '''
+class Operation:
+    """
     Base class for migration operations.
-    '''
+    """
 
     def apply(self, database):
-        raise NotImplementedError()   # pragma: no cover
+        raise NotImplementedError()  # pragma: no cover
 
 
 class ModelOperation(Operation):
-    '''
+    """
     Base class for migration operations that work on a specific model.
-    '''
+    """
 
     def __init__(self, model_class):
-        '''
+        """
         Initializer.
-        '''
+        """
         self.model_class = model_class
         self.table_name = model_class.table_name()
 
     def _alter_table(self, database, cmd):
-        '''
+        """
         Utility for running ALTER TABLE commands.
-        '''
+        """
         cmd = "ALTER TABLE $db.`%s` %s" % (self.table_name, cmd)
         logger.debug(cmd)
         database.raw(cmd)
 
 
 class CreateTable(ModelOperation):
-    '''
+    """
     A migration operation that creates a table for a given model class.
-    '''
+    """
 
     def apply(self, database):
-        logger.info('    Create table %s', self.table_name)
+        logger.info("    Create table %s", self.table_name)
         if issubclass(self.model_class, BufferModel):
             database.create_table(self.model_class.engine.main_model)
         database.create_table(self.model_class)
 
 
 class AlterTable(ModelOperation):
-    '''
+    """
     A migration operation that compares the table of a given model class to
     the model's fields, and alters the table to match the model. The operation can:
       - add new columns
       - drop obsolete columns
       - modify column types
     Default values are not altered by this operation.
-    '''
+    """
 
     def _get_table_fields(self, database):
         query = "DESC `%s`.`%s`" % (database.db_name, self.table_name)
         return [(row.name, row.type) for row in database.select(query)]
 
     def apply(self, database):
-        logger.info('    Alter table %s', self.table_name)
+        logger.info("    Alter table %s", self.table_name)
 
         # Note that MATERIALIZED and ALIAS fields are always at the end of the DESC,
         # ADD COLUMN ... AFTER doesn't affect it
@@ -73,8 +74,8 @@ class AlterTable(ModelOperation):
         # Identify fields that were deleted from the model
         deleted_fields = set(table_fields.keys()) - set(self.model_class.fields())
         for name in deleted_fields:
-            logger.info('        Drop column %s', name)
-            self._alter_table(database, 'DROP COLUMN %s' % name)
+            logger.info("        Drop column %s", name)
+            self._alter_table(database, "DROP COLUMN %s" % name)
             del table_fields[name]
 
         # Identify fields that were added to the model
@@ -82,11 +83,11 @@ class AlterTable(ModelOperation):
         for name, field in self.model_class.fields().items():
             is_regular_field = not (field.materialized or field.alias)
             if name not in table_fields:
-                logger.info('        Add column %s', name)
-                assert prev_name, 'Cannot add a column to the beginning of the table'
-                cmd = 'ADD COLUMN %s %s' % (name, field.get_sql(db=database))
+                logger.info("        Add column %s", name)
+                assert prev_name, "Cannot add a column to the beginning of the table"
+                cmd = "ADD COLUMN %s %s" % (name, field.get_sql(db=database))
                 if is_regular_field:
-                    cmd += ' AFTER %s' % prev_name
+                    cmd += " AFTER %s" % prev_name
                 self._alter_table(database, cmd)
 
             if is_regular_field:
@@ -98,24 +99,35 @@ class AlterTable(ModelOperation):
         # The order of class attributes can be changed any time, so we can't count on it
         # Secondly, MATERIALIZED and ALIAS fields are always at the end of the DESC, so we can't expect them to save
         # attribute position. Watch https://github.com/Infinidat/infi.clickhouse_orm/issues/47
-        model_fields = {name: field.get_sql(with_default_expression=False, db=database)
-                        for name, field in self.model_class.fields().items()}
+        model_fields = {
+            name: field.get_sql(with_default_expression=False, db=database)
+            for name, field in self.model_class.fields().items()
+        }
         for field_name, field_sql in self._get_table_fields(database):
             # All fields must have been created and dropped by this moment
-            assert field_name in model_fields, 'Model fields and table columns in disagreement'
+            assert (
+                field_name in model_fields
+            ), "Model fields and table columns in disagreement"
 
             if field_sql != model_fields[field_name]:
-                logger.info('        Change type of column %s from %s to %s', field_name, field_sql,
-                            model_fields[field_name])
-                self._alter_table(database, 'MODIFY COLUMN %s %s' % (field_name, model_fields[field_name]))
+                logger.info(
+                    "        Change type of column %s from %s to %s",
+                    field_name,
+                    field_sql,
+                    model_fields[field_name],
+                )
+                self._alter_table(
+                    database,
+                    "MODIFY COLUMN %s %s" % (field_name, model_fields[field_name]),
+                )
 
 
 class AlterTableWithBuffer(ModelOperation):
-    '''
+    """
     A migration operation for altering a buffer table and its underlying on-disk table.
     The buffer table is dropped, the on-disk table is altered, and then the buffer table
     is re-created.
-    '''
+    """
 
     def apply(self, database):
         if issubclass(self.model_class, BufferModel):
@@ -127,139 +139,142 @@ class AlterTableWithBuffer(ModelOperation):
 
 
 class DropTable(ModelOperation):
-    '''
+    """
     A migration operation that drops the table of a given model class.
-    '''
+    """
 
     def apply(self, database):
-        logger.info('    Drop table %s', self.table_name)
+        logger.info("    Drop table %s", self.table_name)
         database.drop_table(self.model_class)
 
 
 class AlterConstraints(ModelOperation):
-    '''
+    """
     A migration operation that adds new constraints from the model to the database
     table, and drops obsolete ones. Constraints are identified by their names, so
     a change in an existing constraint will not be detected unless its name was changed too.
     ClickHouse does not check that the constraints hold for existing data in the table.
-    '''
+    """
 
     def apply(self, database):
-        logger.info('    Alter constraints for %s', self.table_name)
+        logger.info("    Alter constraints for %s", self.table_name)
         existing = self._get_constraint_names(database)
         # Go over constraints in the model
         for constraint in self.model_class._constraints.values():
             # Check if it's a new constraint
             if constraint.name not in existing:
-                logger.info('        Add constraint %s', constraint.name)
-                self._alter_table(database, 'ADD %s' % constraint.create_table_sql())
+                logger.info("        Add constraint %s", constraint.name)
+                self._alter_table(database, "ADD %s" % constraint.create_table_sql())
             else:
                 existing.remove(constraint.name)
         # Remaining constraints in `existing` are obsolete
         for name in existing:
-            logger.info('        Drop constraint %s', name)
-            self._alter_table(database, 'DROP CONSTRAINT `%s`' % name)
+            logger.info("        Drop constraint %s", name)
+            self._alter_table(database, "DROP CONSTRAINT `%s`" % name)
 
     def _get_constraint_names(self, database):
-        '''
+        """
         Returns a set containing the names of existing constraints in the table.
-        '''
+        """
         import re
-        table_def = database.raw('SHOW CREATE TABLE $db.`%s`' % self.table_name)
-        matches = re.findall(r'\sCONSTRAINT\s+`?(.+?)`?\s+CHECK\s', table_def)
+
+        table_def = database.raw("SHOW CREATE TABLE $db.`%s`" % self.table_name)
+        matches = re.findall(r"\sCONSTRAINT\s+`?(.+?)`?\s+CHECK\s", table_def)
         return set(matches)
 
 
 class AlterIndexes(ModelOperation):
-    '''
+    """
     A migration operation that adds new indexes from the model to the database
     table, and drops obsolete ones. Indexes are identified by their names, so
     a change in an existing index will not be detected unless its name was changed too.
-    '''
+    """
 
     def __init__(self, model_class, reindex=False):
-        '''
+        """
         Initializer.
         By default ClickHouse does not build indexes over existing data, only for
         new data. Passing `reindex=True` will run `OPTIMIZE TABLE` in order to build
         the indexes over the existing data.
-        '''
+        """
         super().__init__(model_class)
         self.reindex = reindex
 
     def apply(self, database):
-        logger.info('    Alter indexes for %s', self.table_name)
+        logger.info("    Alter indexes for %s", self.table_name)
         existing = self._get_index_names(database)
         logger.info(existing)
         # Go over indexes in the model
         for index in self.model_class._indexes.values():
             # Check if it's a new index
             if index.name not in existing:
-                logger.info('        Add index %s', index.name)
-                self._alter_table(database, 'ADD %s' % index.create_table_sql())
+                logger.info("        Add index %s", index.name)
+                self._alter_table(database, "ADD %s" % index.create_table_sql())
             else:
                 existing.remove(index.name)
         # Remaining indexes in `existing` are obsolete
         for name in existing:
-            logger.info('        Drop index %s', name)
-            self._alter_table(database, 'DROP INDEX `%s`' % name)
+            logger.info("        Drop index %s", name)
+            self._alter_table(database, "DROP INDEX `%s`" % name)
         # Reindex
         if self.reindex:
-            logger.info('        Build indexes on table')
-            database.raw('OPTIMIZE TABLE $db.`%s` FINAL' % self.table_name)
+            logger.info("        Build indexes on table")
+            database.raw("OPTIMIZE TABLE $db.`%s` FINAL" % self.table_name)
 
     def _get_index_names(self, database):
-        '''
+        """
         Returns a set containing the names of existing indexes in the table.
-        '''
+        """
         import re
-        table_def = database.raw('SHOW CREATE TABLE $db.`%s`' % self.table_name)
-        matches = re.findall(r'\sINDEX\s+`?(.+?)`?\s+', table_def)
+
+        table_def = database.raw("SHOW CREATE TABLE $db.`%s`" % self.table_name)
+        matches = re.findall(r"\sINDEX\s+`?(.+?)`?\s+", table_def)
         return set(matches)
 
 
 class RunPython(Operation):
-    '''
+    """
     A migration operation that executes a Python function.
-    '''
+    """
+
     def __init__(self, func):
-        '''
+        """
         Initializer. The given Python function will be called with a single
         argument - the Database instance to apply the migration to.
-        '''
+        """
         assert callable(func), "'func' argument must be function"
         self._func = func
 
     def apply(self, database):
-        logger.info('    Executing python operation %s', self._func.__name__)
+        logger.info("    Executing python operation %s", self._func.__name__)
         self._func(database)
 
 
 class RunSQL(Operation):
-    '''
+    """
     A migration operation that executes arbitrary SQL statements.
-    '''
+    """
 
     def __init__(self, sql):
-        '''
+        """
         Initializer. The given sql argument must be a valid SQL statement or
         list of statements.
-        '''
+        """
         if isinstance(sql, str):
             sql = [sql]
         assert isinstance(sql, list), "'sql' argument must be string or list of strings"
         self._sql = sql
 
     def apply(self, database):
-        logger.info('    Executing raw SQL operations')
+        logger.info("    Executing raw SQL operations")
         for item in self._sql:
             database.raw(item)
 
 
 class MigrationHistory(Model):
-    '''
+    """
     A model for storing which migrations were already applied to the containing database.
-    '''
+    """
 
     package_name = StringField()
     module_name = StringField()
@@ -269,7 +284,42 @@ class MigrationHistory(Model):
 
     @classmethod
     def table_name(cls):
-        return 'infi_clickhouse_orm_migrations'
+        return "infi_clickhouse_orm_migrations"
+
+class MigrationHistoryReplicated(Model):
+    """
+    A model for storing which migrations were already applied to the containing database.
+    """
+
+    package_name = StringField()
+    module_name = StringField()
+    applied = DateField()
+
+    engine = MergeTree(
+        "applied",
+        ("package_name", "module_name"),
+        replica_table_path="/clickhouse/prod/tables/noshard/{database}/{table}",
+        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 *

src/infi/clickhouse_orm/models.py

@@ -7,7 +7,7 @@ from logging import getLogger
 import pytz
 
 from .fields import Field, StringField
-from .utils import parse_tsv, NO_VALUE, get_subclass_names, arg_to_sql, unescape
+from .utils import on_cluster, parse_tsv, NO_VALUE, get_subclass_names, arg_to_sql, unescape
 from .query import QuerySet
 from .funcs import F
 from .engines import Merge, Distributed
@@ -352,7 +352,7 @@ class Model(metaclass=ModelBase):
         '''
         Returns the SQL statement for creating a table for this model.
         '''
-        parts = ['CREATE TABLE IF NOT EXISTS `%s`.`%s` (' % (db.db_name, cls.table_name())]
+        parts = ['CREATE TABLE IF NOT EXISTS `%s`.`%s` %s (' % (db.db_name, cls.table_name(), on_cluster(db))]
         # Fields
         items = []
         for name, field in cls.fields().items():

src/infi/clickhouse_orm/utils.py

@@ -155,6 +155,12 @@ def get_subclass_names(locals, base_class):
     from inspect import isclass
     return [c.__name__ for c in locals.values() if isclass(c) and issubclass(c, base_class)]
 
+def on_cluster(db):
+    if db.cluster is not None:
+        return "ON CLUSTER '{}'".format(db.cluster)
+    else:
+        return ''
+
 
 class NoValue:
     '''