fix: Split persist docs into column and table comments

The previous version of this macro blended the
`MODIFY COMMENT` command for the table comment
with the `COMMENT COLUMN` commands for the column comments,
inside the same `ALTER TABLE`.

This combined syntax **does not work** with `ON CLUSTER`.
(This was thought to be a clickhouse bug, but that bug is fixed
and the behavior remains.)

The revised macro now sends a separate `ALTER TABLE` for
the table comment `MODIFY COMMENT`, and a separate one for
all the `COMMENT COLUMN` commands.

I've split it up into 2 helper macros that accomplish each
task for maintainability. This also works on non-cluster
setups so there is no need to further complicate it by combining
table and column comment alterations into a single ALTER.

`clickhouse__persist_docs` now runs the relation (table comment)
helper macro if `persist_relations_docs()` is set and descriptions exist
for the relation and the column comment helper macro if
`persist_column_docs()` and columns are defined in the model.

Additionally, the formatting was cleaned up for multiple column
comments so that comments for a different column are not appearing
on the same line. Just makes it easier to read in clickhouse/dbt logs.

Example:

It produced comments like:

```sql
alter table <table>
[ON CLUSTER] modify comment
$dbt_comment_literal_block$<table comment>
$dbt_comment_literal_block$, comment column `col1_name`
$dbt_comment_literal_block$<column comment>
$dbt_comment_lieteral_block$, comment column `col2_name`

...
```

A bit messy, but works on a single node.

Although `ON CLUSTER` is correctly added, this syntax is
incorrect and ONLY produces table and column comments for
the initiator node. Now it does:

```sql
alter table <table>
[ON CLUSTER] modify comment
$dbt_comment_literal_block$ <table comment>
$dbt_comment_literal_block$

alter table <table>
[ON CLUSTER]
comment column `col1_name`
 $dbt_comment_literal_block$<column multi
linecomment>$dbt_comment_literal_block$,
comment column `col2_name`
 $dbt_comment$ <column comment>$dbt_comment$,

...
```

Author: emirkmo

dbt/include/clickhouse/macros/persist_docs.sql

@@ -10,29 +10,45 @@
   alter table {{ relation }} {{ on_cluster_clause(relation) }} modify comment '{{ comment }}'
 {% endmacro %}
 
