Repository: ClickHouse/dbt-clickhouse (language: python)

Issue:
Adding PROJECTION to models
**Is your feature request related to a problem? Please describe.**
It would be awesome if we could add [projections](https://clickhouse.com/docs/en/sql-reference/statements/alter/projection) to models, this is very helpful when using a semantic layer on top of the data marts, to facilitate and speed up querying for certain filters.

**Describe the solution you'd like**
IMHO, the ideal solution would be to configure the projection as a `config` parameter (not sure though if dbt supports this):

```sql
-- models/marts/touchpoints;
{{
    config(
        materialized='materialized_view',
        engine='MergeTree()',
        order_by='workspace_id, toStartOfHour(updated_at), touchpoint_id',
        projections=[
            { "name": "some_projection", "select": ["touchpoint_id", "some_other_col"], "order_by": ["touchpoint_id"] }
        ]
    )
}}
select * from {{ source('other', 'touchpoints') }}
```

**Describe alternatives you've considered**
The alternative that we use currently is using a migration script to put the projection on various models, after they have been built, which is not really nice, since the code creating a projection and the model are separate from each other and we need to take care that the database migrations are always up to date with the dbt models.



Discussion / hints:
FWIW. We are adding projections using [post_hooks](https://docs.getdbt.com/reference/resource-configs/pre-hook-post-hook).

It works alright, But we get timeouts for materializing slow projections. `ClickHouse exception:  The read operation timed out`
<img width="544" alt="image" src="https://github.com/user-attachments/assets/a785d694-bb88-4f6e-8cfb-d5bdc62fd10a">


Fix patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index aae6da39..3d9b686d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,6 +1,19 @@
-### Release XXXXX
+### Release [1.8.2], xxxx-xx-xx
 #### Improvements
-* Added a new *experimental* incremental strategy - `insert_overwrite` was added which replaces existing data in a target table partition by partition, ensuring only the specified partitions are overwritten with new data, which helps maintain performance and data consistency. This feature was not tested with distribution models and might not work with such materializations.  
+* You can now add projections to tables. The projection config should be added to the model config, for instance:
+```python
+{{ config(
+       materialized='%s',
+       projections=[
+           {
+               'name': 'your_projection_name',
+               'query': 'your_projection_query'
+           }
+       ]
+) }}
+```
+Please be aware that this feature was added to distributed materialization as well, but might not fully function.
+* Added a new *experimental* incremental strategy - `insert_overwrite` was added which replaces existing data in a target table partition by partition, ensuring only the specified partitions are overwritten with new data, which helps maintain performance and data consistency. This feature was not tested with distribution models and might not work with such materializations. 
 
 
 ### Release [1.8.1], 2024-07-11
diff --git a/dbt/include/clickhouse/macros/materializations/distributed_table.sql b/dbt/include/clickhouse/macros/materializations/distributed_table.sql
index 43b2d7ff..e84e8396 100644
--- a/dbt/include/clickhouse/macros/materializations/distributed_table.sql
+++ b/dbt/include/clickhouse/macros/materializations/distributed_table.sql
@@ -112,6 +112,15 @@
   create table {{ relation.include(database=False) }}
   {{ on_cluster_clause(relation) }} (
       {{col_list | join(', ')}}
+
+    {% if config.get('projections') %}
+      {% set projections = config.get('projections') %}
+      {% for projection in projections %}
+        , PROJECTION {{ projection.get("name") }} (
+            {{ projection.get("query") }}
+        )
+      {% endfor %}
+  {% endif %}
   )
   
   {{ engine_clause() }}
diff --git a/dbt/include/clickhouse/macros/materializations/table.sql b/dbt/include/clickhouse/macros/materializations/table.sql
index b3cad47b..cefb4a0f 100644
--- a/dbt/include/clickhouse/macros/materializations/table.sql
+++ b/dbt/include/clickhouse/macros/materializations/table.sql
@@ -147,10 +147,28 @@
         {% call statement('create_table_empty') %}
             {{ create_table }}
         {% endcall %}
+         {% if config.get('projections')%}
+             {% call statement('add_projections') %}
+                {{ projection_statement(relation) }}
+            {%endcall  %}
+         {% endif %}
+
+
         {{ clickhouse__insert_into(relation, sql, has_contract) }}
     {%- endif %}
 {%- endmacro %}
 
+{% macro projection_statement(relation) %}
+    {%- set projections = config.get('projections', default=[]) -%}
+
+    {%- for projection in projections %}
+        ALTER TABLE {{ relation }} ADD PROJECTION {{ projection.get('name') }}
+        (
+            {{ projection.get('query') }}
+        )
+    {%- endfor %}
+{%- endmacro %}
+
 {% macro create_table_or_empty(temporary, relation, sql, has_contract) -%}
     {%- set sql_header = config.get('sql_header', none) -%}
 


Test patch:
diff --git a/tests/integration/adapter/projections/test_projections.py b/tests/integration/adapter/projections/test_projections.py
new file mode 100644
index 00000000..45615fa1
--- /dev/null
+++ b/tests/integration/adapter/projections/test_projections.py
@@ -0,0 +1,124 @@
+import os
+import time
+import uuid
+
+import pytest
+from dbt.tests.util import relation_from_name, run_dbt
+
+PEOPLE_SEED_CSV = """
+id,name,age,department
+1231,Dade,33,engineering
+6666,Ksenia,48,engineering
+8888,Kate,50,engineering
+1232,Eugene,40,malware
+9999,Paul,25,sales
+""".lstrip()
+
+PEOPLE_MODEL_WITH_PROJECTION = """
+{{ config(
+       materialized='%s',
+       projections=[
+           {
+               'name': 'projection_avg_age',
+               'query': 'SELECT department, avg(age) AS avg_age GROUP BY department'
+           }
+       ]
+) }}
+
+select
+    id,
+    name,
+    age,
+    department
+from {{ source('raw', 'people') }}
+"""
+
+SEED_SCHEMA_YML = """
+version: 2
+
+sources:
+  - name: raw
+    schema: "{{ target.schema }}"
+    tables:
+      - name: people
+"""
+
+
+class TestProjections:
+    @pytest.fixture(scope="class")
+    def seeds(self):
+        return {
+            "people.csv": PEOPLE_SEED_CSV,
+            "schema.yml": SEED_SCHEMA_YML,
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "people_with_projection.sql": PEOPLE_MODEL_WITH_PROJECTION % "table",
+            "distributed_people_with_projection.sql": PEOPLE_MODEL_WITH_PROJECTION
+            % "distributed_table",
+        }
+
+    def test_create_and_verify_projection(self, project):
+        run_dbt(["seed"])
+        run_dbt()
+
+        relation = relation_from_name(project.adapter, "people_with_projection")
+        unique_query_identifier = str(uuid.uuid4())
+        query = f""" -- {unique_query_identifier}
+        SELECT department, avg(age) AS avg_age FROM {project.test_schema}.{relation.name}
+        GROUP BY department ORDER BY department"""
+
+        # Check that the projection works as expected
+        result = project.run_sql(query, fetch="all")
+        assert len(result) == 3  # We expect 3 departments in the result
+        assert result == [('engineering', 43.666666666666664), ('malware', 40.0), ('sales', 25.0)]
+
+        # waiting for system.log table to be created
+        time.sleep(10)
+
+        # check that the latest query used the projection
+        result = project.run_sql(
+            f"SELECT query, projections FROM clusterAllReplicas(default, 'system.query_log') "
+            f"WHERE query like '%{unique_query_identifier}%' "
+            f"and query not like '%system.query_log%' and read_rows > 0 ORDER BY query_start_time DESC",
+            fetch="all",
+        )
+        assert len(result) > 0
+        assert query in result[0][0]
+
+        assert result[0][1] == [f'{project.test_schema}.{relation.name}.projection_avg_age']
+
+    @pytest.mark.xfail
+    @pytest.mark.skipif(
+        os.environ.get('DBT_CH_TEST_CLUSTER', '').strip() == '', reason='Not on a cluster'
+    )
+    def test_create_and_verify_distributed_projection(self, project):
+        run_dbt(["seed"])
+        run_dbt()
+        relation = relation_from_name(project.adapter, "distributed_people_with_projection")
+        unique_query_identifier = str(uuid.uuid4())
+        query = f"""-- {unique_query_identifier}
+                 SELECT department, avg(age) AS avg_age FROM {project.test_schema}.{relation.name} GROUP BY 
+                 department ORDER BY department"""
+
+        # Check that the projection works as expected
+        result = project.run_sql(query, fetch="all")
+        assert len(result) == 3  # We expect 3 departments in the result
+        assert result == [('engineering', 43.666666666666664), ('malware', 40.0), ('sales', 25.0)]
+
+        # waiting for system.log table to be created
+        time.sleep(10)
+
+        # check that the latest query used the projection
+        result = project.run_sql(
+            f"SELECT query, projections FROM clusterAllReplicas(default, 'system.query_log') "
+            f"WHERE query like '%{unique_query_identifier}%' "
+            f"and query not like '%system.query_log%' and read_rows > 0 ORDER BY query_start_time DESC",
+            fetch="all",
+        )
+        assert len(result) > 0
+        assert query in result[0][0]
+
+        assert result[0][1] == [f'{project.test_schema}.{relation.name}_local.projection_avg_age']