Fix column quoting in incremental merge strategy for SQL keywords

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## Changes
 
+- Fix column quoting in incremental merge strategy for SQL keywords - ensures proper handling of columns with SQL keyword names in incremental materializations
+- [#299](https://github.com/dremio/dbt-dremio/pull/299) Enhance persist_docs macro to wrap model and column metadata (including descriptions, tags and tests) into a Markdown wiki for Dremio.
 - Updated dbt-dremio to match dbt-core v1.10 with sample mode
 - Enhanced persist_docs macro to wrap model and column metadata (including descriptions, tags and tests) into a Markdown wiki for Dremio.
 - Refactored CI

dbt/include/dremio/macros/materializations/incremental/strategies.sql

@@ -32,13 +32,13 @@ limitations under the License.*/
         {% if unique_key is sequence and unique_key is not mapping and unique_key is not string %}
             {% for key in unique_key %}
                 {% set this_key_match %}
-                    DBT_INTERNAL_SOURCE.{{ key }} = DBT_INTERNAL_DEST.{{ key }}
+                    DBT_INTERNAL_SOURCE.{{ adapter.quote(key) }} = DBT_INTERNAL_DEST.{{ adapter.quote(key) }}
                 {% endset %}
                 {% do predicates.append(this_key_match) %}
             {% endfor %}
         {% else %}
             {% set unique_key_match %}
-                DBT_INTERNAL_SOURCE.{{ unique_key }} = DBT_INTERNAL_DEST.{{ unique_key }}
+                DBT_INTERNAL_SOURCE.{{ adapter.quote(unique_key) }} = DBT_INTERNAL_DEST.{{ adapter.quote(unique_key) }}
             {% endset %}
             {% do predicates.append(unique_key_match) %}
         {% endif %}
@@ -64,17 +64,17 @@ limitations under the License.*/
         ({{ dest_cols_csv }})
     values
         ({% for column_name in dest_columns | map(attribute="name") -%}
-            DBT_INTERNAL_SOURCE.{{ column_name }}
+            DBT_INTERNAL_SOURCE.{{ adapter.quote(column_name) }}
             {%- if not loop.last %}, {%- endif %}
         {%- endfor %})
 
 {% endmacro %}
 
-{% macro dbt_dremio_get_incremental_sql(strategy, source, target, dest_columns, unique_key) %}
+{% macro dbt_dremio_get_incremental_sql(strategy, source, target, dest_columns, unique_key, incremental_predicates=none) %}
   {%- if strategy == 'append' -%}
     {{ dremio__get_incremental_append_sql(source, target, dest_columns) }}
   {%- elif strategy == 'merge' -%}
-    {{dremio__get_incremental_merge_sql(target, source, unique_key, dest_columns, incremental_predicates=none)}} 
+    {{dremio__get_incremental_merge_sql(target, source, unique_key, dest_columns, incremental_predicates)}} 
   {%- else -%}
     {% set no_sql_for_strategy_msg -%}
       No known SQL for the incremental strategy provided: {{ strategy }}

tests/functional/adapter/materialization/test_incremental.py

@@ -64,6 +64,28 @@
 {% endif %}
 """
 
+models__incremental_keyword_columns_sql = """
+{{ config(
+    materialized = 'incremental',
+    unique_key = 'id',
+    incremental_strategy='merge'
+) }}
+
+{% if not is_incremental() %}
+
+-- data for first invocation of model
+
+select 1 as id, 'test' as "language", 95 as "count"
+
+{% else %}
+
+-- data for subsequent incremental update
+
+select 2 as id, 'test2' as "language", 88 as "count"
+
+{% endif %}
+"""
+
 ResultHolder = namedtuple(
     "ResultHolder",
     [
@@ -204,3 +226,37 @@ def check_scenario_correctness(self, expected_fields, test_case_fields, project)
         check_relations_equal(
             project.adapter, [expected_fields.relation, test_case_fields.relation]
         )
+
+
+class TestIncrementalColumnQuoting:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "incremental_keyword_columns.sql": models__incremental_keyword_columns_sql,
+            "schema.yml": schema_base_yml
+        }
+
+    def test_incremental_keyword_columns(self, project):
+        results = run_dbt(["run", "--select", "incremental_keyword_columns"])
+        assert len(results) == 1
+
+        results = run_dbt(["run", "--select", "incremental_keyword_columns"])
+        assert len(results) == 1
+
+        relation = relation_from_name(project.adapter, "incremental_keyword_columns")
+        result = project.run_sql(
+            f"select count(*) as num_rows from {relation}", fetch="one"
+        )
+        assert result[0] == 2
+
+        rows = project.run_sql(
+            f'select id, "language", "count" from {relation} order by id', 
+            fetch="all"
+        )
+        assert len(rows) == 2
+        assert rows[0][0] == 1
+        assert rows[0][1] == 'test'
+        assert rows[0][2] == 95
+        assert rows[1][0] == 2
+        assert rows[1][1] == 'test2'
+        assert rows[1][2] == 88