-{% macro clickhouse__persist_docs(relation, model, for_relation, for_columns) %}
-  {%- set alter_comments = [] %}
-
+{% macro clickhouse__persist_docs(relation, model, for_relation, for_columns) -%}
+  {# Persist table comment if enabled and description provided #}
   {%- if for_relation and config.persist_relation_docs() and model.description -%}
-    {% set escaped_comment = clickhouse_escape_comment(model.description) %}
-    {% do alter_comments.append("modify comment {comment}".format(comment=escaped_comment)) %}
+    {{ _persist_table_comment(relation, model.description) }}
   {%- endif -%}
 
+  {#- Persist column comments if enabled and columns defined -#}
   {%- if for_columns and config.persist_column_docs() and model.columns -%}
-    {% set existing_columns = adapter.get_columns_in_relation(relation) | map(attribute="name") | list %}
-    {% for column_name in model.columns if (column_name in existing_columns) %}
-      {%- set comment = model.columns[column_name]['description'] -%}
-      {%- if comment %}
-        {% set escaped_comment = clickhouse_escape_comment(comment) %}
-        {% do alter_comments.append("comment column `{column_name}` {comment}".format(column_name=column_name, comment=escaped_comment)) %}
-      {%- endif %}
-    {%- endfor -%}
+    {{ _persist_column_comments(relation, model.columns) }}
   {%- endif -%}
+{%- endmacro %}
+
+{#- Helper macro: persist the table comment for a ClickHouse relation. -#}
+{% macro _persist_table_comment(relation, description) -%}
+  {#- Escape the description to be safe for ClickHouse #}
+  {%- set escaped_comment = clickhouse_escape_comment(description) -%}
+  {#- Build and run the ALTER TABLE ... MODIFY COMMENT statement -#}
+  {%- set sql = "modify comment {comment}".format(comment=escaped_comment) -%}
+  {{ run_query(one_alter_relation(relation, sql)) }}
+{%- endmacro %}
 
-  {%- if alter_comments | length > 0 -%}
-    {% do run_query(one_alter_relation(relation, alter_comments|join(', '))) %}
+{#- Helper macro: persist comments for multiple columns on a ClickHouse table.
+ relation: target table relation
+ columns: dict mapping column names to metadata (including 'description') -#}
+{% macro _persist_column_comments(relation, columns) -%}
+  {#- Gather existing columns in the relation to avoid altering non-existent ones -#}
+  {%- set existing_columns = adapter.get_columns_in_relation(relation) | map(attribute="name") | list -%}
+  {#- Collect ALTER statements for each column with a description -#}
+  {%- set alterations = [] -%}
+  {%- for column_name, info in columns.items() if info.description and column_name in existing_columns -%}
+    {%- set escaped_comment = clickhouse_escape_comment(info.description) -%}
+    {%- do alterations.append("\ncomment column `{column_name}` {comment}".format(column_name=column_name, comment=escaped_comment)) -%}
+  {%- endfor -%}
+  {#- Execute a single ALTER TABLE statement for all column comments -#}
+  {%- if alterations -%}
+    {{ run_query(one_alter_relation(relation, alterations | join(", "))) }}
   {%- endif -%}
-{% endmacro %}
+{%- endmacro %}
+
 
 {#
   By using dollar-quoting like this, users can embed anything they want into their comments

tests/integration/adapter/clickhouse/test_clickhouse_comments.py

@@ -106,99 +106,3 @@ def test_comment(self, project, model_name):
             assert column_comment.startswith("XXX")
 
         assert column_node['metadata']['comment'].startswith("YYY")
-
-        # Ensure comment is propoagated to all replicas on cluster
-        #cluster = project.test_config['cluster']
-
-#         local_relation = relation_from_name(project.adapter, model_name)
-#         if cluster:
-#             ensure_column_comments_consistent_across_replicas(project, cluster, local_relation)
-#             ensure_table_comment_on_cluster(project, cluster, local_relation)
-
-
-
-# def ensure_table_comment_on_cluster(project, cluster, local_relation):
-#     """Ensure all replicas have same comment for given relation"""
-#     # Returns 'ok' if exactly one distinct comment exists across all replicas for this table; otherwise 'mismatch'.
-#     sql = f"""
-#     SELECT
-#         if(COUNT(DISTINCT comment) = 1, 'ok', 'mismatch') AS status
-#     FROM clusterAllReplicas('{cluster}', system.tables)
-#     WHERE database = currentDatabase()
-#       AND name = '{local_relation.identifier}'
-#     """
-#     result = project.run_sql(sql, fetch="one")
-#     assert result[0] == "ok"
-
-#     sql = f"""
-#     SELECT
-#       hostname(),
-#       comment
-#     FROM clusterAllReplicas('{cluster}', system.tables)
-#     WHERE `table` = '{local_relation.identifier}'
-#     """
-
-#     result = project.run_sql(sql, fetch="all")
-
-#     for _table_row in result:
-#         assert _table_row[-1].startswith("YYY")
-
-# def ensure_column_comments_consistent_across_replicas(project, cluster, local_relation):
-#     # This query groups by column name and checks that each has exactly one distinct comment across replicas.
-#     check_sql = f"""
-#         SELECT
-#             name AS column_name,
-#             COUNT(DISTINCT comment) AS distinct_comment_count,
-#             groupArray((hostName(), comment)) AS per_replica_comments
-#         FROM clusterAllReplicas('{cluster}', system.columns)
-#         WHERE database = currentDatabase()
-#           AND table = '{local_relation.identifier}'
-#         GROUP BY column_name
-#     """
-#     rows = project.run_sql(check_sql, fetch="all")
-
-#     mismatches = [r for r in rows if r[1] != 1]
-#     if mismatches:
-#         print("Column comment mismatches:", mismatches)
-
-#     assert not mismatches
-
-#     sql = f"""
-#     SELECT
-#       name,
-#       groupArray(hostname()) as hosts,
-#       groupUniqArray(comment) as comments,
-#       length(comments) as num_comments
-#     FROM clusterAllReplicas('{cluster}', system.columns)
-#     WHERE table = '{local_relation.identifier}'
-#     GROUP BY name
-#     """
-
-#     result = project.run_sql(sql, fetch="all")
-
-#     print("WOW"*100)
-#     print("\n\n")
-#     print(result)
-#     print("WOW"*100)
-
-#     for _col in result:
-#         assert _col[-1] == 1
-
-#     assert result == []
-
-
-#     sql = f"""
-#     SELECT
-#       name,
-#       count(hostname())
-#     FROM clusterAllReplicas('{cluster}', system.columns)
-#     WHERE table = '{local_relation.identifier}'
-#     GROUP BY name
-#     """
-#     result = project.run_sql(sql, fetch="all")
-#     assert result[-1] == NUM_CLUSTER_NODES
-
-#     assert result == []
-
-
-