fix(clickhouse): suppress ON CLUSTER for object DDL in a Replicated database

docs/integrations/engines/clickhouse.md

@@ -49,7 +49,7 @@ In general, you must be connected to a ClickHouse server to execute commands. By
 
 To associate an object with a cluster, DDL commands that create or modify it must include the text `ON CLUSTER [your cluster name]`.
 
-If you provide a cluster name in your SQLMesh connection configuration, SQLMesh will automatically inject the `ON CLUSTER` statement into the DDL commands for all objects created while executing the project. We provide more information about clusters in SQLMesh [below](#cluster-specification).
+If you provide a cluster name in your SQLMesh connection configuration, SQLMesh automatically adds `ON CLUSTER` to object DDL, except for metadata DDL targeting `Replicated` databases, which replicate those changes themselves. We provide more information about clusters in SQLMesh [below](#cluster-specification).
 
 #### ClickHouse Cloud mode
 
@@ -96,7 +96,9 @@ For example, we could create a table `my_schema.my_table` on cluster `TheCluster
 
 To create SQLMesh objects on a cluster, provide the cluster name to the `cluster` key in the SQLMesh connection definition (see all connection parameters [below](#localbuilt-in-scheduler)).
 
-SQLMesh will automatically inject the `ON CLUSTER` clause and cluster name you provide into all project DDL statements.
+SQLMesh automatically adds the `ON CLUSTER` clause and cluster name you provide to object DDL, except for metadata DDL targeting `Replicated` databases. SQLMesh retains `ON CLUSTER` for database creation and removal and for partition operations.
+
+Partition overwrites in multi-shard `Replicated` databases remain unvalidated. ClickHouse can replicate an ALTER containing both `REPLACE PARTITION` and `DROP PARTITION`, which can conflict with `ON CLUSTER`. This behavior predates the metadata DDL policy described above and depends on the ClickHouse version.
 
 ## Model definition
 

sqlmesh/core/engine_adapter/base.py

@@ -1043,12 +1043,15 @@ def _build_create_table_exp(
     ) -> exp.Create:
         exists = False if replace else exists
         catalog_name = None
+        target_table: t.Optional[exp.Table] = None
         if not isinstance(table_name_or_schema, exp.Schema):
             table_name_or_schema = exp.to_table(table_name_or_schema)
             catalog_name = table_name_or_schema.catalog
+            target_table = table_name_or_schema
         else:
             if isinstance(table_name_or_schema.this, exp.Table):
                 catalog_name = table_name_or_schema.this.catalog
+                target_table = table_name_or_schema.this
 
         properties = (
             self._build_table_properties_exp(
@@ -1057,6 +1060,9 @@ def _build_create_table_exp(
                 target_columns_to_types=target_columns_to_types,
                 table_description=table_description,
                 table_kind=table_kind,
+                # Passed so an adapter can vary properties by target object, not just by
+                # connection. Additive: every override accepts **kwargs.
+                table=target_table,
             )
             if kwargs or table_description
             else None
@@ -1364,6 +1370,8 @@ def create_view(
                 else None
             ),
             physical_cluster=create_kwargs.pop("physical_cluster", None),
+            # See the note on the table-properties call site above.
+            table=exp.to_table(view_name),
         )
         if create_view_properties:
             for view_property in create_view_properties.expressions: