Repository: dbt-athena/dbt-athena (python)

Issue:
[Bug] Using Data Source name in Replace Columns statement doesn't work
### Is this a new bug in dbt-athena?

- [X] I believe this is a new bug in dbt-athena
- [X] I have searched the existing issues, and I could not find an existing issue for this bug

### Current Behavior

Hi,

I believe this is more likely an issue with Athena therefore I contacted with internal team as well, however until there is an update from there maybe we can slightly update the repo to make this work.

Problem:
When running the following statement in AWS Athena. When I include `AwsDataCatalog` in the query, it fails, but it works fine if I drop it. 

> ALTER TABLE `AwsDataCatalog`.`default`.`selman_schema_test` REPLACE COLUMNS (column1 date, column2 string);

This statement works without `AwsDataCatalog`:

> ALTER TABLE `default`.`selman_schema_test` REPLACE COLUMNS (column1 date, column2 string);

I'm using below option to test schema changes on parquet files with dbt-athena; 

_+on_schema_change: "sync_all_columns"_ 

Therefore when there is a schema change, dbt-athena plugin generates replace columns statement as the first query above and it fails. maybe we can update the code to exclude datasource when generating SQL statement, if that's possible.

Any suggestion is appreciated, thanks.


### Expected Behavior

Expected behaivour mentioned above.

### Steps To Reproduce

_No response_

### Environment

```markdown
- OS: Amazon Linux 2
- Python: 3.9.5
- dbt: 1.7.0
- dbt-athena-community: 1.7.0
```


### Additional Context

_No response_

Pull request description:
fix: Remove catalog from the DDL SQL generated by on_schema_change=sync_all_columns
# Description

This resolves #677 

When generating Hive DDL, the relation spec must not include the catalog name (i.e. "awsdatacatalog"), only the database (schema) and table name. #677 discovered that the DDL generated when `on_schema_change` is set to `sync_all_columns` generates `alter table awsdatacatalog.schema_name.table_name replace columns (…)` (with backticks around the relation spec components, but I don't know how to make Markdown include literal backticks). This DDL fails with a syntax error.

I don't know why all DDL doesn't fail, why this was only a problem when `on_schema_change` is set to `sync_all_columns`. For example, I can see multiple `ALTER TABLE … SET TBLPROPERTIES …` that are run when I run the functional tests, and none of them have the catalog in the relation spec.

My fix makes sure that the catalog is never included when rendering relation names in Hive DDL.

There were no functional tests that covered `on_schema_change`, and since I had to add one to figure out why #677 happened, I added tests for all four modes, even though only the `sync_all_columns` mode changes in this PR.

## Checklist

