Skip to content

Commit f1cfe4c

Browse files
committed
fix(clickhouse): preserve cluster execution for partition alters
Signed-off-by: mday-io <mdaytn@gmail.com>
1 parent cf156d0 commit f1cfe4c

2 files changed

Lines changed: 47 additions & 3 deletions

File tree

‎sqlmesh/core/engine_adapter/clickhouse.py‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -771,7 +771,17 @@ def alter_table(
771771
altered_table = (
772772
alter_expression.this if isinstance(alter_expression.this, exp.Table) else None
773773
)
774-
if self._should_use_on_cluster(altered_table):
774+
# Replicated databases propagate metadata ALTERs, but not partition
775+
# operations. Partition overwrites must still reach every shard.
776+
alters_partitions = any(
777+
isinstance(action, (exp.DropPartition, exp.ReplacePartition))
778+
for action in alter_expression.args.get("actions", [])
779+
)
780+
if (
781+
self.engine_run_mode.is_cluster
782+
if alters_partitions
783+
else self._should_use_on_cluster(altered_table)
784+
):
775785
alter_expression.set(
776786
"cluster", exp.OnCluster(this=exp.to_identifier(self.cluster))
777787
)

‎tests/core/engine_adapter/test_clickhouse.py‎

Lines changed: 36 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1954,8 +1954,12 @@ def test_alter_table_decides_on_cluster_per_expression(
19541954

19551955
adapter.alter_table(
19561956
[
1957-
parse_one("ALTER TABLE a_db.t ADD COLUMN x UInt64", dialect="clickhouse"),
1958-
parse_one("ALTER TABLE r_db.t ADD COLUMN x UInt64", dialect="clickhouse"),
1957+
parse_one("ALTER TABLE a_db.t ADD COLUMN x UInt64", dialect="clickhouse").assert_is(
1958+
exp.Alter
1959+
),
1960+
parse_one("ALTER TABLE r_db.t ADD COLUMN x UInt64", dialect="clickhouse").assert_is(
1961+
exp.Alter
1962+
),
19591963
]
19601964
)
19611965

@@ -1964,6 +1968,36 @@ def test_alter_table_decides_on_cluster_per_expression(
19641968
assert "ON CLUSTER" not in calls[1]
19651969

19661970

1971+
@pytest.mark.parametrize("cluster", [None, "my_cluster"])
1972+
@pytest.mark.parametrize("database_engine", ["Atomic", "Replicated"])
1973+
@pytest.mark.parametrize("operation", ["drop", "replace", "both"])
1974+
def test_alter_partition_keeps_on_cluster(
1975+
make_mocked_engine_adapter: t.Callable, mocker, cluster, database_engine, operation
1976+
):
1977+
adapter = make_mocked_engine_adapter(ClickhouseEngineAdapter, cluster=cluster)
1978+
mocker.patch.object(ClickhouseEngineAdapter, "_has_replicated_database", True)
1979+
mocker.patch.object(ClickhouseEngineAdapter, "_database_engine", return_value=database_engine)
1980+
1981+
# Use the same AST builder as incremental partition overwrites.
1982+
alter = adapter._build_alter_partition_exp(
1983+
exp.to_table("my_db.target"),
1984+
exp.to_table("my_db.source"),
1985+
{"1"} if operation in ("replace", "both") else set(),
1986+
{"2"} if operation in ("drop", "both") else set(),
1987+
)
1988+
adapter.alter_table([alter])
1989+
1990+
actions = []
1991+
if operation in ("replace", "both"):
1992+
actions.append('REPLACE PARTITION ID \'1\' FROM "my_db"."source"')
1993+
if operation in ("drop", "both"):
1994+
actions.append("DROP PARTITION ID '2'")
1995+
cluster_sql = ' ON CLUSTER "my_cluster"' if cluster else ""
1996+
assert to_sql_calls(adapter) == [
1997+
f'ALTER TABLE "my_db"."target"{cluster_sql} {", ".join(actions)}'
1998+
]
1999+
2000+
19672001
def test_cross_engine_rename_and_exchange_are_refused(
19682002
make_mocked_engine_adapter: t.Callable, mocker
19692003
):

0 commit comments

Comments
 (0)