escape column comments and add functional test

dbt/include/spark/macros/adapters.sql

@@ -231,7 +231,8 @@
 
 {% macro get_column_comment_sql(column_name, column_dict) -%}
   {% if column_name in column_dict and column_dict[column_name]["description"] -%}
-    {% set column_comment_clause = "comment '" ~ column_dict[column_name]["description"] ~ "'" %}
+    {% set escaped_description = column_dict[column_name]["description"] | replace("'", "\\'") %}
+    {% set column_comment_clause = "comment '" ~ escaped_description ~ "'" %}
   {%- endif -%}
   {{ adapter.quote(column_name) }} {{ column_comment_clause }}
 {% endmacro %}

tests/functional/adapter/persist_docs/fixtures.py

@@ -21,6 +21,11 @@
 select 1 as id, 'Joe' as name
 """
 
+_MODELS__VIEW_DELTA_MODEL = """
+{{ config(materialized='view') }}
+select id, count(*) as count from {{ ref('table_delta_model') }}
+"""
+
 _MODELS__TABLE_DELTA_MODEL_MISSING_COLUMN = """
 {{ config(materialized='table', file_format='delta') }}
 select 1 as id, 'Joe' as different_name

tests/functional/adapter/persist_docs/test_persist_docs.py

@@ -9,7 +9,7 @@
     _MODELS__TABLE_DELTA_MODEL_MISSING_COLUMN,
     _PROPERTIES__MODELS,
     _PROPERTIES__SEEDS,
-    _SEEDS__BASIC,
+    _SEEDS__BASIC, _MODELS__VIEW_DELTA_MODEL,
 )
 
 
@@ -76,6 +76,47 @@ def test_delta_comments(self, project):
                     assert result[2].startswith("Some stuff here and then a call to")
 
 
+@pytest.mark.skip_profile("apache_spark", "spark_session")
+class TestPersistDocsDeltaView:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "view_delta_model.sql": _MODELS__VIEW_DELTA_MODEL,
+            "schema.yml": _PROPERTIES__MODELS,
+        }
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "models": {
+                "test": {
+                    "+persist_docs": {
+                        "relation": True,
+                        "columns": True,
+                    },
+                }
+            },
+        }
+
+    def test_delta_comments(self, project):
+        run_dbt(["run"])
+
+        results = project.run_sql(
+            "describe extended {schema}.{table}".format(
+                schema=project.test_schema, table="view_delta_model"
+            ),
+            fetch="all",
+        )
+
+        for result in results:
+            if result[0] == "Comment":
+                assert result[1].startswith(f"View model description")
+            if result[0] == "id":
+                assert result[2].startswith("id Column description")
+            if result[0] == "count":
+                self.assertEqual(result[2], "")
+
+
 @pytest.mark.skip_profile("apache_spark", "spark_session")
 class TestPersistDocsMissingColumn:
     @pytest.fixture(scope="class")