Support [MergeTree settings](https://clickhouse.com/docs/en/engines/table-engines/mergetree-family/mergetree#settings) in creating table

Author: jayvynl

CHANGELOG.md

@@ -1,5 +1,6 @@
 ### 1.1.1
 - [Black](https://github.com/psf/black) code style.
+- Support [MergeTree settings](https://clickhouse.com/docs/en/engines/table-engines/mergetree-family/mergetree#settings) in creating table.
 
 ### 1.1.0
 - Change `AutoFiled` and `SmallAutoField` to clickhouse `Int64`, so that id worker can generate value for them.

README.md

@@ -186,7 +186,10 @@ class Event(models.ClickhouseModel):
         db_table = 'event'
         engine = models.ReplacingMergeTree(
             order_by=['id'],
-            partition_by=Func('timestamp', function='toYYYYMMDD')
+            partition_by=Func('timestamp', function='toYYYYMMDD'),
+            index_granularity=1024,
+            index_granularity_bytes=1 << 20,
+            enable_mixed_granularity_parts=1,
         )
         indexes = [
             models.Index(

clickhouse_backend/backend/schema.py

@@ -188,6 +188,8 @@ def _model_indexes_sql(self, model):
         return output
 
     def _get_expression(self, model, *expressions):
+        if not expressions:
+            return ""
         index_expressions = []
         for expression in expressions:
             index_expression = IndexExpression(expression)
@@ -197,43 +199,30 @@ def _get_expression(self, model, *expressions):
 
         query = Query(model, alias_cols=False)
         expression_list = ExpressionList(*index_expressions).resolve_expression(query)
-        compiler = query.get_compiler(
-            connection=self.connection,
-        )
+        compiler = query.get_compiler(connection=self.connection)
         return Expressions(
             model._meta.db_table, expression_list, compiler, self.quote_value
         )
 
     def _model_extra_sql(self, model, engine):
-        extra_parts = []
         from clickhouse_backend.models.engines import BaseMergeTree
 
         if isinstance(engine, BaseMergeTree):
             order_by = engine.order_by
             partition_by = engine.partition_by
             primary_key = engine.primary_key
 
-            if order_by:
-                if not isinstance(order_by, (list, tuple)):
-                    order_by = [order_by]
-                extra_parts.append(
-                    "ORDER BY (%s)" % self._get_expression(model, *order_by)
-                )
-            else:
-                extra_parts.append("ORDER BY tuple()")
+            if order_by is not None:
+                yield "ORDER BY (%s)" % self._get_expression(model, *order_by)
             if partition_by:
-                if not isinstance(partition_by, (list, tuple)):
-                    partition_by = [partition_by]
-                extra_parts.append(
-                    "PARTITION BY (%s)" % self._get_expression(model, *partition_by)
-                )
-            if primary_key:
-                if not isinstance(primary_key, (list, tuple)):
-                    primary_key = [primary_key]
-                extra_parts.append(
-                    "PRIMARY KEY (%s)" % self._get_expression(model, *primary_key)
-                )
-        return extra_parts
+                yield "PARTITION BY (%s)" % self._get_expression(model, *partition_by)
+            if primary_key is not None:
+                yield "PRIMARY KEY (%s)" % self._get_expression(model, *primary_key)
+            if engine.settings:
+                result = []
+                for setting, value in engine.settings.items():
+                    result.append(f"{setting}={self.quote_value(value)}")
+                yield "SETTINGS %s" % ", ".join(result)
 
     def add_field(self, model, field):
         """

clickhouse_backend/models/engines.py

@@ -1,4 +1,7 @@
+from itertools import zip_longest
+
 from django.db.models.expressions import Func, Value
+from django.utils.itercompat import is_iterable
 
 __all__ = [
     "Engine",
@@ -20,7 +23,35 @@
 ]
 
 
+def _check_positive(value, name):
+    if not isinstance(value, int) and value <= 0:
+        raise ValueError(f"{name} must be positive integer.")
+    return value
+
+
+def _check_not_negative(value, name):
+    if not isinstance(value, int) and value < 0:
+        raise ValueError(f"{name} must not be negative.")
+    return value
+
+
+def _check_bool(value, name):
+    if value not in (0, 1):
+        raise ValueError(f"{name} must be one of (0, 1, True, False)")
+    return int(value)
+
+
+def _check_str(value, name):
+    if not isinstance(value, str):
+        raise ValueError(f"{name} must be string")
+    return value
+
+
 class Engine(Func):
+    @property
+    def function(self):
+        return self.__class__.__name__
+
     def deconstruct(self):
         path, args, kwargs = super().deconstruct()
         if path.startswith("clickhouse_backend.models.engines"):
@@ -31,99 +62,197 @@ def deconstruct(self):
 
 
 class BaseMergeTree(Engine):
-    def __init__(self, *expressions, output_field=None, **extra):
-        self.order_by = extra.pop("order_by", None)
-        assert self.order_by is not None, "order_by is required by MergeTree family."
-        self.partition_by = extra.pop("partition_by", None)
-        self.primary_key = extra.pop("primary_key", None)
-        super().__init__(*expressions, output_field=output_field, **extra)
+    max_arity = None  # The max number of arguments the function accepts.
+    # https://clickhouse.com/docs/en/engines/table-engines/mergetree-family/mergetree#settings
+    setting_types = {
+        _check_positive: [
+            "index_granularity",
+            "min_index_granularity_bytes",
+            "merge_max_block_size",
+            "min_bytes_for_wide_part",
+            "min_rows_for_wide_part",
+            "max_parts_in_total",
+            "max_compress_block_size",
+            "min_compress_block_size",
+            "max_partitions_to_read",
+        ],
+        _check_not_negative: [
+            "index_granularity_bytes",
+            "min_merge_bytes_to_use_direct_io",
+            "merge_with_ttl_timeout",
+            "merge_with_recompression_ttl_timeout",
+            "try_fetch_recompressed_part_timeout",
+        ],
+        _check_bool: [
+            "enable_mixed_granularity_parts",
+            "use_minimalistic_part_header_in_zookeeper",
+            "write_final_mark",
+        ],
+        _check_str: ["storage_policy"],
+    }
+
+    def __init__(
+        self,
+        *expressions,
+        order_by=None,
+        partition_by=None,
+        primary_key=None,
+        **settings,
+    ):
+        if self.max_arity is not None and len(expressions) > self.max_arity:
+            raise TypeError(
+                "'%s' takes at most %s %s (%s given)"
+                % (
+                    self.__class__.__name__,
+                    self.max_arity,
+                    "argument" if self.max_arity == 1 else "arguments",
+                    len(expressions),
+                )
+            )
+
+        assert (
+            order_by is not None or primary_key is not None
+        ), "order_by or primary_key is missing."
+        self.order_by = order_by
+        self.primary_key = primary_key
+        self.partition_by = partition_by
+
+        if order_by is not None:
+            if isinstance(order_by, str) or not is_iterable(order_by):
+                self.order_by = [order_by]
+            if any(o is None for o in self.order_by):
+                raise ValueError("None is not allowed in order_by")
+
+        if primary_key is not None:
+            if isinstance(primary_key, str) or not is_iterable(primary_key):
+                self.primary_key = [primary_key]
+            if any(o is None for o in self.primary_key):
+                raise ValueError("None is not allowed in primary_key")
+
+        # https://clickhouse.com/docs/en/engines/table-engines/mergetree-family/mergetree#choosing-a-primary-key-that-differs-from-the-sorting-key
+        # primary key expression tuple must be a prefix of the sorting key expression tuple.
+        if self.order_by is not None and self.primary_key is not None:
+            for o, p in zip_longest(self.order_by, self.primary_key):
+                if p is None:
+                    break
+                if p != o:
+                    raise ValueError("primary_key must be a prefix of order_by")
+
+        normalized_settings = {}
+        for setting, value in settings.items():
+            for validate, keys in self.setting_types.items():
+                if setting in keys:
+                    normalized_settings[setting] = validate(value, setting)
+                    break
+            else:
+                raise TypeError(f"{setting} is not a valid setting.")
+        self.settings = normalized_settings
+        super().__init__(*expressions)
 
 
 class MergeTree(BaseMergeTree):
-    function = "MergeTree"
     arity = 0
 
 
 class ReplacingMergeTree(BaseMergeTree):
-    function = "ReplacingMergeTree"
+    max_arity = 2
 
 
 class SummingMergeTree(BaseMergeTree):
-    function = "SummingMergeTree"
+    pass
 
 
 class AggregatingMergeTree(BaseMergeTree):
-    function = "AggregatingMergeTree"
     arity = 0
 
 
 class CollapsingMergeTree(BaseMergeTree):
-    function = "CollapsingMergeTree"
     arity = 1
 
 
 class VersionedCollapsingMergeTree(BaseMergeTree):
-    function = "CollapsingMergeTree"
     arity = 2
 
 
 class GraphiteMergeTree(BaseMergeTree):
-    function = "GraphiteMergeTree"
     arity = 1
 
+    def __init__(self, *expressions, **extra):
+        if expressions:
+            expressions = (Value(expressions[0]), *expressions[1:])
+        super().__init__(*expressions, **extra)
+
 
 class ReplicatedMixin:
     def __init__(self, *expressions, **extra):
         if self.arity is not None and len(expressions) != self.arity + 2:
             raise TypeError(
-                "'%s' takes exactly %s %s (%s given)"
+                "'%s' takes exactly %s arguments (%s given)"
                 % (
                     self.__class__.__name__,
                     self.arity + 2,
-                    "arguments",
                     len(expressions),
                 )
             )
         if self.arity is None and len(expressions) < 2:
             raise TypeError(
-                "'%s' takes at least %s %s (%s given)"
+                "'%s' takes at least 2 arguments (%s given)"
                 % (
                     self.__class__.__name__,
-                    2,
-                    "arguments",
                     len(expressions),
                 )
             )
-        replicated_params = tuple(Value(arg) for arg in self.expressions[:2])
+        if self.max_arity is not None and len(expressions) > self.max_arity + 2:
+            raise TypeError(
+                "'%s' takes at most %s arguments (%s given)"
+                % (
+                    self.__class__.__name__,
+                    self.max_arity + 2,
+                    len(expressions),
+                )
+            )
+        replicated_params = map(Value, self.expressions[:2])
         super().__init__(*expressions[2:], **extra)
-        self.expressions = replicated_params + self.expressions
+        self.expressions = (*replicated_params, *self.expressions)
 
 
-class ReplicatedMergeTree(ReplicatedMixin, MergeTree):
-    function = "ReplicatedMergeTree"
+class ReplicatedMergeTree(MergeTree):
+    def __init__(self, *expressions, **extra):
+        # https://github.com/ClickHouse/ClickHouse/issues/8675
+        # https://clickhouse.com/docs/en/engines/table-engines/mergetree-family/replication#creating-replicated-tables
+        # You can specify default arguments for Replicated table engine in the server configuration file.
+        # In this case, you can omit arguments when creating tables.
+        if expressions:
+            if len(expressions) != 2:
+                raise TypeError(
+                    "'ReplicatedMergeTree' takes at 0 or 2 arguments (%s given)"
+                    % len(expressions)
+                )
+            expressions = map(Value, expressions)
+        super().__init__(*expressions, **extra)
 
 
 class ReplicatedReplacingMergeTree(ReplicatedMixin, ReplacingMergeTree):
-    function = "ReplicatedReplacingMergeTree"
+    pass
 
 
 class ReplicatedSummingMergeTree(ReplicatedMixin, SummingMergeTree):
-    function = "ReplicatedSummingMergeTree"
+    pass
 
 
 class ReplicatedAggregatingMergeTree(ReplicatedMixin, AggregatingMergeTree):
-    function = "ReplicatedAggregatingMergeTree"
+    pass
 
 
 class ReplicatedCollapsingMergeTree(ReplicatedMixin, CollapsingMergeTree):
-    function = "ReplicatedCollapsingMergeTree"
+    pass
 
 
 class ReplicatedVersionedCollapsingMergeTree(
     ReplicatedMixin, VersionedCollapsingMergeTree
 ):
-    function = "ReplicatedCollapsingMergeTree"
+    pass
 
 
 class ReplicatedGraphiteMergeTree(ReplicatedMixin, GraphiteMergeTree):
-    function = "ReplicatedGraphiteMergeTree"
+    pass

example/testapp/models.py

@@ -25,7 +25,11 @@ class Meta:
         ordering = ["-id"]
         db_table = "event"
         engine = models.ReplacingMergeTree(
-            order_by=["id"], partition_by=models.toYYYYMMDD("timestamp")
+            order_by=["id"],
+            partition_by=models.toYYYYMMDD("timestamp"),
+            index_granularity=1024,
+            index_granularity_bytes=1 << 20,
+            enable_mixed_granularity_parts=1,
         )
         indexes = [
             models.Index(