Skip to content

Commit 1ac33c2

Browse files
authored
Merge branch 'main' into fix/standalone-audit-last-in-file-macros
2 parents dad0eb8 + d69c262 commit 1ac33c2

10 files changed

Lines changed: 863 additions & 45 deletions

File tree

‎sqlmesh/core/engine_adapter/clickhouse.py‎

Lines changed: 37 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,21 @@ def inject_virtual_catalog(self, gateway: str) -> None:
6161
configured = self._extra_config.get("virtual_catalog")
6262
self._default_catalog = f"__{gateway}__" if configured is None else configured
6363

64+
def _to_sql(self, expression: exp.Expr, quote: bool = True, **kwargs: t.Any) -> str:
65+
"""Render SQL without the virtual catalog, which ClickHouse does not support."""
66+
if self._default_catalog and any(self._virtual_catalog_references(expression)):
67+
expression = expression.copy()
68+
for reference in list(self._virtual_catalog_references(expression)):
69+
reference.set("catalog", None)
70+
return super()._to_sql(expression, quote=quote, **kwargs)
71+
72+
def _virtual_catalog_references(self, expression: exp.Expr) -> t.Iterator[exp.Expr]:
73+
return (
74+
reference
75+
for reference in expression.find_all(exp.Table, exp.Column)
76+
if reference.text("catalog") == self._default_catalog
77+
)
78+
6479
@property
6580
def engine_run_mode(self) -> EngineRunMode:
6681
if self._extra_config.get("cloud_mode"):
@@ -502,8 +517,14 @@ def _create_table_like(
502517
**kwargs: t.Any,
503518
) -> None:
504519
"""Create table with identical structure as source table"""
520+
target_table_sql = self._strip_virtual_catalog(target_table_name).sql(
521+
dialect=self.dialect, identify=True
522+
)
523+
source_table_sql = self._strip_virtual_catalog(source_table_name).sql(
524+
dialect=self.dialect, identify=True
525+
)
505526
self.execute(
506-
f"CREATE TABLE {target_table_name}{self._on_cluster_sql()} AS {source_table_name}"
527+
f"CREATE TABLE {target_table_sql}{self._on_cluster_sql()} AS {source_table_sql}"
507528
)
508529