- [x] You followed [contributing section](https://github.com/dbt-athena/dbt-athena#contributing)
- [x] You kept your Pull Request small and focused on a single feature or bug fix.
- [x] You added unit testing when necessary
- [x] You added functional testing when necessary


Patch:
diff --git a/dbt/adapters/athena/relation.py b/dbt/adapters/athena/relation.py
index 3584fa3..7862c71 100644
--- a/dbt/adapters/athena/relation.py
+++ b/dbt/adapters/athena/relation.py
@@ -26,6 +26,13 @@ class AthenaIncludePolicy(Policy):
     identifier: bool = True
 
 
+@dataclass
+class AthenaHiveIncludePolicy(Policy):
+    database: bool = False
+    schema: bool = True
+    identifier: bool = True
+
+
 @dataclass(frozen=True, eq=False, repr=False)
 class AthenaRelation(BaseRelation):
     quote_character: str = '"'  # Presto quote character
@@ -42,10 +49,13 @@ class AthenaRelation(BaseRelation):
         - https://cwiki.apache.org/confluence/display/Hive/LanguageManual+DDL
         """
 
-        old_value = self.quote_character
+        old_quote_character = self.quote_character
         object.__setattr__(self, "quote_character", "`")  # Hive quote char
+        old_include_policy = self.include_policy
+        object.__setattr__(self, "include_policy", AthenaHiveIncludePolicy())
         rendered = self.render()
-        object.__setattr__(self, "quote_character", old_value)
+        object.__setattr__(self, "quote_character", old_quote_character)
+        object.__setattr__(self, "include_policy", old_include_policy)
         return str(rendered)
 
     def render_pure(self) -> str:


Test patch:
diff --git a/tests/functional/adapter/test_ha_iceberg.py b/tests/functional/adapter/test_ha_iceberg.py
index 94cf049..769ce58 100644
--- a/tests/functional/adapter/test_ha_iceberg.py
+++ b/tests/functional/adapter/test_ha_iceberg.py
@@ -62,7 +62,7 @@ class TestTableIcebergTableUnique:
         out, _ = capsys.readouterr()
         # in case of 2nd run we expect that the target table is renamed to __bkp
         alter_statement = (
-            f"alter table `awsdatacatalog`.`{project.test_schema}`.`{relation_name}` "
+            f"alter table `{project.test_schema}`.`{relation_name}` "
             f"rename to `{project.test_schema}`.`{relation_name}__bkp`"
         )
         delete_bkp_table_log = (
diff --git a/tests/functional/adapter/test_on_schema_change.py b/tests/functional/adapter/test_on_schema_change.py
new file mode 100644
index 0000000..bebd911
--- /dev/null
+++ b/tests/functional/adapter/test_on_schema_change.py
@@ -0,0 +1,100 @@
+import json
+
+import pytest
+
+from dbt.contracts.results import RunStatus
+from dbt.tests.util import run_dbt, run_dbt_and_capture
+
+models__table_base_model = """
+{{
+  config(
+    materialized='incremental',
+    incremental_strategy='append',
+    on_schema_change=var("on_schema_change"),
+    table_type=var("table_type"),
+  )
+}}
+
+select
+    1 as id,
+    'test 1' as name
+{%- if is_incremental() -%}
+    ,current_date as updated_at
+{%- endif -%}
+"""
+
+
+class TestOnSchemaChange:
+    @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", "ignore", "fail"]:
+                models[f"{table_type}_on_schema_change_{on_schema_change}.sql"] = models__table_base_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__sync_all_columns(self, project, table_type):
+        relation_name = f"{table_type}_on_schema_change_sync_all_columns"
+        vars = {"on_schema_change": "sync_all_columns", "table_type": table_type}
+        args = ["run", "--select", relation_name, "--vars", json.dumps(vars)]
+
+        model_run_initial = run_dbt(args)
+        assert model_run_initial.results[0].status == RunStatus.Success
+
+        model_run_incremental = run_dbt(args)
+        assert model_run_incremental.results[0].status == RunStatus.Success
+
+        new_column_names = self._column_names(project, relation_name)
+        assert new_column_names == ["id", "name", "updated_at"]
+
+    @pytest.mark.parametrize("table_type", ["hive", "iceberg"])
+    def test__append_new_columns(self, project, table_type):
+        relation_name = f"{table_type}_on_schema_change_append_new_columns"
+        vars = {"on_schema_change": "append_new_columns", "table_type": table_type}
+        args = ["run", "--select", relation_name, "--vars", json.dumps(vars)]
+
+        model_run_initial = run_dbt(args)
+        assert model_run_initial.results[0].status == RunStatus.Success
+
+        model_run_incremental = run_dbt(args)
+        assert model_run_incremental.results[0].status == RunStatus.Success
+
+        new_column_names = self._column_names(project, relation_name)
+        assert new_column_names == ["id", "name", "updated_at"]
+
+    @pytest.mark.parametrize("table_type", ["hive", "iceberg"])
+    def test__ignore(self, project, table_type):
+        relation_name = f"{table_type}_on_schema_change_ignore"
+        vars = {"on_schema_change": "ignore", "table_type": table_type}
+        args = ["run", "--select", relation_name, "--vars", json.dumps(vars)]
+
+        model_run_initial = run_dbt(args)
+        assert model_run_initial.results[0].status == RunStatus.Success
+
+        model_run_incremental = run_dbt(args)
+        assert model_run_incremental.results[0].status == RunStatus.Success
+
+        new_column_names = self._column_names(project, relation_name)
+        assert new_column_names == ["id", "name"]
+
+    @pytest.mark.parametrize("table_type", ["hive", "iceberg"])
+    def test__fail(self, project, table_type):
+        relation_name = f"{table_type}_on_schema_change_fail"
+        vars = {"on_schema_change": "fail", "table_type": table_type}
+        args = ["run", "--select", relation_name, "--vars", json.dumps(vars)]
+
+        model_run_initial = run_dbt(args)
+        assert model_run_initial.results[0].status == RunStatus.Success
+
+        model_run_incremental, log = run_dbt_and_capture(args, expect_pass=False)
+        assert model_run_incremental.results[0].status == RunStatus.Error
+        assert "The source and target schemas on this incremental model are out of sync!" in log
+
+        new_column_names = self._column_names(project, relation_name)
+        assert new_column_names == ["id", "name"]
diff --git a/tests/unit/test_relation.py b/tests/unit/test_relation.py
index 8da091a..4e8c61c 100644
--- a/tests/unit/test_relation.py
+++ b/tests/unit/test_relation.py
@@ -29,15 +29,15 @@ class TestRelation:
 
 
 class TestAthenaRelation:
-    def test_render_hive_uses_hive_style_quotation(self):
+    def test_render_hive_uses_hive_style_quotation_and_only_schema_and_table_names(self):
         relation = AthenaRelation.create(
             identifier=TABLE_NAME,
             database=DATA_CATALOG_NAME,
             schema=DATABASE_NAME,
         )
-        assert relation.render_hive() == f"`{DATA_CATALOG_NAME}`.`{DATABASE_NAME}`.`{TABLE_NAME}`"
+        assert relation.render_hive() == f"`{DATABASE_NAME}`.`{TABLE_NAME}`"
 
-    def test_render_hive_resets_quote_character_after_call(self):
+    def test_render_hive_resets_quote_character_and_include_policy_after_call(self):
         relation = AthenaRelation.create(
             identifier=TABLE_NAME,
             database=DATA_CATALOG_NAME,