fix: use correct schema for MV target tables (#244)

* fix: use correct schema when updating MVs

The existing implementation passes just the name for `target_table`,
which ultimately means that the target schema is not included when the
final SQL is generated. By passing the entire relation object, the
correct target schema will be present in the final SQL.

* update MV tests

Provide a custom schema to make sure that the full target table
name (schema + relation name) is included in the CREATE MATERIALIZED
VIEW statement

dbt/include/clickhouse/macros/materializations/materialized_view.sql

@@ -41,7 +41,7 @@
       {{ clickhouse__get_create_materialized_view_as_sql(target_relation, sql) }}
     {%- endcall %}
   {% elif existing_relation.can_exchange %}
-    {{ log('Replacing existing materialized view' + target_relation.name) }}
+    {{ log('Replacing existing materialized view ' + target_relation.name) }}
     {% call statement('drop existing materialized view') %}
       drop view if exists {{ mv_relation }} {{ cluster_clause }}
     {% endcall %}
@@ -50,10 +50,10 @@
     {%- endcall %}
     {% do exchange_tables_atomic(backup_relation, existing_relation) %}
     {% call statement('create new materialized view') %}
-      {{ clickhouse__create_mv_sql(mv_relation, existing_relation.name, cluster_clause, sql) }}
+      {{ clickhouse__create_mv_sql(mv_relation, existing_relation, cluster_clause, sql) }}
     {% endcall %}
   {% else %}
-    {{ log('Replacing existing materialized view' + target_relation.name) }}
+    {{ log('Replacing existing materialized view ' + target_relation.name) }}
     {{ clickhouse__replace_mv(target_relation, existing_relation, intermediate_relation, backup_relation, sql) }}
   {% endif %}
 

tests/integration/adapter/materialized_view/test_materialized_view.py

@@ -25,6 +25,7 @@
        materialized='materialized_view',
        engine='MergeTree()',
        order_by='(id)',
+       schema='custom_schema',
 ) }}
 
 {% if var('run_type', '') == '' %}
@@ -92,6 +93,7 @@ def test_create(self, project):
         2. create a model as a materialized view, selecting from the table created in (1)
         3. insert data into the base table and make sure it's there in the target table created in (2)
         """
+        schema = quote_identifier(project.test_schema + "_custom_schema")
         results = run_dbt(["seed"])
         assert len(results) == 1
         columns = project.run_sql("DESCRIBE TABLE people", fetch="all")
@@ -101,10 +103,10 @@ def test_create(self, project):
         results = run_dbt()
         assert len(results) == 1
 
-        columns = project.run_sql("DESCRIBE TABLE hackers", fetch="all")
+        columns = project.run_sql(f"DESCRIBE TABLE {schema}.hackers", fetch="all")
         assert columns[0][1] == "Int32"
 
-        columns = project.run_sql("DESCRIBE hackers_mv", fetch="all")
+        columns = project.run_sql(f"DESCRIBE {schema}.hackers_mv", fetch="all")
         assert columns[0][1] == "Int32"
 
         check_relation_types(
@@ -123,7 +125,7 @@ def test_create(self, project):
         """
         )
 
-        result = project.run_sql("select count(*) from hackers", fetch="all")
+        result = project.run_sql(f"select count(*) from {schema}.hackers", fetch="all")
         assert result[0][0] == 4
 
 
@@ -145,6 +147,7 @@ def models(self):
         }
 
     def test_update(self, project):
+        schema = quote_identifier(project.test_schema + "_custom_schema")
         # create our initial materialized view
         run_dbt(["seed"])
         run_dbt()
@@ -162,6 +165,6 @@ def test_update(self, project):
 
         # assert that we now have both of Dade's aliases in our hackers table
         result = project.run_sql(
-            "select distinct hacker_alias from hackers where name = 'Dade'", fetch="all"
+            f"select distinct hacker_alias from {schema}.hackers where name = 'Dade'", fetch="all"
         )
         assert len(result) == 2