fix: replicated database materializations

CHANGELOG.md

@@ -15,7 +15,10 @@ The index config should be added to the model config. for instance:
          }]
   ) }}
   ```
-
+
+### Bug Fixes
+* Materializations are now compatible with `Replicated` database engine, as they will no longer use `ON CLUSTER` statements.
+
 ### Release [1.8.7], 2025-01-05
 
 ### New Features

dbt/adapters/clickhouse/impl.py

@@ -176,7 +176,7 @@ def can_exchange(self, schema: str, rel_type: str) -> bool:
     def should_on_cluster(self, materialized: str = '', engine: str = '') -> bool:
         conn = self.connections.get_if_exists()
         if conn and conn.credentials.cluster:
-            return ClickHouseRelation.get_on_cluster(conn.credentials.cluster, materialized, engine)
+            return ClickHouseRelation.get_on_cluster(conn.credentials.cluster, materialized, engine, conn.credentials.database_engine)
         return ClickHouseRelation.get_on_cluster('', materialized, engine)
 
     @available.parse_none
@@ -324,14 +324,15 @@ def list_relations_without_caching(
                 and rel_type == ClickHouseRelationType.Table
                 and db_engine in ('Atomic', 'Replicated')
             )
+            can_on_cluster = (on_cluster >= 1) and db_engine != 'Replicated'
 
             relation = self.Relation.create(
                 database='',
                 schema=schema,
                 identifier=name,
                 type=rel_type,
                 can_exchange=can_exchange,
-                can_on_cluster=(on_cluster >= 1),
+                can_on_cluster=can_on_cluster,
             )
             relations.append(relation)
 

dbt/adapters/clickhouse/relation.py

@@ -78,17 +78,17 @@ def should_on_cluster(self) -> bool:
 
     @classmethod
     def get_on_cluster(
-        cls: Type[Self], cluster: str = '', materialized: str = '', engine: str = ''
+        cls: Type[Self], cluster: str = '', materialized: str = '', engine: str = '', database_engine: str = ''
     ) -> bool:
+        if 'replicated' in database_engine.lower():
+            return False
         if cluster.strip():
             return (
                 materialized in ('view', 'dictionary')
                 or 'distributed' in materialized
                 or 'Replicated' in engine
             )
-
-        else:
-            return False
+        return False
 
     @classmethod
     def create_from(
@@ -126,7 +126,8 @@ def create_from(
         else:
             materialized = relation_config.config.get('materialized') or ''
             engine = relation_config.config.get('engine') or ''
-            can_on_cluster = cls.get_on_cluster(cluster, materialized, engine)
+            database_engine = quoting.credentials.database_engine or ''
+            can_on_cluster = cls.get_on_cluster(cluster, materialized, engine, database_engine)
 
         return cls.create(
             database='',

dbt/include/clickhouse/macros/adapters.sql

@@ -59,14 +59,13 @@
 
 {% macro clickhouse__drop_relation(relation, obj_type='table') -%}
   {% call statement('drop_relation', auto_begin=False) -%}
-    {# drop relation on cluster by default if cluster is set #}
-    drop {{ obj_type }} if exists {{ relation }} {{ on_cluster_clause(relation.without_identifier(), True)}}
+    drop {{ obj_type }} if exists {{ relation }} {{ on_cluster_clause(relation, True)}}
   {%- endcall %}
 {% endmacro %}
 
 {% macro clickhouse__rename_relation(from_relation, to_relation, obj_type='table') -%}
   {% call statement('drop_relation') %}
-    drop {{ obj_type }} if exists {{ to_relation }} {{ on_cluster_clause(to_relation.without_identifier())}}
+    drop {{ obj_type }} if exists {{ to_relation }} {{ on_cluster_clause(to_relation)}}
   {% endcall %}
   {% call statement('rename_relation') %}
     rename {{ obj_type }} {{ from_relation }} to {{ to_relation }} {{ on_cluster_clause(from_relation)}}

tests/integration/adapter/replicated_database/test_replicated_database.py

@@ -0,0 +1,31 @@
+import pytest
+
+from dbt.tests.adapter.basic.test_incremental import BaseIncremental
+from dbt.tests.adapter.basic.test_base import BaseSimpleMaterializations
+from dbt.tests.adapter.basic.files import model_incremental, schema_base_yml
+
+
+class TestReplicatedDatabaseSimpleMaterialization(BaseSimpleMaterializations):
+    """Contains tests for table, view and swappable view materialization."""
+    @pytest.fixture(scope="class")
+    def test_config(self, test_config):
+        test_config["db_engine"] = "Replicated('/clickhouse/databases/{uuid}', '{shard}', '{replica}')"
+        return test_config
+
+
+class TestReplicatedDatabaseIncremental(BaseIncremental):
+    @pytest.fixture(scope="class")
+    def test_config(self, test_config):
+        test_config["db_engine"] = "Replicated('/clickhouse/databases/{uuid}', '{shard}', '{replica}')"
+        return test_config
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        config_materialized_incremental = """
+          {{ config(order_by='(some_date, id, name)', inserts_only=True, materialized='incremental', unique_key='id') }}
+        """
+        incremental_sql = config_materialized_incremental + model_incremental
+        return {
+            "incremental.sql": incremental_sql,
+            "schema.yml": schema_base_yml,
+        }