fix: implement quote seed column

dbt/adapters/athena/impl.py

@@ -124,6 +124,8 @@ class AthenaAdapter(SQLAdapter):
     AdapterSpecificConfigs = AthenaConfig
     Column = AthenaColumn
 
+    quote_character: str = '"'  # Presto quote character
+
     # There is no such concept as constraints in Athena
     CONSTRAINT_SUPPORT = {
         ConstraintType.check: ConstraintSupport.NOT_SUPPORTED,
@@ -397,9 +399,22 @@ def clean_up_table(self, relation: AthenaRelation) -> None:
         if table_location := self.get_glue_table_location(relation):
             self.delete_from_s3(table_location)
 
+    def quote(self, identifier: str) -> str:
+        return f"{self.quote_character}{identifier}{self.quote_character}"
+
     @available
-    def quote_seed_column(self, column: str, quote_config: Optional[bool]) -> str:
-        return str(super().quote_seed_column(column, False))
+    def quote_seed_column(
+        self, column: str, quote_config: Optional[bool], quote_character: Optional[str] = None
+    ) -> str:
+        if quote_character:
+            old_value = self.quote_character
+            object.__setattr__(self, "quote_character", quote_character)
+            quoted_column = str(super().quote_seed_column(column, quote_config))
+            object.__setattr__(self, "quote_character", old_value)
+        else:
+            quoted_column = str(super().quote_seed_column(column, quote_config))
+
+        return quoted_column
 
     @available
     def upload_seed_to_s3(

dbt/include/athena/macros/materializations/seeds/helpers.sql

@@ -64,7 +64,7 @@
           {%- set type = column_override.get(col_name, inferred_type) -%}
           {%- set type = type if type != "string" else "varchar" -%}
           {%- set column_name = (col_name | string) -%}
-          {{ adapter.quote_seed_column(column_name, quote_seed_column) }} {{ ddl_data_type(type) }} {%- if not loop.last -%}, {% endif -%}
+          {{ adapter.quote_seed_column(column_name, quote_seed_column, "`") }} {{ ddl_data_type(type) }} {%- if not loop.last -%}, {% endif -%}
         {%- endfor -%}
     )
     location '{{ location }}'
@@ -131,7 +131,7 @@
     create external table {{ tmp_relation.render_hive() }} (
         {%- for col_name in agate_table.column_names -%}
             {%- set column_name = (col_name | string) -%}
-            {{ adapter.quote_seed_column(column_name, quote_seed_column) }} string {%- if not loop.last -%}, {% endif -%}
+            {{ adapter.quote_seed_column(column_name, quote_seed_column, "`") }} string {%- if not loop.last -%}, {% endif -%}
         {%- endfor -%}
     )
     row format serde 'org.apache.hadoop.hive.serde2.OpenCSVSerde'

tests/functional/adapter/test_quote_seed_column.py

@@ -0,0 +1,39 @@
+import pytest
+
+from dbt.tests.adapter.basic.files import (
+    base_materialized_var_sql,
+    base_table_sql,
+    base_view_sql,
+    schema_base_yml,
+    seeds_base_csv,
+)
+from dbt.tests.adapter.basic.test_base import BaseSimpleMaterializations
+
+seed_base_csv_underscore_column = seeds_base_csv.replace("name", "_name")
+
+quote_columns_seed_schema = """
+
+seeds:
+- name: base
+  config:
+    quote_columns: true
+
+"""
+
+
+class TestSimpleMaterializationsHive(BaseSimpleMaterializations):
+    @pytest.fixture(scope="class")
+    def models(self):
+        schema = schema_base_yml + quote_columns_seed_schema
+        return {
+            "view_model.sql": base_view_sql,
+            "table_model.sql": base_table_sql,
+            "swappable.sql": base_materialized_var_sql,
+            "schema.yml": schema,
+        }
+
+    @pytest.fixture(scope="class")
+    def seeds(self):
+        return {
+            "base.csv": seed_base_csv_underscore_column,
+        }

tests/unit/test_adapter.py

@@ -600,10 +600,17 @@ def test_clean_up_table_delete_table(self, dbt_debug_caplog, mock_aws_service):
         objs = s3.list_objects_v2(Bucket=BUCKET)
         assert objs["KeyCount"] == 0
 
-    @patch("dbt.adapters.athena.impl.SQLAdapter.quote_seed_column")
-    def test_quote_seed_column(self, parent_quote_seed_column):
-        self.adapter.quote_seed_column("col", None)
-        parent_quote_seed_column.assert_called_once_with("col", False)
+    @pytest.mark.parametrize(
+        "column,quote_config,quote_character,expected",
+        [
+            pytest.param("col", False, None, "col"),
+            pytest.param("col", True, None, '"col"'),
+            pytest.param("col", False, "`", "col"),
+            pytest.param("col", True, "`", "`col`"),
+        ],
+    )
+    def test_quote_seed_column(self, column, quote_config, quote_character, expected):
+        assert self.adapter.quote_seed_column(column, quote_config, quote_character) == expected
 
     @mock_glue
     @mock_athena