Fix quoting on columns for seeds/incremental models

Added tests

core/dbt/include/global_project/macros/materializations/common/merge.sql

@@ -48,7 +48,8 @@
 
 
 {% macro common_get_delete_insert_merge_sql(target, source, unique_key, dest_columns) -%}
-    {%- set dest_cols_csv = dest_columns | map(attribute="name") | join(', ') -%}
+
+    {%- set dest_cols_csv = get_quoted_csv(dest_columns | map(attribute="name")) -%}
 
     {% if unique_key is not none %}
     delete from {{ target }}

core/dbt/include/global_project/macros/materializations/seed/seed.sql

@@ -19,7 +19,7 @@
         {%- for col_name in agate_table.column_names -%}
             {%- set inferred_type = adapter.convert_type(agate_table, loop.index0) -%}
             {%- set type = column_override.get(col_name, inferred_type) -%}
-            {{ col_name | string }} {{ type }} {%- if not loop.last -%}, {%- endif -%}
+            {{ adapter.quote(col_name | string) }} {{ type }} {%- if not loop.last -%}, {%- endif -%}
         {%- endfor -%}
     )
   {% endset %}
@@ -46,8 +46,19 @@
 {% endmacro %}
 
 
+{% macro get_quoted_csv(column_names) %}
+    {% set quoted = [] %}
+    {% for col in column_names -%}
+        {%- do quoted.append(adapter.quote(col)) -%}
+    {%- endfor %}
+
+    {%- set dest_cols_csv = quoted | join(', ') -%}
+    {{ return(dest_cols_csv) }}
+{% endmacro %}
+
+
 {% macro basic_load_csv_rows(model, batch_size, agate_table) %}
-    {% set cols_sql = ", ".join(agate_table.column_names) %}
+    {% set cols_sql = get_quoted_csv(agate_table.column_names) %}
     {% set bindings = [] %}
 
     {% set statements = [] %}

plugins/snowflake/dbt/include/snowflake/macros/materializations/merge.sql

@@ -6,7 +6,7 @@
        is provided, then this macro will do a proper merge instead.
     #}
 
-    {%- set dest_cols_csv = dest_columns | map(attribute="name") | join(', ') -%}
+    {%- set dest_cols_csv = get_quoted_csv(dest_columns | map(attribute='name')) -%}
 
     {%- if unique_key is none -%}
 

test/integration/052_column_quoting/data/seed.csv

@@ -0,0 +1,4 @@
+"col_A","col_B"
+1,2
+3,4
+5,6

test/integration/052_column_quoting/models/model.sql

@@ -0,0 +1,16 @@
+{% set col_a = '"col_A"' %}
+{% set col_b = '"col_B"' %}
+{% if adapter.type() == 'bigquery' %}
+    {% set col_a = '`col_A`' %}
+    {% set col_b = '`col_B`' %}
+{% endif %}
+
+{{config(
+    materialized = 'incremental',
+    unique_key = col_a,
+    incremental_strategy = "delete+insert"
+    )}}
+
+select
+{{ col_a }}, {{ col_b }}
+from {{ref('seed')}}

test/integration/052_column_quoting/test_column_quotes.py

@@ -0,0 +1,37 @@
+from test.integration.base import DBTIntegrationTest,  use_profile
+import os
+
+
+class TestColumnQuoting(DBTIntegrationTest):
+    @property
+    def schema(self):
+        return 'dbt_column_quoting_052'
+
+    @staticmethod
+    def dir(value):
+        return os.path.normpath(value)
+
+    @property
+    def models(self):
+        return self.dir('models')
+
+    def _run_columnn_quotes(self):
+        self.run_dbt(['seed'])
+        self.run_dbt()
+        self.run_dbt()
+
+    @use_profile('postgres')
+    def test_postgres_column_quotes(self):
+        self._run_columnn_quotes()
+
+    @use_profile('redshift')
+    def test_redshift_column_quotes(self):
+        self._run_columnn_quotes()
+
+    @use_profile('snowflake')
+    def test_snowflake_column_quotes(self):
+        self._run_columnn_quotes()
+
+    @use_profile('bigquery')
+    def test_bigquery_column_quotes(self):
+        self._run_columnn_quotes()