update alter_relation_add_remove_columns to quote column names (#1204)

Author: colin-rogers-dbt

dbt-adapters/.changes/unreleased/Fixes-20250716-154809.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: update alter_relation_add_remove_columns to quote column names
+time: 2025-07-16T15:48:09.903045-07:00
+custom:
+  Author: colin-rogers-dbt
+  Issue: "63"

dbt-adapters/src/dbt/include/global_project/macros/adapters/columns.sql

@@ -130,11 +130,11 @@
      alter {{ relation.type }} {{ relation.render() }}
 
             {% for column in add_columns %}
-               add column {{ column.name }} {{ column.data_type }}{{ ',' if not loop.last }}
+               add column {{ column.quoted }} {{ column.data_type }}{{ ',' if not loop.last }}
             {% endfor %}{{ ',' if add_columns and remove_columns }}
 
             {% for column in remove_columns %}
-                drop column {{ column.name }}{{ ',' if not loop.last }}
+                drop column {{ column.quoted }}{{ ',' if not loop.last }}
             {% endfor %}
 
   {%- endset -%}

dbt-athena/src/dbt/adapters/athena/column.py

@@ -17,6 +17,18 @@ class AthenaColumn(Column):
         "TEXT": "VARCHAR",
     }
 
+    @property
+    def quoted(self) -> str:
+        """Override quoted property to ensure double quotes are used for Athena column identifiers"""
+        return f'"{self.column}"'
+
+    @property
+    def quoted_hive(self) -> str:
+        """
+        For some reason, sometimes we need to use double quotes, sometimes we need to use backticks.
+        """
+        return f"`{self.column}`"
+
     def is_iceberg(self) -> bool:
         return self.table_type == TableType.ICEBERG
 

dbt-athena/src/dbt/include/athena/macros/materializations/models/incremental/column_helpers.sql

@@ -7,7 +7,7 @@
       alter {{ relation.type }} {{ relation.render_hive() }}
           add columns (
             {%- for column in add_columns -%}
-                {{ column.name }} {{ ddl_data_type(column.data_type, table_type) }}{{ ', ' if not loop.last }}
+                {{ column.quoted_hive }} {{ ddl_data_type(column.data_type, table_type) }}{{ ', ' if not loop.last }}
             {%- endfor -%}
           )
   {%- endset -%}
@@ -24,12 +24,13 @@
 
   {%- for column in remove_columns -%}
     {% set sql -%}
-      alter {{ relation.type }} {{ relation.render_hive() }} drop column {{ column.name }}
+      alter {{ relation.type }} {{ relation.render_hive() }} drop column {{ column.quoted_hive }}
     {% endset %}
     {% do run_query(sql) %}
   {%- endfor -%}
 {% endmacro %}
 
+
 {% macro alter_relation_replace_columns(relation, replace_columns = none, table_type = 'hive') -%}
   {% if replace_columns is none %}
     {% set replace_columns = [] %}
@@ -39,7 +40,7 @@
       alter {{ relation.type }} {{ relation.render_hive() }}
           replace columns (
             {%- for column in replace_columns -%}
-                {{ column.name }} {{ ddl_data_type(column.data_type, table_type) }}{{ ', ' if not loop.last }}
+                {{ column.quoted_hive }} {{ ddl_data_type(column.data_type, table_type) }}{{ ', ' if not loop.last }}
             {%- endfor -%}
           )
   {%- endset -%}

dbt-athena/tests/functional/adapter/test_on_schema_change.py

@@ -100,3 +100,88 @@ def test__fail(self, project, table_type):
 
         new_column_names = self._column_names(project, relation_name)
         assert new_column_names == ["id", "name"]
+
+
+# Test fixtures for special character column names
+models__table_special_chars_model = """
+{{
+  config(
+    materialized='incremental',
+    incremental_strategy='append',
+    on_schema_change=var("on_schema_change"),
+    table_type=var("table_type"),
+  )
+}}
+
+select
+    1 as id,
+    'keyword' as "select"
+{%- if is_incremental() -%}
+    ,'new data' as "field-with-dash"
+{%- endif -%}
+"""
+
+
+class TestOnSchemaChangeSpecialChars:
+    """Test schema changes with special character column names for Athena
+
+    Tests column quoting with SQL keywords and dashes in column names.
+    """
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        models = {}
+        for table_type in ["hive", "iceberg"]:
+            for on_schema_change in ["sync_all_columns", "append_new_columns"]:
+                models[f"{table_type}_special_chars_{on_schema_change}.sql"] = (
+                    models__table_special_chars_model
+                )
+        return models
+
+    def _column_names(self, project, relation_name):
+        result = project.run_sql(f"show columns from {relation_name}", fetch="all")
+        column_names = [row[0].strip() for row in result]
+        return column_names
+
+    @pytest.mark.parametrize("table_type", ["hive", "iceberg"])
+    def test__append_new_columns_special_chars(self, project, table_type):
+        """Test that columns with special characters are properly quoted when added"""
+        relation_name = f"{table_type}_special_chars_append_new_columns"
+        vars = {"on_schema_change": "append_new_columns", "table_type": table_type}
+        args = ["run", "--select", relation_name, "--vars", json.dumps(vars)]
+
+        # First run - creates initial table with quoted column names
+        run_dbt(args)
+
+        # Second run - should append new columns with special characters
+        run_dbt(args)
+
+        # Verify all columns including ones with special characters exist
+        new_column_names = self._column_names(project, relation_name)
+        expected_columns = [
+            "id",
+            "select",
+            "field-with-dash",
+        ]
+        assert sorted(new_column_names) == sorted(expected_columns)
+
+    def test__sync_all_columns_special_chars(self, project):
+        """Test that columns with special characters are properly quoted when synced (add/remove)"""
+        relation_name = "hive_special_chars_sync_all_columns"
+        vars = {"on_schema_change": "sync_all_columns", "table_type": "hive"}
+        args = ["run", "--select", relation_name, "--vars", json.dumps(vars)]
+
+        # First run - creates initial table
+        run_dbt(args)
+
+        # Second run - should sync columns (replace all columns)
+        run_dbt(args)
+
+        # Verify columns were properly synced
+        new_column_names = self._column_names(project, relation_name)
+        expected_columns = [
+            "id",
+            "select",
+            "field-with-dash",
+        ]
+        assert sorted(new_column_names) == sorted(expected_columns)