1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
|
from __future__ import annotations
from typing import Any
from sqlalchemy import Column
from sqlalchemy.exc import CompileError
from sqlalchemy.sql import sqltypes
from sqlalchemy.sql.compiler import DDLCompiler
from clickhouse_connect.cc_sqlalchemy.datatypes.base import ChSqlaType
from clickhouse_connect.cc_sqlalchemy.datatypes.sqltypes import Nullable
from clickhouse_connect.cc_sqlalchemy.sql import format_table
from clickhouse_connect.datatypes.base import TypeDef
from clickhouse_connect.driver.binding import format_str, quote_identifier
class ClickHouseDDLHelper:
dialect_names = ("clickhousedb", "clickhouse")
@classmethod
def get_option(cls, obj: Any, name: str) -> Any:
kwargs = getattr(obj, "kwargs", None)
if kwargs is not None:
for prefix in cls.dialect_names:
key = f"{prefix}_{name}"
if key in kwargs and kwargs[key] is not None:
return kwargs[key]
dialect_options = getattr(obj, "dialect_options", None)
if dialect_options:
for prefix in cls.dialect_names:
options = dialect_options.get(prefix)
if options and options.get(name) is not None:
return options.get(name)
return None
@classmethod
def is_dictionary(cls, table: Any) -> bool:
return getattr(table, "__visit_name__", None) == "dictionary" or cls.get_option(table, "table_type") == "dictionary"
@classmethod
def dictionary_option(cls, table: Any, name: str) -> Any:
attr_name = "primary_key_def" if name == "primary_key" else name
value = getattr(table, attr_name, None)
if value is not None:
return value
return cls.get_option(table, f"dictionary_{name}")
@staticmethod
def explicit_column_nullable(column: Column) -> bool | None:
user_defined = getattr(column, "_user_defined_nullable", None)
if isinstance(user_defined, bool):
return user_defined
return None
@staticmethod
def column_nullable(column: Column) -> bool:
column_type = getattr(column, "type", None)
if isinstance(column_type, ChSqlaType) and column_type.nullable:
return True
explicit_nullable = ClickHouseDDLHelper.explicit_column_nullable(column)
if explicit_nullable is not None:
return explicit_nullable
return False
@staticmethod
def effective_column_type(column: Column):
column_type = column.type
if not isinstance(column_type, ChSqlaType):
return column_type
if column_type.nullable:
return column_type
explicit_nullable = ClickHouseDDLHelper.explicit_column_nullable(column)
if not explicit_nullable:
return column_type
return Nullable(column_type)
@staticmethod
def without_nullable(type_):
if not isinstance(type_, ChSqlaType) or not type_.nullable:
return type_
type_def = type_.type_def
wrappers = tuple(wrapper for wrapper in type_def.wrappers if wrapper != "Nullable")
return type_.__class__(type_def=TypeDef(wrappers, type_def.keys, type_def.values))
@staticmethod
def render_settings(settings: dict[str, Any] | None) -> str:
if not settings:
return ""
return ", ".join(f"{key} = {ClickHouseDDLHelper._render_setting_value(value)}" for key, value in settings.items())
@staticmethod
def render_comment(comment: str | None) -> str:
if comment is None:
return "''"
escaped = comment.replace("'", "''")
return f"'{escaped}'"
@staticmethod
def _render_setting_value(value: Any) -> str:
if isinstance(value, bool):
return "1" if value else "0"
if isinstance(value, (int, float)):
return str(value)
return format_str(str(value))
def column_specification(dialect, column: Column) -> str:
compiler = dialect.ddl_compiler(dialect, None)
return compiler.get_column_specification(column)
class ChDDLCompiler(DDLCompiler):
def visit_create_schema(self, create, **_):
return f"CREATE DATABASE {quote_identifier(create.element)}"
def visit_drop_schema(self, drop, **_):
return f"DROP DATABASE {quote_identifier(drop.element)}"
def visit_create_table(self, create, **_):
table = create.element
if_not_exists = " IF NOT EXISTS" if getattr(create, "if_not_exists", False) else ""
if ClickHouseDDLHelper.is_dictionary(table):
return self._visit_create_dictionary(create, table, if_not_exists)
engine = getattr(table, "engine", None) or ClickHouseDDLHelper.get_option(table, "engine")
if engine is None:
raise CompileError(
f"ClickHouse table '{table.name}' requires an engine — specify e.g. MergeTree(order_by='id') as a table argument"
)
text = f"CREATE TABLE{if_not_exists} {format_table(table)} ("
text += ", ".join([self.get_column_specification(c.element) for c in create.columns])
text += ") " + engine.compile()
if table.comment:
text += f" COMMENT {self.sql_compiler.render_literal_value(table.comment, sqltypes.STRINGTYPE)}"
return text
def _visit_create_dictionary(self, create, dictionary, if_not_exists: str):
text = f"CREATE DICTIONARY{if_not_exists} {format_table(dictionary)} ("
text += ", ".join([self.get_column_specification(c.element) for c in create.columns])
text += ")"
primary_key = ClickHouseDDLHelper.dictionary_option(dictionary, "primary_key")
if primary_key:
text += f" PRIMARY KEY {primary_key}"
source = ClickHouseDDLHelper.dictionary_option(dictionary, "source")
if source:
text += f" SOURCE({source})"
layout = ClickHouseDDLHelper.dictionary_option(dictionary, "layout")
if layout:
layout = layout if "(" in layout else f"{layout}()"
text += f" LAYOUT({layout})"
lifetime = ClickHouseDDLHelper.dictionary_option(dictionary, "lifetime")
if lifetime:
text += f" LIFETIME({lifetime})"
if dictionary.comment:
text += f" COMMENT {self.sql_compiler.render_literal_value(dictionary.comment, sqltypes.STRINGTYPE)}"
return text
def visit_drop_table(self, drop, **_):
table = drop.element
if_exists = " IF EXISTS" if getattr(drop, "if_exists", False) else ""
if ClickHouseDDLHelper.is_dictionary(table):
return f"DROP DICTIONARY{if_exists} {format_table(table)}"
return f"DROP TABLE{if_exists} {format_table(table)}"
def visit_add_column(self, create, **_):
return f"ALTER TABLE {format_table(create.element)} ADD COLUMN {self.get_column_specification(create.column)}"
def visit_drop_column(self, drop, **_):
return f"ALTER TABLE {format_table(drop.element)} DROP COLUMN {quote_identifier(drop.column.name)}"
def get_column_specification(self, column: Column, **_):
text = f"{quote_identifier(column.name)} {ClickHouseDDLHelper.effective_column_type(column).compile()}"
materialized = ClickHouseDDLHelper.get_option(column, "materialized")
alias = ClickHouseDDLHelper.get_option(column, "alias")
if materialized is not None:
text += f" MATERIALIZED {self.render_default_string(materialized)}"
return text
if alias is not None:
text += f" ALIAS {self.render_default_string(alias)}"
return text
default = self.get_column_default_string(column)
if default is not None:
text += f" DEFAULT {default}"
codec = ClickHouseDDLHelper.get_option(column, "codec")
if codec is not None:
codec_sql = codec if isinstance(codec, str) else ", ".join(str(item) for item in codec)
text += f" CODEC({codec_sql})"
ttl = ClickHouseDDLHelper.get_option(column, "ttl")
if ttl is not None:
text += f" TTL {self.render_default_string(ttl)}"
if column.comment:
text += f" COMMENT {self.sql_compiler.render_literal_value(column.comment, sqltypes.STRINGTYPE)}"
return text
|