509530
def _get_partition_ids(
@@ -648,7 +669,7 @@ def _strip_virtual_catalog(self, name: "TableName") -> exp.Table:
648669
SQL is sent to the wire, since ClickHouse only supports a two-level
649670
``[database].[table]`` naming scheme.
650671
"""
651-
table = exp.to_table(name)
672+
table = exp.to_table(name, dialect=self.dialect)
652673
if self._default_catalog and table.catalog == self._default_catalog:
653674
table.set("catalog", None)
654675
return table
@@ -660,8 +681,12 @@ def _exchange_tables(
660681
) -> None:
661682
from clickhouse_connect.driver.exceptions import DatabaseError # type: ignore
662683

663-
old_table_sql = exp.to_table(old_table_name).sql(dialect=self.dialect, identify=True)
664-
new_table_sql = exp.to_table(new_table_name).sql(dialect=self.dialect, identify=True)
684+
old_table_sql = self._strip_virtual_catalog(old_table_name).sql(
685+
dialect=self.dialect, identify=True
686+
)
687+
new_table_sql = self._strip_virtual_catalog(new_table_name).sql(
688+
dialect=self.dialect, identify=True
689+
)
665690

666691
try:
667692
self.execute(
@@ -685,8 +710,12 @@ def _rename_table(
685710
old_table_name: TableName,
686711
new_table_name: TableName,
687712
) -> None:
688-
old_table_sql = exp.to_table(old_table_name).sql(dialect=self.dialect, identify=True)
689-
new_table_sql = exp.to_table(new_table_name).sql(dialect=self.dialect, identify=True)
713+
old_table_sql = self._strip_virtual_catalog(old_table_name).sql(
714+
dialect=self.dialect, identify=True
715+
)
716+
new_table_sql = self._strip_virtual_catalog(new_table_name).sql(
717+
dialect=self.dialect, identify=True
718+
)
690719

691720
self.execute(f"RENAME TABLE {old_table_sql} TO {new_table_sql}{self._on_cluster_sql()}")
692721

@@ -974,7 +1003,7 @@ def _build_view_properties_exp(
9741003
def _build_create_comment_table_exp(
9751004
self, table: exp.Table, table_comment: str, table_kind: str, **kwargs: t.Any
9761005
) -> exp.Comment | str:
977-
table_sql = table.sql(dialect=self.dialect, identify=True)
1006+
table_sql = self._strip_virtual_catalog(table).sql(dialect=self.dialect, identify=True)
9781007

9791008
truncated_comment = self._truncate_table_comment(table_comment)
9801009
comment_sql = exp.Literal.string(truncated_comment).sql(dialect=self.dialect)
@@ -989,7 +1018,7 @@ def _build_create_comment_column_exp(
9891018
table_kind: str = "TABLE",
9901019
**kwargs: t.Any,
9911020
) -> exp.Comment | str:
992-
table_sql = table.sql(dialect=self.dialect, identify=True)
1021+
table_sql = self._strip_virtual_catalog(table).sql(dialect=self.dialect, identify=True)
9931022
column_sql = exp.to_column(column_name).sql(dialect=self.dialect, identify=True)
9941023

9951024
truncated_comment = self._truncate_table_comment(column_comment)

‎sqlmesh/core/engine_adapter/duckdb.py‎

Lines changed: 34 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -173,16 +173,8 @@ def _create_table(
173173
track_rows_processed: bool = True,
174174
**kwargs: t.Any,
175175
) -> None:
176-
catalog = self.get_current_catalog()
177-
catalog_type_tuple = self.fetchone(
178-
exp.select("type")
179-
.from_("duckdb_databases()")
180-
.where(exp.column("database_name").eq(catalog))
181-
)
182-
catalog_type = catalog_type_tuple[0] if catalog_type_tuple else None
183-
184176
partitioned_by_exps = None
185-
if catalog_type == "ducklake":
177+
if self._get_catalog_type(self.get_current_catalog()) == "ducklake":
186178
partitioned_by_exps = kwargs.pop("partitioned_by", None)
187179

188180
super()._create_table(
@@ -215,6 +207,39 @@ def _create_table(
215207
)
216208
self.execute(f"ALTER TABLE {table_name_str} SET PARTITIONED BY ({partitioned_by_str});")
217209

210+
def _drop_object(
211+
self,
212+
name: TableName | SchemaName,
213+
exists: bool = True,
214+
kind: str = "TABLE",
215+
cascade: bool = False,
216+
**drop_args: t.Any,
217+
) -> None:
218+
# DuckLake catalogs do not implement DROP TABLE / DROP VIEW ... CASCADE and raise
219+
# "Cascade Drop not supported in DuckLake". Views in DuckDB are late-binding, so
220+
# dropping the underlying table without CASCADE is safe there.
221+
if cascade and kind.upper() in ("TABLE", "VIEW"):
222+
catalog = exp.to_table(name).catalog or self.get_current_catalog()
223+
if self._get_catalog_type(catalog) == "ducklake":
224+
cascade = False
225+
226+
super()._drop_object(name=name, exists=exists, kind=kind, cascade=cascade, **drop_args)
227+
228+
def _get_catalog_type(self, catalog: t.Optional[str]) -> t.Optional[str]:
229+
"""Returns the type of the given catalog (e.g. 'duckdb', 'ducklake') as reported by duckdb_databases()."""
230+
if not catalog:
231+
return None
232+
catalog_type_tuple = self.fetchone(
233+
exp.select("type")
234+
.from_("duckdb_databases()")
235+
.where(
236+
exp.func("LOWER", exp.column("database_name")).eq(
237+
exp.func("LOWER", exp.Literal.string(catalog))
238+
)
239+
)
240+
)
241+
return catalog_type_tuple[0] if catalog_type_tuple else None
242+
218243
@property
219244
def _is_motherduck(self) -> bool:
220245
return self._extra_config.get("is_motherduck", False)

‎sqlmesh/core/renderer.py‎

Lines changed: 88 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,65 @@
4545
logger = logging.getLogger(__name__)
4646

4747

48+
class TableMapping(t.Dict[str, str]):
49+
"""A table name mapping that caches the dialect-normalized form of its keys.
50+
51+
`exp.replace_tables` normalizes every key of the mapping it's given, so resolving a single
52+
table against a mapping of every model in an environment costs O(N). Resolving it against
53+
this mapping costs a dictionary lookup, since each key is normalized once per dialect.
54+
"""
55+
56+
def __init__(self, *args: t.Any, **kwargs: t.Any):
57+
super().__init__(*args, **kwargs)
58+
self._normalized_keys: t.Dict[DialectType, t.Dict[str, str]] = {}
59+
60+
def normalized_keys(self, dialect: DialectType) -> t.Dict[str, str]:
61+
"""Returns a mapping from each normalized key to the last key that normalizes to it."""
62+
normalized_keys = self._normalized_keys.get(dialect)
63+
if normalized_keys is None:
64+
normalized_keys = {exp.normalize_table_name(key, dialect=dialect): key for key in self}
65+
self._normalized_keys[dialect] = normalized_keys
66+
return normalized_keys
67+
68+
def __setitem__(self, key: str, value: str) -> None:
69+
self._normalized_keys.clear()
70+
super().__setitem__(key, value)
71+
72+
def __delitem__(self, key: str) -> None:
73+
self._normalized_keys.clear()
74+
super().__delitem__(key)
75+
76+
def __ior__(self, other: t.Any) -> TableMapping: # type: ignore[override,misc]
77+
self._normalized_keys.clear()
78+
return super().__ior__(other)
79+
80+
def update(self, *args: t.Any, **kwargs: t.Any) -> None:
81+
self._normalized_keys.clear()
82+
super().update(*args, **kwargs)
83+
84+
def setdefault(self, key: str, default: str) -> str: # type: ignore[override]
85+
self._normalized_keys.clear()
86+
return super().setdefault(key, default)
87+
88+
def pop(self, key: str, *args: t.Any) -> t.Any:
89+
self._normalized_keys.clear()
90+
return super().pop(key, *args)
91+
92+
def popitem(self) -> t.Tuple[str, str]:
93+
self._normalized_keys.clear()
94+
return super().popitem()
95+
96+
def clear(self) -> None:
97+
self._normalized_keys.clear()
98+
super().clear()
99+
100+
101+
def _normalize_keys(mapping: t.Dict[str, str], dialect: DialectType) -> t.Dict[str, str]:
102+
if isinstance(mapping, TableMapping):
103+
return mapping.normalized_keys(dialect)
104+
return {exp.normalize_table_name(key, dialect=dialect): key for key in mapping}
105+
106+
48107
class BaseExpressionRenderer:
49108
def __init__(
50109
self,
@@ -325,20 +384,35 @@ def update_cache(self, expression: t.Optional[exp.Expr]) -> None:
325384

326385
def _resolve_table(
327386
self,
328-
table_name: str | exp.Expr,
387+
table_name: str,
329388
snapshots: t.Optional[t.Dict[str, Snapshot]] = None,
330389
table_mapping: t.Optional[t.Dict[str, str]] = None,
331390
deployability_index: t.Optional[DeployabilityIndex] = None,
332391
) -> exp.Table:
333-
table = exp.replace_tables(
334-
t.cast(exp.Table, exp.maybe_parse(table_name, into=exp.Table, dialect=self._dialect)),
335-
{
336-
**self._to_table_mapping((snapshots or {}).values(), deployability_index),
337-
**(table_mapping or {}),
338-
},
339-
dialect=self._dialect,
340-
copy=False,
392+
table = t.cast(
393+
exp.Table, exp.maybe_parse(table_name, into=exp.Table, dialect=self._dialect)
341394
)
395+
396+
mapping: t.Dict[str, str] = {}
397+
if table_mapping:
398+
# An explicit mapping takes precedence over snapshots, so when one of its keys matches
399+
# the table, that key alone decides the result. Among equivalent keys, the last wins.
400+
key = _normalize_keys(table_mapping, self._dialect).get(
401+
exp.normalize_table_name(table, dialect=self._dialect)
402+
)
403+
if key is not None:
404+
mapping = {key: table_mapping[key]}
405+
406+
if not mapping and snapshots:
407+
# An exact FQN match avoids scanning unrelated snapshots.
408+
snapshot = snapshots.get(table_name)
409+
# Keys normalized under different dialects may differ in casing or quoting.
410+
# Fall back to the full mapping so exp.replace_tables can reconcile them.
411+
mapping = self._to_table_mapping(
412+
[snapshot] if snapshot else snapshots.values(), deployability_index
413+
)
414+
415+
table = exp.replace_tables(table, mapping, dialect=self._dialect, copy=False)
342416
# We quote the table here to mimic the behavior of _resolve_tables, otherwise we may end
343417
# up normalizing twice, because _to_table_mapping returns the mapped names unquoted.
344418
return (
@@ -363,6 +437,11 @@ def _resolve_tables(
363437

364438
expression = expression.copy()
365439
with self._normalize_and_quote(expression) as expression:
440+
# An expression with no table (e.g. most session or virtual properties) has nothing
441+
# to expand or replace, so skip building the O(N) expand set and mapping.
442+
if not expression.find(exp.Table):
443+
return expression
444+
366445
snapshots = snapshots or {}
367446
table_mapping = table_mapping or {}
368447
mapping = {

‎sqlmesh/core/snapshot/definition.py‎

Lines changed: 9 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
from sqlmesh.core.model import Model, ModelKindMixin, ModelKindName, ViewKind, CustomKind
2525
from sqlmesh.core.model.definition import _Model
2626
from sqlmesh.core.node import IntervalUnit, NodeType
27+
from sqlmesh.core.renderer import TableMapping
2728
from sqlmesh.utils import sanitize_name, unique
2829
from sqlmesh.utils.dag import DAG
2930
from sqlmesh.utils.date import (
@@ -2007,14 +2008,17 @@ def to_view_mapping(
20072008
environment_naming_info: EnvironmentNamingInfo,
20082009
default_catalog: t.Optional[str] = None,
20092010
dialect: t.Optional[str] = None,
2010-
) -> t.Dict[str, str]:
2011-
return {
2012-
snapshot.name: snapshot.display_name(
2013-
environment_naming_info, default_catalog=default_catalog, dialect=dialect
2011+
) -> TableMapping:
2012+
return TableMapping(
2013+
(
2014+
snapshot.name,
2015+
snapshot.display_name(
2016+
environment_naming_info, default_catalog=default_catalog, dialect=dialect
2017+
),
20142018
)
20152019
for snapshot in snapshots
20162020
if snapshot.is_model
2017-
}
2021+
)
20182022

20192023

20202024
def has_paused_forward_only(

‎sqlmesh/core/snapshot/evaluator.py‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -312,6 +312,8 @@ def promote(
312312
self._get_virtual_data_objects(target_snapshots, environment_naming_info)
313313

314314
deployability_index = deployability_index or DeployabilityIndex.all_deployable()
315+
# Renderers look snapshots up by model name, not by SnapshotId.
316+
snapshots_by_name = {s.name: s for s in (snapshots or {}).values()}
315317
with self.concurrent_context():
316318
concurrent_apply_to_snapshots(
317319
target_snapshots,
@@ -320,7 +322,7 @@ def promote(
320322
start=start,
321323
end=end,
322324
execution_time=execution_time,
323-
snapshots=snapshots,
325+
snapshots=snapshots_by_name,
324326
table_mapping=table_mapping,
325327
environment_naming_info=environment_naming_info,
326328
deployability_index=deployability_index, # type: ignore
@@ -1260,7 +1262,7 @@ def _promote_snapshot(
12601262
start: t.Optional[TimeLike] = None,
12611263
end: t.Optional[TimeLike] = None,
12621264
execution_time: t.Optional[TimeLike] = None,
1263-
snapshots: t.Optional[t.Dict[SnapshotId, Snapshot]] = None,
1265+
snapshots: t.Optional[t.Dict[str, Snapshot]] = None,
12641266
table_mapping: t.Optional[t.Dict[str, str]] = None,
12651267
) -> None:
12661268
if not snapshot.is_model:
@@ -1299,9 +1301,9 @@ def _promote_snapshot(
12991301
**render_kwargs,
13001302
)
13011303

1302-
snapshot_by_name = {s.name: s for s in (snapshots or {}).values()}
1303-
render_kwargs["snapshots"] = snapshot_by_name
1304-
adapter.execute(snapshot.model.render_on_virtual_update(**render_kwargs))
1304+
adapter.execute(
1305+
snapshot.model.render_on_virtual_update(snapshots=snapshots, **render_kwargs)
1306+
)
13051307

13061308
if on_complete is not None:
13071309
on_complete(snapshot)

0 commit comments

Comments
 (0)